initial commit
This commit is contained in:
Massimo Di Pierro committed
commit
d421c1321b
580 files changed
+115527
No files matched your search
@@ -0,0 +1,7 @@
|
||||
from pymysql.tests.test_issues import *
|
||||
from pymysql.tests.test_example import *
|
||||
from pymysql.tests.test_basic import *
|
||||
|
||||
if __name__ == "__main__":
|
||||
import unittest
|
||||
unittest.main()
|
||||
@@ -0,0 +1,19 @@
|
||||
import pymysql
|
||||
import unittest
|
||||
|
||||
class PyMySQLTestCase(unittest.TestCase):
|
||||
databases = [
|
||||
{"host":"localhost","user":"root",
|
||||
"passwd":"","db":"test_pymysql", "use_unicode": True},
|
||||
{"host":"localhost","user":"root","passwd":"","db":"test_pymysql2"}]
|
||||
|
||||
def setUp(self):
|
||||
self.connections = []
|
||||
|
||||
for params in self.databases:
|
||||
self.connections.append(pymysql.connect(**params))
|
||||
|
||||
def tearDown(self):
|
||||
for connection in self.connections:
|
||||
connection.close()
|
||||
|
||||
@@ -0,0 +1,141 @@
|
||||
from pymysql.tests import base
|
||||
from pymysql import util
|
||||
|
||||
import time
|
||||
import datetime
|
||||
|
||||
class TestConversion(base.PyMySQLTestCase):
|
||||
def test_datatypes(self):
|
||||
""" test every data type """
|
||||
conn = self.connections[0]
|
||||
c = conn.cursor()
|
||||
c.execute("create table test_datatypes (b bit, i int, l bigint, f real, s varchar(32), u varchar(32), bb blob, d date, dt datetime, ts timestamp, td time, t time, st datetime)")
|
||||
try:
|
||||
# insert values
|
||||
v = (True, -3, 123456789012, 5.7, "hello'\" world", u"Espa\xc3\xb1ol", "binary\x00data".encode(conn.charset), datetime.date(1988,2,2), datetime.datetime.now(), datetime.timedelta(5,6), datetime.time(16,32), time.localtime())
|
||||
c.execute("insert into test_datatypes (b,i,l,f,s,u,bb,d,dt,td,t,st) values (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)", v)
|
||||
c.execute("select b,i,l,f,s,u,bb,d,dt,td,t,st from test_datatypes")
|
||||
r = c.fetchone()
|
||||
self.assertEqual(util.int2byte(1), r[0])
|
||||
self.assertEqual(v[1:8], r[1:8])
|
||||
# mysql throws away microseconds so we need to check datetimes
|
||||
# specially. additionally times are turned into timedeltas.
|
||||
self.assertEqual(datetime.datetime(*v[8].timetuple()[:6]), r[8])
|
||||
self.assertEqual(v[9], r[9]) # just timedeltas
|
||||
self.assertEqual(datetime.timedelta(0, 60 * (v[10].hour * 60 + v[10].minute)), r[10])
|
||||
self.assertEqual(datetime.datetime(*v[-1][:6]), r[-1])
|
||||
|
||||
c.execute("delete from test_datatypes")
|
||||
|
||||
# check nulls
|
||||
c.execute("insert into test_datatypes (b,i,l,f,s,u,bb,d,dt,td,t,st) values (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)", [None] * 12)
|
||||
c.execute("select b,i,l,f,s,u,bb,d,dt,td,t,st from test_datatypes")
|
||||
r = c.fetchone()
|
||||
self.assertEqual(tuple([None] * 12), r)
|
||||
|
||||
c.execute("delete from test_datatypes")
|
||||
|
||||
# check sequence type
|
||||
c.execute("insert into test_datatypes (i, l) values (2,4), (6,8), (10,12)")
|
||||
c.execute("select l from test_datatypes where i in %s order by i", ((2,6),))
|
||||
r = c.fetchall()
|
||||
self.assertEqual(((4,),(8,)), r)
|
||||
finally:
|
||||
c.execute("drop table test_datatypes")
|
||||
|
||||
def test_dict(self):
|
||||
""" test dict escaping """
|
||||
conn = self.connections[0]
|
||||
c = conn.cursor()
|
||||
c.execute("create table test_dict (a integer, b integer, c integer)")
|
||||
try:
|
||||
c.execute("insert into test_dict (a,b,c) values (%(a)s, %(b)s, %(c)s)", {"a":1,"b":2,"c":3})
|
||||
c.execute("select a,b,c from test_dict")
|
||||
self.assertEqual((1,2,3), c.fetchone())
|
||||
finally:
|
||||
c.execute("drop table test_dict")
|
||||
|
||||
def test_big_blob(self):
|
||||
""" test tons of data """
|
||||
conn = self.connections[0]
|
||||
c = conn.cursor()
|
||||
c.execute("create table test_big_blob (b blob)")
|
||||
try:
|
||||
data = "pymysql" * 1024
|
||||
c.execute("insert into test_big_blob (b) values (%s)", (data,))
|
||||
c.execute("select b from test_big_blob")
|
||||
self.assertEqual(data.encode(conn.charset), c.fetchone()[0])
|
||||
finally:
|
||||
c.execute("drop table test_big_blob")
|
||||
|
||||
class TestCursor(base.PyMySQLTestCase):
|
||||
# this test case does not work quite right yet, however,
|
||||
# we substitute in None for the erroneous field which is
|
||||
# compatible with the DB-API 2.0 spec and has not broken
|
||||
# any unit tests for anything we've tried.
|
||||
|
||||
#def test_description(self):
|
||||
# """ test description attribute """
|
||||
# # result is from MySQLdb module
|
||||
# r = (('Host', 254, 11, 60, 60, 0, 0),
|
||||
# ('User', 254, 16, 16, 16, 0, 0),
|
||||
# ('Password', 254, 41, 41, 41, 0, 0),
|
||||
# ('Select_priv', 254, 1, 1, 1, 0, 0),
|
||||
# ('Insert_priv', 254, 1, 1, 1, 0, 0),
|
||||
# ('Update_priv', 254, 1, 1, 1, 0, 0),
|
||||
# ('Delete_priv', 254, 1, 1, 1, 0, 0),
|
||||
# ('Create_priv', 254, 1, 1, 1, 0, 0),
|
||||
# ('Drop_priv', 254, 1, 1, 1, 0, 0),
|
||||
# ('Reload_priv', 254, 1, 1, 1, 0, 0),
|
||||
# ('Shutdown_priv', 254, 1, 1, 1, 0, 0),
|
||||
# ('Process_priv', 254, 1, 1, 1, 0, 0),
|
||||
# ('File_priv', 254, 1, 1, 1, 0, 0),
|
||||
# ('Grant_priv', 254, 1, 1, 1, 0, 0),
|
||||
# ('References_priv', 254, 1, 1, 1, 0, 0),
|
||||
# ('Index_priv', 254, 1, 1, 1, 0, 0),
|
||||
# ('Alter_priv', 254, 1, 1, 1, 0, 0),
|
||||
# ('Show_db_priv', 254, 1, 1, 1, 0, 0),
|
||||
# ('Super_priv', 254, 1, 1, 1, 0, 0),
|
||||
# ('Create_tmp_table_priv', 254, 1, 1, 1, 0, 0),
|
||||
# ('Lock_tables_priv', 254, 1, 1, 1, 0, 0),
|
||||
# ('Execute_priv', 254, 1, 1, 1, 0, 0),
|
||||
# ('Repl_slave_priv', 254, 1, 1, 1, 0, 0),
|
||||
# ('Repl_client_priv', 254, 1, 1, 1, 0, 0),
|
||||
# ('Create_view_priv', 254, 1, 1, 1, 0, 0),
|
||||
# ('Show_view_priv', 254, 1, 1, 1, 0, 0),
|
||||
# ('Create_routine_priv', 254, 1, 1, 1, 0, 0),
|
||||
# ('Alter_routine_priv', 254, 1, 1, 1, 0, 0),
|
||||
# ('Create_user_priv', 254, 1, 1, 1, 0, 0),
|
||||
# ('Event_priv', 254, 1, 1, 1, 0, 0),
|
||||
# ('Trigger_priv', 254, 1, 1, 1, 0, 0),
|
||||
# ('ssl_type', 254, 0, 9, 9, 0, 0),
|
||||
# ('ssl_cipher', 252, 0, 65535, 65535, 0, 0),
|
||||
# ('x509_issuer', 252, 0, 65535, 65535, 0, 0),
|
||||
# ('x509_subject', 252, 0, 65535, 65535, 0, 0),
|
||||
# ('max_questions', 3, 1, 11, 11, 0, 0),
|
||||
# ('max_updates', 3, 1, 11, 11, 0, 0),
|
||||
# ('max_connections', 3, 1, 11, 11, 0, 0),
|
||||
# ('max_user_connections', 3, 1, 11, 11, 0, 0))
|
||||
# conn = self.connections[0]
|
||||
# c = conn.cursor()
|
||||
# c.execute("select * from mysql.user")
|
||||
#
|
||||
# self.assertEqual(r, c.description)
|
||||
|
||||
def test_fetch_no_result(self):
|
||||
""" test a fetchone() with no rows """
|
||||
conn = self.connections[0]
|
||||
c = conn.cursor()
|
||||
c.execute("create table test_nr (b varchar(32))")
|
||||
try:
|
||||
data = "pymysql"
|
||||
c.execute("insert into test_nr (b) values (%s)", (data,))
|
||||
self.assertEqual(None, c.fetchone())
|
||||
finally:
|
||||
c.execute("drop table test_nr")
|
||||
|
||||
__all__ = ["TestConversion","TestCursor"]
|
||||
|
||||
if __name__ == "__main__":
|
||||
import unittest
|
||||
unittest.main()
|
||||
@@ -0,0 +1,32 @@
|
||||
import pymysql
|
||||
from pymysql.tests import base
|
||||
|
||||
class TestExample(base.PyMySQLTestCase):
|
||||
def test_example(self):
|
||||
conn = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='', db='mysql')
|
||||
|
||||
|
||||
cur = conn.cursor()
|
||||
|
||||
cur.execute("SELECT Host,User FROM user")
|
||||
|
||||
# print cur.description
|
||||
|
||||
# r = cur.fetchall()
|
||||
# print r
|
||||
# ...or...
|
||||
u = False
|
||||
|
||||
for r in cur.fetchall():
|
||||
u = u or conn.user in r
|
||||
|
||||
self.assertTrue(u)
|
||||
|
||||
cur.close()
|
||||
conn.close()
|
||||
|
||||
__all__ = ["TestExample"]
|
||||
|
||||
if __name__ == "__main__":
|
||||
import unittest
|
||||
unittest.main()
|
||||
@@ -0,0 +1,240 @@
|
||||
import pymysql
|
||||
from pymysql.tests import base
|
||||
|
||||
import sys
|
||||
|
||||
try:
|
||||
import imp
|
||||
reload = imp.reload
|
||||
except AttributeError:
|
||||
pass
|
||||
|
||||
import datetime
|
||||
|
||||
class TestOldIssues(base.PyMySQLTestCase):
|
||||
def test_issue_3(self):
|
||||
""" undefined methods datetime_or_None, date_or_None """
|
||||
conn = self.connections[0]
|
||||
c = conn.cursor()
|
||||
c.execute("create table issue3 (d date, t time, dt datetime, ts timestamp)")
|
||||
try:
|
||||
c.execute("insert into issue3 (d, t, dt, ts) values (%s,%s,%s,%s)", (None, None, None, None))
|
||||
c.execute("select d from issue3")
|
||||
self.assertEqual(None, c.fetchone()[0])
|
||||
c.execute("select t from issue3")
|
||||
self.assertEqual(None, c.fetchone()[0])
|
||||
c.execute("select dt from issue3")
|
||||
self.assertEqual(None, c.fetchone()[0])
|
||||
c.execute("select ts from issue3")
|
||||
self.assertTrue(isinstance(c.fetchone()[0], datetime.datetime))
|
||||
finally:
|
||||
c.execute("drop table issue3")
|
||||
|
||||
def test_issue_4(self):
|
||||
""" can't retrieve TIMESTAMP fields """
|
||||
conn = self.connections[0]
|
||||
c = conn.cursor()
|
||||
c.execute("create table issue4 (ts timestamp)")
|
||||
try:
|
||||
c.execute("insert into issue4 (ts) values (now())")
|
||||
c.execute("select ts from issue4")
|
||||
self.assertTrue(isinstance(c.fetchone()[0], datetime.datetime))
|
||||
finally:
|
||||
c.execute("drop table issue4")
|
||||
|
||||
def test_issue_5(self):
|
||||
""" query on information_schema.tables fails """
|
||||
con = self.connections[0]
|
||||
cur = con.cursor()
|
||||
cur.execute("select * from information_schema.tables")
|
||||
|
||||
def test_issue_6(self):
|
||||
""" exception: TypeError: ord() expected a character, but string of length 0 found """
|
||||
conn = pymysql.connect(host="localhost",user="root",passwd="",db="mysql")
|
||||
c = conn.cursor()
|
||||
c.execute("select * from user")
|
||||
conn.close()
|
||||
|
||||
def test_issue_8(self):
|
||||
""" Primary Key and Index error when selecting data """
|
||||
conn = self.connections[0]
|
||||
c = conn.cursor()
|
||||
c.execute("""CREATE TABLE `test` (`station` int(10) NOT NULL DEFAULT '0', `dh`
|
||||
datetime NOT NULL DEFAULT '0000-00-00 00:00:00', `echeance` int(1) NOT NULL
|
||||
DEFAULT '0', `me` double DEFAULT NULL, `mo` double DEFAULT NULL, PRIMARY
|
||||
KEY (`station`,`dh`,`echeance`)) ENGINE=MyISAM DEFAULT CHARSET=latin1;""")
|
||||
try:
|
||||
self.assertEqual(0, c.execute("SELECT * FROM test"))
|
||||
c.execute("ALTER TABLE `test` ADD INDEX `idx_station` (`station`)")
|
||||
self.assertEqual(0, c.execute("SELECT * FROM test"))
|
||||
finally:
|
||||
c.execute("drop table test")
|
||||
|
||||
def test_issue_9(self):
|
||||
""" sets DeprecationWarning in Python 2.6 """
|
||||
try:
|
||||
reload(pymysql)
|
||||
except DeprecationWarning:
|
||||
self.fail()
|
||||
|
||||
def test_issue_10(self):
|
||||
""" Allocate a variable to return when the exception handler is permissive """
|
||||
conn = self.connections[0]
|
||||
conn.errorhandler = lambda cursor, errorclass, errorvalue: None
|
||||
cur = conn.cursor()
|
||||
cur.execute( "create table t( n int )" )
|
||||
cur.execute( "create table t( n int )" )
|
||||
|
||||
def test_issue_13(self):
|
||||
""" can't handle large result fields """
|
||||
conn = self.connections[0]
|
||||
cur = conn.cursor()
|
||||
cur.execute("create table issue13 (t text)")
|
||||
try:
|
||||
# ticket says 18k
|
||||
size = 18*1024
|
||||
cur.execute("insert into issue13 (t) values (%s)", ("x" * size,))
|
||||
cur.execute("select t from issue13")
|
||||
# use assert_ so that obscenely huge error messages don't print
|
||||
r = cur.fetchone()[0]
|
||||
self.assert_("x" * size == r)
|
||||
finally:
|
||||
cur.execute("drop table issue13")
|
||||
|
||||
def test_issue_14(self):
|
||||
""" typo in converters.py """
|
||||
self.assertEqual('1', pymysql.converters.escape_item(1, "utf8"))
|
||||
self.assertEqual('1', pymysql.converters.escape_item(1L, "utf8"))
|
||||
|
||||
self.assertEqual('1', pymysql.converters.escape_object(1))
|
||||
self.assertEqual('1', pymysql.converters.escape_object(1L))
|
||||
|
||||
def test_issue_15(self):
|
||||
""" query should be expanded before perform character encoding """
|
||||
conn = self.connections[0]
|
||||
c = conn.cursor()
|
||||
c.execute("create table issue15 (t varchar(32))")
|
||||
try:
|
||||
c.execute("insert into issue15 (t) values (%s)", (u'\xe4\xf6\xfc'))
|
||||
c.execute("select t from issue15")
|
||||
self.assertEqual(u'\xe4\xf6\xfc', c.fetchone()[0])
|
||||
finally:
|
||||
c.execute("drop table issue15")
|
||||
|
||||
def test_issue_16(self):
|
||||
""" Patch for string and tuple escaping """
|
||||
conn = self.connections[0]
|
||||
c = conn.cursor()
|
||||
c.execute("create table issue16 (name varchar(32) primary key, email varchar(32))")
|
||||
try:
|
||||
c.execute("insert into issue16 (name, email) values ('pete', 'floydophone')")
|
||||
c.execute("select email from issue16 where name=%s", ("pete",))
|
||||
self.assertEqual("floydophone", c.fetchone()[0])
|
||||
finally:
|
||||
c.execute("drop table issue16")
|
||||
|
||||
def test_issue_17(self):
|
||||
""" could not connect mysql use passwod """
|
||||
conn = self.connections[0]
|
||||
host = self.databases[0]["host"]
|
||||
db = self.databases[0]["db"]
|
||||
c = conn.cursor()
|
||||
# grant access to a table to a user with a password
|
||||
try:
|
||||
c.execute("create table issue17 (x varchar(32) primary key)")
|
||||
c.execute("insert into issue17 (x) values ('hello, world!')")
|
||||
c.execute("grant all privileges on %s.issue17 to 'issue17user'@'%%' identified by '1234'" % db)
|
||||
conn.commit()
|
||||
|
||||
conn2 = pymysql.connect(host=host, user="issue17user", passwd="1234", db=db)
|
||||
c2 = conn2.cursor()
|
||||
c2.execute("select x from issue17")
|
||||
self.assertEqual("hello, world!", c2.fetchone()[0])
|
||||
finally:
|
||||
c.execute("drop table issue17")
|
||||
|
||||
def _uni(s, e):
|
||||
# hack for py3
|
||||
if sys.version_info[0] > 2:
|
||||
return unicode(bytes(s, sys.getdefaultencoding()), e)
|
||||
else:
|
||||
return unicode(s, e)
|
||||
|
||||
class TestNewIssues(base.PyMySQLTestCase):
|
||||
def test_issue_34(self):
|
||||
try:
|
||||
pymysql.connect(host="localhost", port=1237, user="root")
|
||||
self.fail()
|
||||
except pymysql.OperationalError, e:
|
||||
self.assertEqual(2003, e.args[0])
|
||||
except:
|
||||
self.fail()
|
||||
|
||||
def test_issue_33(self):
|
||||
conn = pymysql.connect(host="localhost", user="root", db=self.databases[0]["db"], charset="utf8")
|
||||
c = conn.cursor()
|
||||
try:
|
||||
c.execute(_uni("create table hei\xc3\x9fe (name varchar(32))", "utf8"))
|
||||
c.execute(_uni("insert into hei\xc3\x9fe (name) values ('Pi\xc3\xb1ata')", "utf8"))
|
||||
c.execute(_uni("select name from hei\xc3\x9fe", "utf8"))
|
||||
self.assertEqual(_uni("Pi\xc3\xb1ata","utf8"), c.fetchone()[0])
|
||||
finally:
|
||||
c.execute(_uni("drop table hei\xc3\x9fe", "utf8"))
|
||||
|
||||
# Will fail without manual intervention:
|
||||
#def test_issue_35(self):
|
||||
#
|
||||
# conn = self.connections[0]
|
||||
# c = conn.cursor()
|
||||
# print "sudo killall -9 mysqld within the next 10 seconds"
|
||||
# try:
|
||||
# c.execute("select sleep(10)")
|
||||
# self.fail()
|
||||
# except pymysql.OperationalError, e:
|
||||
# self.assertEqual(2013, e.args[0])
|
||||
|
||||
def test_issue_36(self):
|
||||
conn = self.connections[0]
|
||||
c = conn.cursor()
|
||||
# kill connections[0]
|
||||
original_count = c.execute("show processlist")
|
||||
kill_id = None
|
||||
for id,user,host,db,command,time,state,info in c.fetchall():
|
||||
if info == "show processlist":
|
||||
kill_id = id
|
||||
break
|
||||
# now nuke the connection
|
||||
conn.kill(kill_id)
|
||||
# make sure this connection has broken
|
||||
try:
|
||||
c.execute("show tables")
|
||||
self.fail()
|
||||
except:
|
||||
pass
|
||||
# check the process list from the other connection
|
||||
self.assertEqual(original_count - 1, self.connections[1].cursor().execute("show processlist"))
|
||||
del self.connections[0]
|
||||
|
||||
def test_issue_37(self):
|
||||
conn = self.connections[0]
|
||||
c = conn.cursor()
|
||||
self.assertEqual(1, c.execute("SELECT @foo"))
|
||||
self.assertEqual((None,), c.fetchone())
|
||||
self.assertEqual(0, c.execute("SET @foo = 'bar'"))
|
||||
c.execute("set @foo = 'bar'")
|
||||
|
||||
def test_issue_38(self):
|
||||
conn = self.connections[0]
|
||||
c = conn.cursor()
|
||||
datum = "a" * 1024 * 1023 # reduced size for most default mysql installs
|
||||
|
||||
try:
|
||||
c.execute("create table issue38 (id integer, data mediumblob)")
|
||||
c.execute("insert into issue38 values (1, %s)", datum)
|
||||
finally:
|
||||
c.execute("drop table issue38")
|
||||
__all__ = ["TestOldIssues", "TestNewIssues"]
|
||||
|
||||
if __name__ == "__main__":
|
||||
import unittest
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user