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
|
#!/usr/bin/env python -O
# -*- coding: utf-8 -*-
import unittest
import mariadb
from test.base_test import create_connection, is_skysql, is_maxscale, is_mysql
from test.conf_test import conf
class CursorTest(unittest.TestCase):
def setUp(self):
self.connection = create_connection()
def tearDown(self):
del self.connection
def test_ping(self):
if is_maxscale():
self.skipTest("MAXSCALE return wrong thread id")
new_conn = create_connection()
id = new_conn.connection_id
self.connection.kill(id)
try:
new_conn.ping()
except mariadb.InterfaceError:
pass
del new_conn
new_conn = create_connection()
new_conn.auto_reconnect = True
id = new_conn.connection_id
self.connection.kill(id)
new_conn.ping()
new_id = new_conn.connection_id
self.assertTrue(id != new_id)
del new_conn
def test_change_user(self):
if is_skysql():
self.skipTest("SkySQL failure")
if is_maxscale():
self.skipTest("MAXSCALE doesn't get new user immediately")
if self.connection.server_name == "localhost":
curs = self.connection.cursor(buffered=True)
curs.execute("select * from information_schema.plugins "
"where plugin_name='unix_socket' "
"and plugin_status='ACTIVE'")
if curs.rowcount > 0:
del curs
self.skipTest("unix_socket is active")
del curs
default_conf = conf()
cursor = self.connection.cursor()
cursor.execute("drop user if exists foo")
if is_mysql() and self.connection.server_version < 80000:
cursor.execute("create user foo@'%'")
cursor.execute("GRANT ALL on `"
+ default_conf["database"] +
"`.* TO foo@'%' IDENTIFIED BY "
"'heyPassw-!µ20§rd'")
else:
cursor.execute("create user foo@'%' IDENTIFIED "
"BY 'heyPassw-!µ20§rd'")
cursor.execute("GRANT ALL on `" + default_conf["database"] +
"`.* TO foo@'%'")
new_conn = create_connection()
new_conn.change_user("foo", "heyPassw-!µ20§rd", "")
self.assertEqual("foo", new_conn.user)
cursor.execute("drop user foo")
del new_conn
del cursor
def test_reconnect(self):
if is_maxscale():
self.skipTest("MAXSCALE wrong thread id")
new_conn = create_connection()
conn1_id = new_conn.connection_id
self.connection.kill(conn1_id)
new_conn.reconnect()
conn2_id = new_conn.connection_id
self.assertFalse(conn1_id == conn2_id)
del new_conn
def test_reset(self):
if self.connection.server_version < 100204:
self.skipTest("RESET not supported")
cursor = self.connection.cursor()
cursor.execute("SELECT 1 UNION SELECT 2")
try:
self.connection.ping()
except mariadb.InterfaceError:
pass
self.connection.reset()
self.connection.ping()
del cursor
def test_warnings(self):
conn = self.connection
cursor = conn.cursor()
cursor.execute("SET session sql_mode=''")
cursor.execute("CREATE TEMPORARY TABLE test_warnings (a tinyint)")
cursor.execute("INSERT INTO test_warnings VALUES (300)")
self.assertEqual(conn.warnings, 1)
self.assertEqual(conn.warnings, cursor.warnings)
del cursor
def test_server_infos(self):
self.assertTrue(self.connection.server_info)
self.assertTrue(self.connection.server_version > 0)
def test_escape(self):
cursor = self.connection.cursor()
cursor.execute("CREATE TEMPORARY TABLE test_escape (a varchar(100))")
str = 'This is a \ and a \"' # noqa: W605
cmd = "INSERT INTO test_escape VALUES('%s')" % str
try:
cursor.execute(cmd)
except mariadb.DatabaseError:
pass
str = self.connection.escape_string(str)
cmd = "INSERT INTO test_escape VALUES('%s')" % str
cursor.execute(cmd)
del cursor
def test_conpy279(self):
conn = self.connection
default_conf = conf()
if "password" not in default_conf:
default_conf["password"] = None
try:
conn.change_user(None, None, None)
except TypeError:
pass
conn.change_user(default_conf["user"], default_conf["password"], None)
conn.close()
if __name__ == '__main__':
unittest.main()
|
