| 1 | import unittest2 |
|---|
| 2 | |
|---|
| 3 | from pymysql.tests import base |
|---|
| 4 | from pymysql import util |
|---|
| 5 | |
|---|
| 6 | |
|---|
| 7 | class TestNextset(base.PyMySQLTestCase): |
|---|
| 8 | |
|---|
| 9 | def setUp(self): |
|---|
| 10 | super(TestNextset, self).setUp() |
|---|
| 11 | self.con = self.connections[0] |
|---|
| 12 | |
|---|
| 13 | def test_nextset(self): |
|---|
| 14 | cur = self.con.cursor() |
|---|
| 15 | cur.execute("SELECT 1; SELECT 2;") |
|---|
| 16 | self.assertEqual([(1,)], list(cur)) |
|---|
| 17 | |
|---|
| 18 | r = cur.nextset() |
|---|
| 19 | self.assertTrue(r) |
|---|
| 20 | |
|---|
| 21 | self.assertEqual([(2,)], list(cur)) |
|---|
| 22 | self.assertIsNone(cur.nextset()) |
|---|
| 23 | |
|---|
| 24 | def test_skip_nextset(self): |
|---|
| 25 | cur = self.con.cursor() |
|---|
| 26 | cur.execute("SELECT 1; SELECT 2;") |
|---|
| 27 | self.assertEqual([(1,)], list(cur)) |
|---|
| 28 | |
|---|
| 29 | cur.execute("SELECT 42") |
|---|
| 30 | self.assertEqual([(42,)], list(cur)) |
|---|
| 31 | |
|---|
| 32 | def test_ok_and_next(self): |
|---|
| 33 | cur = self.con.cursor() |
|---|
| 34 | cur.execute("SELECT 1; commit; SELECT 2;") |
|---|
| 35 | self.assertEqual([(1,)], list(cur)) |
|---|
| 36 | self.assertTrue(cur.nextset()) |
|---|
| 37 | self.assertTrue(cur.nextset()) |
|---|
| 38 | self.assertEqual([(2,)], list(cur)) |
|---|
| 39 | self.assertFalse(bool(cur.nextset())) |
|---|
| 40 | |
|---|
| 41 | @unittest2.expectedFailure |
|---|
| 42 | def test_multi_cursor(self): |
|---|
| 43 | cur1 = self.con.cursor() |
|---|
| 44 | cur2 = self.con.cursor() |
|---|
| 45 | |
|---|
| 46 | cur1.execute("SELECT 1; SELECT 2;") |
|---|
| 47 | cur2.execute("SELECT 42") |
|---|
| 48 | |
|---|
| 49 | self.assertEqual([(1,)], list(cur1)) |
|---|
| 50 | self.assertEqual([(42,)], list(cur2)) |
|---|
| 51 | |
|---|
| 52 | r = cur1.nextset() |
|---|
| 53 | self.assertTrue(r) |
|---|
| 54 | |
|---|
| 55 | self.assertEqual([(2,)], list(cur1)) |
|---|
| 56 | self.assertIsNone(cur1.nextset()) |
|---|
| 57 | |
|---|
| 58 | def test_multi_statement_warnings(self): |
|---|
| 59 | cursor = self.con.cursor() |
|---|
| 60 | |
|---|
| 61 | try: |
|---|
| 62 | cursor.execute('DROP TABLE IF EXISTS a; ' |
|---|
| 63 | 'DROP TABLE IF EXISTS b;') |
|---|
| 64 | except TypeError: |
|---|
| 65 | self.fail() |
|---|
| 66 | |
|---|
| 67 | #TODO: How about SSCursor and nextset? |
|---|
| 68 | # It's very hard to implement correctly... |
|---|
