1
1
import pymysql
2
2
3
3
import pytest
4
- import testing .mysqld
5
4
6
5
from aws_xray_sdk .core import patch
7
6
from aws_xray_sdk .core import xray_recorder
8
7
from aws_xray_sdk .core .context import Context
9
8
from aws_xray_sdk .ext .pymysql import unpatch
10
9
10
+ MYSQL_USER = "root"
11
+ MYSQL_PASSWORD = "root"
12
+ MYSQL_HOST = "localhost"
13
+ MYSQL_PORT = 3306
14
+ MYSQL_DB_NAME = "test_db"
11
15
12
16
@pytest .fixture (scope = 'module' , autouse = True )
13
17
def patch_module ():
@@ -32,46 +36,41 @@ def construct_ctx():
32
36
33
37
def test_execute_dsn_kwargs ():
34
38
q = 'SELECT 1'
35
- with testing .mysqld .Mysqld () as mysqld :
36
- dsn = mysqld .dsn ()
37
- conn = pymysql .connect (database = dsn ['db' ],
38
- user = dsn ['user' ],
39
- password = '' ,
40
- host = dsn ['host' ],
41
- port = dsn ['port' ])
42
- cur = conn .cursor ()
43
- cur .execute (q )
39
+ conn = pymysql .connect (database = MYSQL_DB_NAME ,
40
+ user = MYSQL_USER ,
41
+ password = MYSQL_PASSWORD ,
42
+ host = MYSQL_HOST ,
43
+ port = MYSQL_PORT )
44
+ cur = conn .cursor ()
45
+ cur .execute (q )
44
46
45
47
subsegment = xray_recorder .current_segment ().subsegments [- 1 ]
46
48
assert subsegment .name == 'execute'
47
49
sql = subsegment .sql
48
50
assert sql ['database_type' ] == 'MySQL'
49
- assert sql ['user' ] == dsn [ 'user' ]
51
+ assert sql ['user' ] == MYSQL_USER
50
52
assert sql ['driver_version' ] == 'PyMySQL'
51
53
assert sql ['database_version' ]
52
54
53
55
54
56
def test_execute_bad_query ():
55
57
q = "SELECT blarg"
56
- with testing .mysqld .Mysqld () as mysqld :
57
- dsn = mysqld .dsn ()
58
- conn = pymysql .connect (database = dsn ['db' ],
59
- user = dsn ['user' ],
60
- password = '' ,
61
- host = dsn ['host' ],
62
- port = dsn ['port' ])
63
-
64
- cur = conn .cursor ()
65
- try :
66
- cur .execute (q )
67
- except Exception :
68
- pass
69
-
58
+ conn = pymysql .connect (database = MYSQL_DB_NAME ,
59
+ user = MYSQL_USER ,
60
+ password = MYSQL_PASSWORD ,
61
+ host = MYSQL_HOST ,
62
+ port = MYSQL_PORT )
63
+ cur = conn .cursor ()
64
+ try :
65
+ cur .execute (q )
66
+ except Exception :
67
+ pass
68
+
70
69
subsegment = xray_recorder .current_segment ().subsegments [- 1 ]
71
70
assert subsegment .name == "execute"
72
71
sql = subsegment .sql
73
72
assert sql ['database_type' ] == 'MySQL'
74
- assert sql ['user' ] == dsn [ 'user' ]
73
+ assert sql ['user' ] == MYSQL_USER
75
74
assert sql ['driver_version' ] == 'PyMySQL'
76
75
assert sql ['database_version' ]
77
76
0 commit comments