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
|
import re
import pytest
from asyncmy.connection import Connection
from asyncmy.errors import OperationalError
from conftest import connection_kwargs
@pytest.mark.asyncio
async def test_connect():
connection = Connection(**connection_kwargs)
await connection.connect()
assert connection._connected
assert re.match(
r"\d+\.\d+\.\d+([^0-9].*)?",
connection.get_server_info(),
)
assert connection.get_proto_info() == 10
assert connection.get_host_info() != "Not Connected"
await connection.ensure_closed()
@pytest.mark.asyncio
async def test_read_timeout():
with pytest.raises(OperationalError):
connection = Connection(read_timeout=1, **connection_kwargs)
await connection.connect()
async with connection.cursor() as cursor:
await cursor.execute("DO SLEEP(3)")
@pytest.mark.asyncio
async def test_transaction(connection):
await connection.begin()
await connection.query(
"""INSERT INTO test.asyncmy(`decimal`, date, datetime, `float`,
string, `tinyint`) VALUES (%s,'%s','%s',%s,'%s',%s)"""
% (
1,
"2020-08-08",
"2020-08-08 00:00:00",
1,
"1",
1,
),
True,
)
await connection.rollback()
|
