| 31 | |
| 32 | |
| 33 | class CursorTest(base.PyMySQLTestCase): |
| 34 | def setUp(self): |
| 35 | super().setUp() |
| 36 | |
| 37 | conn = self.connect() |
| 38 | self.safe_create_table( |
| 39 | conn, |
| 40 | "test", |
| 41 | "create table test (data varchar(10))", |
| 42 | ) |
| 43 | cursor = conn.cursor() |
| 44 | cursor.execute( |
| 45 | "insert into test (data) values ('row1'), ('row2'), ('row3'), ('row4'), ('row5')" |
| 46 | ) |
| 47 | conn.commit() |
| 48 | cursor.close() |
| 49 | self.test_connection = pymysql.connect(**self.databases[0]) |
| 50 | self.addCleanup(self.test_connection.close) |
| 51 | |
| 52 | def test_cursor_is_iterator(self): |
| 53 | """Test that the cursor is an iterator""" |
| 54 | conn = self.test_connection |
| 55 | cursor = conn.cursor() |
| 56 | cursor.execute("select * from test") |
| 57 | self.assertEqual(cursor.__iter__(), cursor) |
| 58 | self.assertEqual(cursor.__next__(), ("row1",)) |
| 59 | |
| 60 | def test_cleanup_rows_unbuffered(self): |
| 61 | conn = self.test_connection |
| 62 | cursor = conn.cursor(pymysql.cursors.SSCursor) |
| 63 | |
| 64 | cursor.execute("select * from test as t1, test as t2") |
| 65 | for counter, row in enumerate(cursor): |
| 66 | if counter > 10: |
| 67 | break |
| 68 | |
| 69 | del cursor |
| 70 | |
| 71 | c2 = conn.cursor() |
| 72 | |
| 73 | c2.execute("select 1") |
| 74 | self.assertEqual(c2.fetchone(), (1,)) |
| 75 | self.assertIsNone(c2.fetchone()) |
| 76 | |
| 77 | def test_cleanup_rows_buffered(self): |
| 78 | conn = self.test_connection |
| 79 | cursor = conn.cursor(pymysql.cursors.Cursor) |
| 80 | |
| 81 | cursor.execute("select * from test as t1, test as t2") |
| 82 | for counter, row in enumerate(cursor): |
| 83 | if counter > 10: |
| 84 | break |
| 85 | |
| 86 | del cursor |
| 87 | |
| 88 | c2 = conn.cursor() |
| 89 | c2.execute("select 1") |
| 90 |
nothing calls this directly
no outgoing calls
no test coverage detected