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
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
|
import copy
import aiomysql.cursors
import pytest
BOB = ("bob", 21, {"k1": "pretty", "k2": [18, 25]})
JIM = ("jim", 56, {"k1": "rich", "k2": [20, 60]})
FRED = ("fred", 100, {"k1": "longevity", "k2": [100, 160]})
@pytest.fixture()
async def prepare(connection):
havejson = True
c = await connection.cursor(aiomysql.cursors.DeserializationCursor)
# create a table ane some data to query
await c.execute("drop table if exists deserialize_cursor")
await c.execute("select VERSION()")
v = await c.fetchone()
version, *db_type = v[0].split('-', 1)
version = float(".".join(version.split('.', 2)[:2]))
ismariadb = db_type and 'mariadb' in db_type[0].lower()
if ismariadb or version < 5.7:
await c.execute(
"""CREATE TABLE deserialize_cursor
(name char(20), age int , claim text)""")
havejson = False
else:
await c.execute(
"""CREATE TABLE deserialize_cursor
(name char(20), age int , claim json)""")
data = [("bob", 21, '{"k1": "pretty", "k2": [18, 25]}'),
("jim", 56, '{"k1": "rich", "k2": [20, 60]}'),
("fred", 100, '{"k1": "longevity", "k2": [100, 160]}')]
await c.executemany("insert into deserialize_cursor values "
"(%s,%s,%s)", data)
return havejson
@pytest.mark.run_loop
async def test_deserialize_cursor(prepare, connection):
havejson = await prepare
if not havejson:
return
bob, jim, fred = copy.deepcopy(BOB), copy.deepcopy(
JIM), copy.deepcopy(FRED)
# all assert test compare to the structure as would come
# out from MySQLdb
conn = connection
c = await conn.cursor(aiomysql.cursors.DeserializationCursor)
# pull back the single row dict for bob and check
await c.execute("SELECT * from deserialize_cursor "
"where name='bob'")
r = await c.fetchone()
assert bob == r, "fetchone via DeserializeCursor failed"
# same again, but via fetchall => tuple)
await c.execute("SELECT * from deserialize_cursor "
"where name='bob'")
r = await c.fetchall()
assert [bob] == r, \
"fetch a 1 row result via fetchall failed via DeserializeCursor"
# get all 3 row via fetchall
await c.execute("SELECT * from deserialize_cursor")
r = await c.fetchall()
assert [bob, jim, fred] == r, "fetchall failed via DictCursor"
# get all 2 row via fetchmany
await c.execute("SELECT * from deserialize_cursor")
r = await c.fetchmany(2)
assert [bob, jim] == r, "fetchmany failed via DictCursor"
await c.execute('commit')
@pytest.mark.run_loop
async def test_deserialize_cursor_low_version(prepare, connection):
havejson = await prepare
if havejson:
return
bob = ("bob", 21, '{"k1": "pretty", "k2": [18, 25]}')
jim = ("jim", 56, '{"k1": "rich", "k2": [20, 60]}')
fred = ("fred", 100, '{"k1": "longevity", "k2": [100, 160]}')
# all assert test compare to the structure as would come
# out from MySQLdb
conn = connection
c = await conn.cursor(aiomysql.cursors.DeserializationCursor)
# pull back the single row dict for bob and check
await c.execute("SELECT * from deserialize_cursor where name='bob'")
r = await c.fetchone()
assert bob == r, "fetchone via DeserializeCursor failed"
# same again, but via fetchall => tuple)
await c.execute("SELECT * from deserialize_cursor "
"where name='bob'")
r = await c.fetchall()
assert [bob] == r, \
"fetch a 1 row result via fetchall failed via DeserializeCursor"
# get all 3 row via fetchall
await c.execute("SELECT * from deserialize_cursor")
r = await c.fetchall()
assert [bob, jim, fred] == r, "fetchall failed via DictCursor"
# get all 2 row via fetchmany
await c.execute("SELECT * from deserialize_cursor")
r = await c.fetchmany(2)
assert [bob, jim] == r, "fetchmany failed via DictCursor"
await c.execute('commit')
@pytest.mark.run_loop
async def test_deserializedictcursor(prepare, connection):
havejson = await prepare
if not havejson:
return
bob = {'name': 'bob', 'age': 21,
'claim': {"k1": "pretty", "k2": [18, 25]}}
conn = connection
c = await conn.cursor(aiomysql.cursors.DeserializationCursor,
aiomysql.cursors.DictCursor)
await c.execute("SELECT * from deserialize_cursor "
"where name='bob'")
r = await c.fetchall()
assert [bob] == r, \
"fetch a 1 row result via fetchall failed via DeserializationCursor"
@pytest.mark.run_loop
async def test_ssdeserializecursor(prepare, connection):
havejson = await prepare
if not havejson:
return
conn = connection
c = await conn.cursor(aiomysql.cursors.SSCursor,
aiomysql.cursors.DeserializationCursor)
await c.execute("SELECT * from deserialize_cursor "
"where name='bob'")
r = await c.fetchall()
assert [BOB] == r, \
"fetch a 1 row result via fetchall failed via DeserializationCursor"
@pytest.mark.run_loop
async def test_ssdeserializedictcursor(prepare, connection):
havejson = await prepare
if not havejson:
return
bob = {'name': 'bob', 'age': 21,
'claim': {"k1": "pretty", "k2": [18, 25]}}
conn = connection
c = await conn.cursor(aiomysql.cursors.SSCursor,
aiomysql.cursors.DeserializationCursor,
aiomysql.cursors.DictCursor)
await c.execute("SELECT * from deserialize_cursor "
"where name='bob'")
r = await c.fetchall()
assert [bob] == r, \
"fetch a 1 row result via fetchall failed via DeserializationCursor"
|
