1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
|
import sys
try:
from pymysql.tests import base
import pymysql.cursors
from pymysql.constants import CLIENT
except Exception:
# For local testing from top-level directory, without installing
sys.path.append("../pymysql")
from pymysql.tests import base
import pymysql.cursors
from pymysql.constants import CLIENT
class TestSSCursor(base.PyMySQLTestCase):
def test_SSCursor(self):
affected_rows = 18446744073709551615
conn = self.connect(client_flag=CLIENT.MULTI_STATEMENTS)
data = [
("America", "", "America/Jamaica"),
("America", "", "America/Los_Angeles"),
("America", "", "America/Lima"),
("America", "", "America/New_York"),
("America", "", "America/Menominee"),
("America", "", "America/Havana"),
("America", "", "America/El_Salvador"),
("America", "", "America/Costa_Rica"),
("America", "", "America/Denver"),
("America", "", "America/Detroit"),
]
cursor = conn.cursor(pymysql.cursors.SSCursor)
# Create table
cursor.execute(
"CREATE TABLE tz_data ("
"region VARCHAR(64),"
"zone VARCHAR(64),"
"name VARCHAR(64))"
)
conn.begin()
# Test INSERT
for i in data:
cursor.execute("INSERT INTO tz_data VALUES (%s, %s, %s)", i)
self.assertEqual(conn.affected_rows(), 1, "affected_rows does not match")
conn.commit()
# Test fetchone()
iter = 0
cursor.execute("SELECT * FROM tz_data")
while True:
row = cursor.fetchone()
if row is None:
break
iter += 1
# Test cursor.rowcount
self.assertEqual(
cursor.rowcount,
affected_rows,
"cursor.rowcount != %s" % (str(affected_rows)),
)
# Test cursor.rownumber
self.assertEqual(
cursor.rownumber, iter, "cursor.rowcount != %s" % (str(iter))
)
# Test row came out the same as it went in
self.assertEqual((row in data), True, "Row not found in source data")
# Test fetchall
cursor.execute("SELECT * FROM tz_data")
self.assertEqual(
len(cursor.fetchall()),
len(data),
"fetchall failed. Number of rows does not match",
)
# Test fetchmany
cursor.execute("SELECT * FROM tz_data")
self.assertEqual(
len(cursor.fetchmany(2)),
2,
"fetchmany failed. Number of rows does not match",
)
# So MySQLdb won't throw "Commands out of sync"
while True:
res = cursor.fetchone()
if res is None:
break
# Test update, affected_rows()
cursor.execute("UPDATE tz_data SET zone = %s", ["Foo"])
conn.commit()
self.assertEqual(
cursor.rowcount,
len(data),
"Update failed. affected_rows != %s" % (str(len(data))),
)
# Test executemany
cursor.executemany("INSERT INTO tz_data VALUES (%s, %s, %s)", data)
self.assertEqual(
cursor.rowcount,
len(data),
"executemany failed. cursor.rowcount != %s" % (str(len(data))),
)
# Test multiple datasets
cursor.execute("SELECT 1; SELECT 2; SELECT 3")
self.assertListEqual(list(cursor), [(1,)])
self.assertTrue(cursor.nextset())
self.assertListEqual(list(cursor), [(2,)])
self.assertTrue(cursor.nextset())
self.assertListEqual(list(cursor), [(3,)])
self.assertFalse(cursor.nextset())
cursor.execute("DROP TABLE IF EXISTS tz_data")
cursor.close()
__all__ = ["TestSSCursor"]
if __name__ == "__main__":
import unittest
unittest.main()
|
