Book a Demo!
CoCalc Logo Icon
StoreFeaturesDocsShareSupportNewsAboutPoliciesSign UpSign In
singlestore-labs
GitHub Repository: singlestore-labs/singlestoredb-python
 Path: blob/main/singlestoredb/mysql/tests/test_nextset.py
469 views
1
# type: ignore

2
import pytest

3


4
import singlestoredb.mysql as sv

5
from singlestoredb.mysql.constants import CLIENT

6
from singlestoredb.mysql.tests import base

7


8


9
class TestNextset(base.PyMySQLTestCase):

10


11
    def test_nextset(self):

12
        con = self.connect(

13
            init_command='SELECT "bar"; SELECT "baz"',

14
            client_flag=CLIENT.MULTI_STATEMENTS,

15
        )

16
        cur = con.cursor()

17
        cur.execute('SELECT 1; SELECT 2;')

18
        self.assertEqual([(1,)], list(cur))

19


20
        r = cur.nextset()

21
        self.assertTrue(r)

22


23
        self.assertEqual([(2,)], list(cur))

24
        self.assertIsNone(cur.nextset())

25


26
    def test_skip_nextset(self):

27
        cur = self.connect(client_flag=CLIENT.MULTI_STATEMENTS).cursor()

28
        cur.execute('SELECT 1; SELECT 2;')

29
        self.assertEqual([(1,)], list(cur))

30


31
        cur.execute('SELECT 42')

32
        self.assertEqual([(42,)], list(cur))

33


34
    def test_nextset_error(self):

35
        con = self.connect(client_flag=CLIENT.MULTI_STATEMENTS)

36
        cur = con.cursor()

37


38
        if type(cur).__name__.startswith('SS'):

39
            self.skipTest('nextset in unbuffered cursor closes connection')

40


41
        for i in range(3):

42
            cur.execute('SELECT %s; xyzzy;', (i,))

43
            self.assertEqual([(i,)], list(cur))

44
            with self.assertRaises(sv.ProgrammingError):

45
                cur.nextset()

46
            self.assertEqual((), cur.fetchall())

47


48
    def test_ok_and_next(self):

49
        cur = self.connect(client_flag=CLIENT.MULTI_STATEMENTS).cursor()

50
        cur.execute('SELECT 1; commit; SELECT 2;')

51
        self.assertEqual([(1,)], list(cur))

52
        self.assertTrue(cur.nextset())

53
        self.assertTrue(cur.nextset())

54
        self.assertEqual([(2,)], list(cur))

55
        self.assertFalse(bool(cur.nextset()))

56


57
    @pytest.mark.xfail

58
    def test_multi_cursor(self):

59
        con = self.connect(client_flag=CLIENT.MULTI_STATEMENTS)

60
        cur1 = con.cursor()

61
        cur2 = con.cursor()

62


63
        cur1.execute('SELECT 1; SELECT 2;')

64
        cur2.execute('SELECT 42')

65


66
        self.assertEqual([(1,)], list(cur1))

67
        self.assertEqual([(42,)], list(cur2))

68


69
        r = cur1.nextset()

70
        self.assertTrue(r)

71


72
        self.assertEqual([(2,)], list(cur1))

73
        self.assertIsNone(cur1.nextset())

74


75
    def test_multi_statement_warnings(self):

76
        con = self.connect(

77
            init_command='SELECT "bar"; SELECT "baz"',

78
            client_flag=CLIENT.MULTI_STATEMENTS,

79
        )

80
        cursor = con.cursor()

81


82
        try:

83
            cursor.execute('DROP TABLE IF EXISTS a; ' 'DROP TABLE IF EXISTS b;')

84
        except TypeError:

85
            self.fail()

86


87
    # TODO: How about SSCursor and nextset?

88
    # It's very hard to implement correctly...

89


90