File: test_connection.py

package info (click to toggle)
aiomysql 0.1.1-2
  • links: PTS, VCS
  • area: main
  • in suites: bookworm, sid, trixie
  • size: 912 kB
  • sloc: python: 6,894; makefile: 213
file content (264 lines) | stat: -rw-r--r-- 8,158 bytes parent folder | download
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
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
import asyncio
import gc
import os

import pytest

import aiomysql


@pytest.fixture()
def fill_my_cnf(mysql_params):
    tests_root = os.path.abspath(os.path.dirname(__file__))

    if "unix_socket" in mysql_params:
        tmpl_path = "fixtures/my.cnf.unix.tmpl"
    else:
        tmpl_path = "fixtures/my.cnf.tcp.tmpl"

    path1 = os.path.join(tests_root, tmpl_path)
    path2 = os.path.join(tests_root, 'fixtures/my.cnf')
    with open(path1) as f1:
        tmpl = f1.read()
    with open(path2, 'w') as f2:
        f2.write(tmpl.format_map(mysql_params))


@pytest.mark.run_loop
async def test_connect_timeout(connection_creator):
    # All exceptions are caught and raised as operational errors
    with pytest.raises(aiomysql.OperationalError):
        await connection_creator(connect_timeout=0.000000000001)


@pytest.mark.run_loop
async def test_config_file(fill_my_cnf, connection_creator, mysql_params):
    tests_root = os.path.abspath(os.path.dirname(__file__))
    path = os.path.join(tests_root, 'fixtures/my.cnf')
    conn = await connection_creator(read_default_file=path)

    if "unix_socket" in mysql_params:
        assert conn.unix_socket == mysql_params["unix_socket"]
    else:
        assert conn.host == mysql_params['host']
        assert conn.port == mysql_params['port']
    assert conn.user, mysql_params['user']

    # make sure connection is working
    cur = await conn.cursor()
    await cur.execute('SELECT 42;')
    (r, ) = await cur.fetchone()
    assert r == 42
    conn.close()


@pytest.mark.run_loop
async def test_config_file_with_different_group(fill_my_cnf,
                                                connection_creator,
                                                mysql_params):
    # same test with config file but actual settings
    # located in not default group.
    tests_root = os.path.abspath(os.path.dirname(__file__))
    path = os.path.join(tests_root, 'fixtures/my.cnf')
    group = 'client_with_unix_socket'
    conn = await connection_creator(read_default_file=path,
                                    read_default_group=group)

    assert conn.charset == 'utf8'
    assert conn.user == 'root'

    # make sure connection is working
    cur = await conn.cursor()
    await cur.execute('SELECT 42;')
    (r, ) = await cur.fetchone()
    assert r == 42
    conn.close()


@pytest.mark.run_loop
async def test_utf8mb4(connection_creator):
    """This test requires MySQL >= 5.5"""
    charset = 'utf8mb4'
    conn = await connection_creator(charset=charset)
    assert conn.charset == charset
    conn.close()


@pytest.mark.run_loop
async def test_largedata(connection_creator):
    """Large query and response (>=16MB)"""
    conn = await connection_creator()
    cur = await conn.cursor()
    await cur.execute("SELECT @@max_allowed_packet")
    r = await cur.fetchone()
    if r[0] < 16 * 1024 * 1024 + 10:
        pytest.skip('Set max_allowed_packet to bigger than 17MB')
    else:
        t = 'a' * (16 * 1024 * 1024)
        await cur.execute("SELECT '" + t + "'")
        r = await cur.fetchone()
        assert r[0] == t


@pytest.mark.run_loop
async def test_escape_string(connection_creator):
    con = await connection_creator()
    cur = await con.cursor()

    assert con.escape("foo'bar") == "'foo\\'bar'"
    # literal is alias for escape
    assert con.literal("foo'bar") == "'foo\\'bar'"
    await cur.execute("SET sql_mode='NO_BACKSLASH_ESCAPES'")
    assert con.escape("foo'bar") == "'foo''bar'"


@pytest.mark.run_loop
async def test_sql_mode_param(connection_creator):
    con = await connection_creator(sql_mode='NO_BACKSLASH_ESCAPES')
    assert con.escape("foo'bar") == "'foo''bar'"


@pytest.mark.run_loop
async def test_init_param(connection_creator):
    init_command = "SET sql_mode='NO_BACKSLASH_ESCAPES';"
    con = await connection_creator(init_command=init_command)
    assert con.escape("foo'bar") == "'foo''bar'"


@pytest.mark.run_loop
async def test_autocommit(connection_creator):
    con = await connection_creator()
    assert con.get_autocommit() is False

    cur = await con.cursor()
    await cur.execute("SET AUTOCOMMIT=1")
    assert con.get_autocommit() is True

    await con.autocommit(False)
    assert con.get_autocommit() is False
    await cur.execute("SELECT @@AUTOCOMMIT")
    r = await cur.fetchone()
    assert r[0] == 0


@pytest.mark.run_loop
async def test_select_db(connection_creator):
    con = await connection_creator()
    current_db = 'test_pymysql'
    other_db = 'test_pymysql2'
    cur = await con.cursor()
    await cur.execute('SELECT database()')
    r = await cur.fetchone()
    assert r[0] == current_db

    await con.select_db(other_db)
    await cur.execute('SELECT database()')
    r = await cur.fetchone()
    assert r[0] == other_db


@pytest.mark.run_loop
async def test_connection_gone_away(connection_creator):
    # test
    # http://dev.mysql.com/doc/refman/5.0/en/gone-away.html
    # http://dev.mysql.com/doc/refman/5.0/en/error-messages-client.html
    # error_cr_server_gone_error
    conn = await connection_creator()
    cur = await conn.cursor()
    await cur.execute("SET wait_timeout=1")
    await asyncio.sleep(2)
    with pytest.raises(aiomysql.OperationalError) as cm:
        await cur.execute("SELECT 1+1")
    # error occures while reading, not writing because of socket buffer.
    # assert cm.exception.args[0] == 2006
    assert cm.value.args[0] in (2006, 2013)
    conn.close()


@pytest.mark.run_loop
async def test_connection_info_methods(connection_creator, mysql_params):
    conn = await connection_creator()
    # trhead id is int
    assert isinstance(conn.thread_id(), int)
    assert conn.character_set_name() in ('latin1', 'utf8mb4')
    if "unix_socket" in mysql_params:
        assert mysql_params["unix_socket"] in conn.get_host_info()
    else:
        assert str(conn.port) in conn.get_host_info()
    assert isinstance(conn.get_server_info(), str)
    # protocol id is int
    assert isinstance(conn.get_proto_info(), int)
    conn.close()


@pytest.mark.run_loop
async def test_connection_set_charset(connection_creator):
    conn = await connection_creator()
    assert conn.character_set_name(), ('latin1' in 'utf8mb4')
    await conn.set_charset('utf8')
    assert conn.character_set_name() == 'utf8'


@pytest.mark.run_loop
async def test_connection_ping(connection_creator):
    conn = await connection_creator()
    await conn.ping()
    assert conn.closed is False
    conn.close()
    await conn.ping()
    assert conn.closed is False


@pytest.mark.run_loop
async def test_connection_properties(connection_creator, mysql_params):
    conn = await connection_creator()
    if "unix_socket" in mysql_params:
        assert conn.unix_socket == mysql_params["unix_socket"]
    else:
        assert conn.host == mysql_params['host']
        assert conn.port == mysql_params['port']
    assert conn.user == mysql_params['user']
    assert conn.db == mysql_params['db']
    assert conn.echo is False
    conn.close()


@pytest.mark.run_loop
async def test_connection_double_ensure_closed(connection_creator):
    conn = await connection_creator()
    assert conn.closed is False
    await conn.ensure_closed()
    assert conn.closed is True
    await conn.ensure_closed()
    assert conn.closed is True


@pytest.mark.run_loop
@pytest.mark.usefixtures("disable_gc")
async def test___del__(connection_creator):
    conn = await connection_creator()
    with pytest.warns(ResourceWarning):
        del conn
        gc.collect()


@pytest.mark.run_loop
async def test_previous_cursor_not_closed(connection_creator):
    conn = await connection_creator()
    cur1 = await conn.cursor()
    await cur1.execute("SELECT 1; SELECT 2")
    cur2 = await conn.cursor()
    await cur2.execute("SELECT 3;")
    resp = await cur2.fetchone()
    assert resp[0] == 3


@pytest.mark.run_loop
async def test_commit_during_multi_result(connection_creator):
    conn = await connection_creator()
    cur = await conn.cursor()
    await cur.execute("SELECT 1; SELECT 2;")
    await conn.commit()
    await cur.execute("SELECT 3;")
    resp = await cur.fetchone()
    assert resp[0] == 3