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
|
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()
|
