dinislage/web2py
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

upgraded to pymysql 0.2

Browse Source
This commit is contained in:
mdipierro
2012-07-12 15:31:24 -05:00
parent 0616d22aec
commit 2780f6f032
18 changed files with 809 additions and 192 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gluon/contrib/pymysql/tests/__init__.py
+9 -3
View File
@@ -1,6 +1,12 @@
from gluon.contrib.pymysql.tests.test_issues import *
from gluon.contrib.pymysql.tests.test_example import *
from gluon.contrib.pymysql.tests.test_basic import *
from pymysql.tests.test_issues import *
from pymysql.tests.test_example import *
from pymysql.tests.test_basic import *
from pymysql.tests.test_DictCursor import *
import sys
if sys.version_info[0] == 2:
# MySQLdb tests were designed for Python 3
from pymysql.tests.thirdparty import *
if __name__ == "__main__":
import unittest
gluon/contrib/pymysql/tests/base.py
+5 -7
View File
@@ -1,20 +1,18 @@
from gluon.contrib import pymysql
import pymysql
import unittest
class PyMySQLTestCase(unittest.TestCase):
# Edit this to suit your test environment.
databases = [
{"host":"localhost","user":"root",
"passwd":"","db":"test_pymysql", "use_unicode": True},
{"host":"localhost","user":"root","passwd":"","db":"test_pymysql2"}]
def setUp(self):
try:
self.connections = []
self.connections = []
for params in self.databases:
self.connections.append(pymysql.connect(**params))
except pymysql.err.OperationalError as e:
self.skipTest('Cannot connect to MySQL - skipping pymysql tests because of (%s) %s' % (type(e), e))
for params in self.databases:
self.connections.append(pymysql.connect(**params))
def tearDown(self):
for connection in self.connections:
gluon/contrib/pymysql/tests/test_DictCursor.py
Executable
+56
View File
@@ -0,0 +1,56 @@
from pymysql.tests import base
import pymysql.cursors
import datetime
class TestDictCursor(base.PyMySQLTestCase):
def test_DictCursor(self):
#all assert test compare to the structure as would come out from MySQLdb
conn = self.connections[0]
c = conn.cursor(pymysql.cursors.DictCursor)
# create a table ane some data to query
c.execute("""CREATE TABLE dictcursor (name char(20), age int , DOB datetime)""")
data = (("bob",21,"1990-02-06 23:04:56"),
("jim",56,"1955-05-09 13:12:45"),
("fred",100,"1911-09-12 01:01:01"))
bob = {'name':'bob','age':21,'DOB':datetime.datetime(1990, 02, 6, 23, 04, 56)}
jim = {'name':'jim','age':56,'DOB':datetime.datetime(1955, 05, 9, 13, 12, 45)}
fred = {'name':'fred','age':100,'DOB':datetime.datetime(1911, 9, 12, 1, 1, 1)}
try:
c.executemany("insert into dictcursor values (%s,%s,%s)", data)
# try an update which should return no rows
c.execute("update dictcursor set age=20 where name='bob'")
bob['age'] = 20
# pull back the single row dict for bob and check
c.execute("SELECT * from dictcursor where name='bob'")
r = c.fetchone()
self.assertEqual(bob,r,"fetchone via DictCursor failed")
# same again, but via fetchall => tuple)
c.execute("SELECT * from dictcursor where name='bob'")
r = c.fetchall()
self.assertEqual((bob,),r,"fetch a 1 row result via fetchall failed via DictCursor")
# same test again but iterate over the
c.execute("SELECT * from dictcursor where name='bob'")
for r in c:
self.assertEqual(bob, r,"fetch a 1 row result via iteration failed via DictCursor")
# get all 3 row via fetchall
c.execute("SELECT * from dictcursor")
r = c.fetchall()
self.assertEqual((bob,jim,fred), r, "fetchall failed via DictCursor")
#same test again but do a list comprehension
c.execute("SELECT * from dictcursor")
r = [x for x in c]
self.assertEqual([bob,jim,fred], r, "list comprehension failed via DictCursor")
# get all 2 row via fetchmany
c.execute("SELECT * from dictcursor")
r = c.fetchmany(2)
self.assertEqual((bob,jim), r, "fetchmany failed via DictCursor")
finally:
c.execute("drop table dictcursor")
__all__ = ["TestDictCursor"]
if __name__ == "__main__":
import unittest
unittest.main()
gluon/contrib/pymysql/tests/test_SSCursor.py
Executable
+100
View File
@@ -0,0 +1,100 @@
import sys
try:
from pymysql.tests import base
import pymysql.cursors
except:
# For local testing from top-level directory, without installing
sys.path.append('../pymysql')
from pymysql.tests import base
import pymysql.cursors
class TestSSCursor(base.PyMySQLTestCase):
def test_SSCursor(self):
affected_rows = 18446744073709551615
conn = self.connections[0]
data = [
('America', '', 'America/Jamaica'),
('America', '', 'America/Los_Angeles'),
('America', '', 'America/Lima'),
('America', '', 'America/New_York'),
('America', '', 'America/Menominee'),
('America', '', 'America/Havana'),
('America', '', 'America/El_Salvador'),
('America', '', 'America/Costa_Rica'),
('America', '', 'America/Denver'),
('America', '', 'America/Detroit'),]
try:
cursor = conn.cursor(pymysql.cursors.SSCursor)
# Create table
cursor.execute(('CREATE TABLE tz_data ('
'region VARCHAR(64),'
'zone VARCHAR(64),'
'name VARCHAR(64))'))
# Test INSERT
for i in data:
cursor.execute('INSERT INTO tz_data VALUES (%s, %s, %s)', i)
self.assertEqual(conn.affected_rows(), 1, 'affected_rows does not match')
conn.commit()
# Test fetchone()
iter = 0
cursor.execute('SELECT * FROM tz_data')
while True:
row = cursor.fetchone()
if row is None:
break
iter += 1
# Test cursor.rowcount
self.assertEqual(cursor.rowcount, affected_rows,
'cursor.rowcount != %s' % (str(affected_rows)))
# Test cursor.rownumber
self.assertEqual(cursor.rownumber, iter,
'cursor.rowcount != %s' % (str(iter)))
# Test row came out the same as it went in
self.assertEqual((row in data), True,
'Row not found in source data')
# Test fetchall
cursor.execute('SELECT * FROM tz_data')
self.assertEqual(len(cursor.fetchall()), len(data),
'fetchall failed. Number of rows does not match')
# Test fetchmany
cursor.execute('SELECT * FROM tz_data')
self.assertEqual(len(cursor.fetchmany(2)), 2,
'fetchmany failed. Number of rows does not match')
# So MySQLdb won't throw "Commands out of sync"
while True:
res = cursor.fetchone()
if res is None:
break
# Test update, affected_rows()
cursor.execute('UPDATE tz_data SET zone = %s', ['Foo'])
conn.commit()
self.assertEqual(cursor.rowcount, len(data),
'Update failed. affected_rows != %s' % (str(len(data))))
# Test executemany
cursor.executemany('INSERT INTO tz_data VALUES (%s, %s, %s)', data)
self.assertEqual(cursor.rowcount, len(data),
'executemany failed. cursor.rowcount != %s' % (str(len(data))))
finally:
cursor.execute('DROP TABLE tz_data')
cursor.close()
__all__ = ["TestSSCursor"]
if __name__ == "__main__":
import unittest
unittest.main()
gluon/contrib/pymysql/tests/test_basic.py
+74 -2
View File
@@ -1,5 +1,5 @@
from gluon.contrib.pymysql.tests import base
from gluon.contrib.pymysql import util
from pymysql.tests import base
from pymysql import util
import time
import datetime
@@ -55,6 +55,31 @@ class TestConversion(base.PyMySQLTestCase):
finally:
c.execute("drop table test_dict")
def test_string(self):
conn = self.connections[0]
c = conn.cursor()
c.execute("create table test_dict (a text)")
test_value = "I am a test string"
try:
c.execute("insert into test_dict (a) values (%s)", test_value)
c.execute("select a from test_dict")
self.assertEqual((test_value,), c.fetchone())
finally:
c.execute("drop table test_dict")
def test_integer(self):
conn = self.connections[0]
c = conn.cursor()
c.execute("create table test_dict (a integer)")
test_value = 12345
try:
c.execute("insert into test_dict (a) values (%s)", test_value)
c.execute("select a from test_dict")
self.assertEqual((test_value,), c.fetchone())
finally:
c.execute("drop table test_dict")
def test_big_blob(self):
""" test tons of data """
conn = self.connections[0]
@@ -67,6 +92,26 @@ class TestConversion(base.PyMySQLTestCase):
self.assertEqual(data.encode(conn.charset), c.fetchone()[0])
finally:
c.execute("drop table test_big_blob")
def test_untyped(self):
""" test conversion of null, empty string """
conn = self.connections[0]
c = conn.cursor()
c.execute("select null,''")
self.assertEqual((None,u''), c.fetchone())
c.execute("select '',null")
self.assertEqual((u'',None), c.fetchone())
def test_datetime(self):
""" test conversion of null, empty string """
conn = self.connections[0]
c = conn.cursor()
c.execute("select time('12:30'), time('23:12:59'), time('23:12:59.05100')")
self.assertEqual((datetime.timedelta(0, 45000),
datetime.timedelta(0, 83579),
datetime.timedelta(0, 83579, 51000)),
c.fetchone())
class TestCursor(base.PyMySQLTestCase):
# this test case does not work quite right yet, however,
@@ -134,6 +179,33 @@ class TestCursor(base.PyMySQLTestCase):
finally:
c.execute("drop table test_nr")
def test_aggregates(self):
""" test aggregate functions """
conn = self.connections[0]
c = conn.cursor()
try:
c.execute('create table test_aggregates (i integer)')
for i in xrange(0, 10):
c.execute('insert into test_aggregates (i) values (%s)', (i,))
c.execute('select sum(i) from test_aggregates')
r, = c.fetchone()
self.assertEqual(sum(range(0,10)), r)
finally:
c.execute('drop table test_aggregates')
def test_single_tuple(self):
""" test a single tuple """
conn = self.connections[0]
c = conn.cursor()
try:
c.execute("create table mystuff (id integer primary key)")
c.execute("insert into mystuff (id) values (1)")
c.execute("insert into mystuff (id) values (2)")
c.execute("select id from mystuff where id in %s", ((1,),))
self.assertEqual([(1,)], list(c.fetchall()))
finally:
c.execute("drop table mystuff")
__all__ = ["TestConversion","TestCursor"]
if __name__ == "__main__":
gluon/contrib/pymysql/tests/test_example.py
+2 -2
View File
@@ -1,5 +1,5 @@
from gluon.contrib import pymysql
from gluon.contrib.pymysql.tests import base
import pymysql
from pymysql.tests import base
class TestExample(base.PyMySQLTestCase):
def test_example(self):
gluon/contrib/pymysql/tests/test_issues.py
+60 -22
View File
@@ -1,5 +1,6 @@
from gluon.contrib import pymysql
from gluon.contrib.pymysql.tests import base
import pymysql
from pymysql.tests import base
import unittest
import sys
@@ -11,6 +12,10 @@ except AttributeError:
import datetime
# backwards compatibility:
if not hasattr(unittest, "skip"):
unittest.skip = lambda message: lambda f: f
class TestOldIssues(base.PyMySQLTestCase):
def test_issue_3(self):
""" undefined methods datetime_or_None, date_or_None """
@@ -89,15 +94,15 @@ KEY (`station`,`dh`,`echeance`)) ENGINE=MyISAM DEFAULT CHARSET=latin1;""")
""" can't handle large result fields """
conn = self.connections[0]
cur = conn.cursor()
cur.execute("create table issue13 (t text)")
try:
cur.execute("create table issue13 (t text)")
# 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
# use assertTrue so that obscenely huge error messages don't print
r = cur.fetchone()[0]
self.assert_("x" * size == r)
self.assertTrue("x" * size == r)
finally:
cur.execute("drop table issue13")
@@ -115,7 +120,7 @@ KEY (`station`,`dh`,`echeance`)) ENGINE=MyISAM DEFAULT CHARSET=latin1;""")
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("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:
@@ -133,6 +138,7 @@ KEY (`station`,`dh`,`echeance`)) ENGINE=MyISAM DEFAULT CHARSET=latin1;""")
finally:
c.execute("drop table issue16")
@unittest.skip("test_issue_17() requires a custom, legacy MySQL configuration and will not be run.")
def test_issue_17(self):
""" could not connect mysql use passwod """
conn = self.connections[0]
@@ -181,23 +187,22 @@ class TestNewIssues(base.PyMySQLTestCase):
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])
@unittest.skip("This test requires 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")
c.execute("show processlist")
kill_id = None
for id,user,host,db,command,time,state,info in c.fetchall():
if info == "show processlist":
@@ -212,8 +217,13 @@ class TestNewIssues(base.PyMySQLTestCase):
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]
try:
c = self.connections[1].cursor()
c.execute("show processlist")
ids = [row[0] for row in c.fetchall()]
self.assertFalse(kill_id in ids)
finally:
del self.connections[0]
def test_issue_37(self):
conn = self.connections[0]
@@ -230,10 +240,38 @@ class TestNewIssues(base.PyMySQLTestCase):
try:
c.execute("create table issue38 (id integer, data mediumblob)")
c.execute("insert into issue38 values (1, %s)", datum)
c.execute("insert into issue38 values (1, %s)", (datum,))
finally:
c.execute("drop table issue38")
__all__ = ["TestOldIssues", "TestNewIssues"]
def disabled_test_issue_54(self):
conn = self.connections[0]
c = conn.cursor()
big_sql = "select * from issue54 where "
big_sql += " and ".join("%d=%d" % (i,i) for i in xrange(0, 100000))
try:
c.execute("create table issue54 (id integer primary key)")
c.execute("insert into issue54 (id) values (7)")
c.execute(big_sql)
self.assertEqual(7, c.fetchone()[0])
finally:
c.execute("drop table issue54")
class TestGitHubIssues(base.PyMySQLTestCase):
def test_issue_66(self):
conn = self.connections[0]
c = conn.cursor()
self.assertEqual(0, conn.insert_id())
try:
c.execute("create table issue66 (id integer primary key auto_increment, x integer)")
c.execute("insert into issue66 (x) values (1)")
c.execute("insert into issue66 (x) values (1)")
self.assertEqual(2, conn.insert_id())
finally:
c.execute("drop table issue66")
__all__ = ["TestOldIssues", "TestNewIssues", "TestGitHubIssues"]
if __name__ == "__main__":
import unittest