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
|
from . import capabilities
import pymysql
from pymysql.tests import base
import warnings
warnings.filterwarnings("error")
class test_MySQLdb(capabilities.DatabaseTest):
db_module = pymysql
connect_args = ()
connect_kwargs = base.PyMySQLTestCase.databases[0].copy()
connect_kwargs.update(
dict(
read_default_file="~/.my.cnf",
use_unicode=True,
binary_prefix=True,
charset="utf8mb4",
sql_mode="ANSI,STRICT_TRANS_TABLES,TRADITIONAL",
)
)
leak_test = False
def quote_identifier(self, ident):
return "`%s`" % ident
def test_TIME(self):
from datetime import timedelta
def generator(row, col):
return timedelta(0, row * 8000)
self.check_data_integrity(("col1 TIME",), generator)
def test_TINYINT(self):
# Number data
def generator(row, col):
v = (row * row) % 256
if v > 127:
v = v - 256
return v
self.check_data_integrity(("col1 TINYINT",), generator)
def test_stored_procedures(self):
db = self.connection
c = self.cursor
try:
self.create_table(("pos INT", "tree CHAR(20)"))
c.executemany(
"INSERT INTO %s (pos,tree) VALUES (%%s,%%s)" % self.table,
list(enumerate("ash birch cedar larch pine".split())),
)
db.commit()
c.execute(
"""
CREATE PROCEDURE test_sp(IN t VARCHAR(255))
BEGIN
SELECT pos FROM %s WHERE tree = t;
END
"""
% self.table
)
db.commit()
c.callproc("test_sp", ("larch",))
rows = c.fetchall()
self.assertEqual(len(rows), 1)
self.assertEqual(rows[0][0], 3)
c.nextset()
finally:
c.execute("DROP PROCEDURE IF EXISTS test_sp")
c.execute("drop table %s" % (self.table))
def test_small_CHAR(self):
# Character data
def generator(row, col):
i = ((row + 1) * (col + 1) + 62) % 256
if i == 62:
return ""
if i == 63:
return None
return chr(i)
self.check_data_integrity(("col1 char(1)", "col2 char(1)"), generator)
def test_bug_2671682(self):
from pymysql.constants import ER
try:
self.cursor.execute("describe some_non_existent_table")
except self.connection.ProgrammingError as msg:
self.assertEqual(msg.args[0], ER.NO_SUCH_TABLE)
def test_ping(self):
self.connection.ping()
def test_literal_int(self):
self.assertTrue("2" == self.connection.literal(2))
def test_literal_float(self):
self.assertEqual("3.1415e0", self.connection.literal(3.1415))
def test_literal_string(self):
self.assertTrue("'foo'" == self.connection.literal("foo"))
|
