python/mysqlclient
1
0
Fork 0
mirror of https://github.com/PyMySQL/mysqlclient.git synced 2026-03-13 08:00:02 +08:00
Code Issues Packages Projects Releases Wiki Activity

Unify test connection configuration

Browse Source
This commit is contained in:
farcepest
2012-10-02 12:37:41 -04:00
parent 4af9b70b91
commit d0e96c7155
5 changed files with 620 additions and 588 deletions
Download Patch File Download Diff File Expand all files Collapse all files

561
MySQLdb/tests/capabilities.py
View File

@@ -1,280 +1,281 @@
#!/usr/bin/env python -O #!/usr/bin/env python -O
""" Script to test database capabilities and the DB-API interface """ Script to test database capabilities and the DB-API interface
for functionality and memory leaks. for functionality and memory leaks.
Adapted from a script by M-A Lemburg. Adapted from a script by M-A Lemburg.
""" """
from time import time from time import time
import array import array
import unittest import unittest
from configdb import connection_factory
class DatabaseTest(unittest.TestCase):
class DatabaseTest(unittest.TestCase):
db_module = None
connect_args = () db_module = None
connect_kwargs = dict() connect_args = ()
create_table_extra = '' connect_kwargs = dict()
rows = 10 create_table_extra = ''
debug = False rows = 10
debug = False
def setUp(self):
import gc def setUp(self):
db = self.db_module.connect(*self.connect_args, **self.connect_kwargs) import gc
self.connection = db db = connection_factory(**self.connect_kwargs)
self.cursor = db.cursor() self.connection = db
# TODO: this needs to be re-evaluated for Python 3 self.cursor = db.cursor()
self.BLOBText = ''.join([chr(i) for i in range(256)] * 100); # TODO: this needs to be re-evaluated for Python 3
self.BLOBUText = u''.join([unichr(i) for i in range(16384)]) self.BLOBText = ''.join([chr(i) for i in range(256)] * 100);
self.BLOBBinary = self.db_module.Binary(''.join([chr(i) for i in range(256)] * 16)) self.BLOBUText = u''.join([unichr(i) for i in range(16384)])
self.BLOBBinary = self.db_module.Binary(''.join([chr(i) for i in range(256)] * 16))
leak_test = True
leak_test = True
def tearDown(self):
if self.leak_test: def tearDown(self):
import gc if self.leak_test:
del self.cursor import gc
orphans = gc.collect() del self.cursor
self.failIf(orphans, "%d orphaned objects found after deleting cursor" % orphans) orphans = gc.collect()
self.failIf(orphans, "%d orphaned objects found after deleting cursor" % orphans)
del self.connection
orphans = gc.collect() del self.connection
self.failIf(orphans, "%d orphaned objects found after deleting connection" % orphans) orphans = gc.collect()
self.failIf(orphans, "%d orphaned objects found after deleting connection" % orphans)
def table_exists(self, name):
try: def table_exists(self, name):
self.cursor.execute('select * from %s where 1=0' % name) try:
except: self.cursor.execute('select * from %s where 1=0' % name)
return False except:
else: return False
return True else:
return True
def quote_identifier(self, ident):
return '"%s"' % ident def quote_identifier(self, ident):
return '"%s"' % ident
def new_table_name(self):
i = id(self.cursor) def new_table_name(self):
while True: i = id(self.cursor)
name = self.quote_identifier('tb%08x' % i) while True:
if not self.table_exists(name): name = self.quote_identifier('tb%08x' % i)
return name if not self.table_exists(name):
i = i + 1 return name
i = i + 1
def create_table(self, columndefs):
def create_table(self, columndefs):
""" Create a table using a list of column definitions given in
columndefs. """ Create a table using a list of column definitions given in
columndefs.
generator must be a function taking arguments (row_number,
col_number) returning a suitable data object for insertion generator must be a function taking arguments (row_number,
into the table. col_number) returning a suitable data object for insertion
into the table.
"""
self.table = self.new_table_name() """
self.cursor.execute('CREATE TABLE %s (%s) %s' % self.table = self.new_table_name()
(self.table, self.cursor.execute('CREATE TABLE %s (%s) %s' %
',\n'.join(columndefs), (self.table,
self.create_table_extra)) ',\n'.join(columndefs),
self.create_table_extra))
def check_data_integrity(self, columndefs, generator):
# insert def check_data_integrity(self, columndefs, generator):
self.create_table(columndefs) # insert
insert_statement = ('INSERT INTO %s VALUES (%s)' % self.create_table(columndefs)
(self.table, insert_statement = ('INSERT INTO %s VALUES (%s)' %
','.join(['%s'] * len(columndefs)))) (self.table,
data = [ [ generator(i,j) for j in range(len(columndefs)) ] ','.join(['%s'] * len(columndefs))))
for i in range(self.rows) ] data = [ [ generator(i,j) for j in range(len(columndefs)) ]
self.cursor.executemany(insert_statement, data) for i in range(self.rows) ]
self.connection.commit() self.cursor.executemany(insert_statement, data)
# verify self.connection.commit()
self.cursor.execute('select * from %s' % self.table) # verify
l = self.cursor.fetchall() self.cursor.execute('select * from %s' % self.table)
self.assertEquals(len(l), self.rows) l = self.cursor.fetchall()
try: self.assertEquals(len(l), self.rows)
for i in range(self.rows): try:
for j in range(len(columndefs)): for i in range(self.rows):
self.assertEquals(l[i][j], generator(i,j)) for j in range(len(columndefs)):
finally: self.assertEquals(l[i][j], generator(i,j))
if not self.debug: finally:
self.cursor.execute('drop table %s' % (self.table)) if not self.debug:
self.cursor.execute('drop table %s' % (self.table))
def test_transactions(self):
columndefs = ( 'col1 INT', 'col2 VARCHAR(255)') def test_transactions(self):
def generator(row, col): columndefs = ( 'col1 INT', 'col2 VARCHAR(255)')
if col == 0: return row def generator(row, col):
else: return ('%i' % (row%10))*255 if col == 0: return row
self.create_table(columndefs) else: return ('%i' % (row%10))*255
insert_statement = ('INSERT INTO %s VALUES (%s)' % self.create_table(columndefs)
(self.table, insert_statement = ('INSERT INTO %s VALUES (%s)' %
','.join(['%s'] * len(columndefs)))) (self.table,
data = [ [ generator(i,j) for j in range(len(columndefs)) ] ','.join(['%s'] * len(columndefs))))
for i in range(self.rows) ] data = [ [ generator(i,j) for j in range(len(columndefs)) ]
self.cursor.executemany(insert_statement, data) for i in range(self.rows) ]
# verify self.cursor.executemany(insert_statement, data)
self.connection.commit() # verify
self.cursor.execute('select * from %s' % self.table) self.connection.commit()
l = self.cursor.fetchall() self.cursor.execute('select * from %s' % self.table)
self.assertEquals(len(l), self.rows) l = self.cursor.fetchall()
for i in range(self.rows): self.assertEquals(len(l), self.rows)
for j in range(len(columndefs)): for i in range(self.rows):
self.assertEquals(l[i][j], generator(i,j)) for j in range(len(columndefs)):
delete_statement = 'delete from %s where col1=%%s' % self.table self.assertEquals(l[i][j], generator(i,j))
self.cursor.execute(delete_statement, (0,)) delete_statement = 'delete from %s where col1=%%s' % self.table
self.cursor.execute('select col1 from %s where col1=%s' % \ self.cursor.execute(delete_statement, (0,))
(self.table, 0)) self.cursor.execute('select col1 from %s where col1=%s' % \
l = self.cursor.fetchall() (self.table, 0))
self.assertFalse(l, "DELETE didn't work") l = self.cursor.fetchall()
self.connection.rollback() self.assertFalse(l, "DELETE didn't work")
self.cursor.execute('select col1 from %s where col1=%s' % \ self.connection.rollback()
(self.table, 0)) self.cursor.execute('select col1 from %s where col1=%s' % \
l = self.cursor.fetchall() (self.table, 0))
self.assertTrue(len(l) == 1, "ROLLBACK didn't work") l = self.cursor.fetchall()
self.cursor.execute('drop table %s' % (self.table)) self.assertTrue(len(l) == 1, "ROLLBACK didn't work")
self.cursor.execute('drop table %s' % (self.table))
def test_truncation(self):
columndefs = ( 'col1 INT', 'col2 VARCHAR(255)') def test_truncation(self):
def generator(row, col): columndefs = ( 'col1 INT', 'col2 VARCHAR(255)')
if col == 0: return row def generator(row, col):
else: return ('%i' % (row%10))*((255-self.rows/2)+row) if col == 0: return row
self.create_table(columndefs) else: return ('%i' % (row%10))*((255-self.rows/2)+row)
insert_statement = ('INSERT INTO %s VALUES (%s)' % self.create_table(columndefs)
(self.table, insert_statement = ('INSERT INTO %s VALUES (%s)' %
','.join(['%s'] * len(columndefs)))) (self.table,
','.join(['%s'] * len(columndefs))))
try:
self.cursor.execute(insert_statement, (0, '0'*256)) try:
except self.connection.DataError: self.cursor.execute(insert_statement, (0, '0'*256))
pass except self.connection.DataError:
else: pass
self.fail("Over-long column did not generate warnings/exception with single insert") else:
self.fail("Over-long column did not generate warnings/exception with single insert")
self.connection.rollback()
self.connection.rollback()
try:
for i in range(self.rows): try:
data = [] for i in range(self.rows):
for j in range(len(columndefs)): data = []
data.append(generator(i,j)) for j in range(len(columndefs)):
self.cursor.execute(insert_statement,tuple(data)) data.append(generator(i,j))
except self.connection.DataError: self.cursor.execute(insert_statement,tuple(data))
pass except self.connection.DataError:
else: pass
self.fail("Over-long columns did not generate warnings/exception with execute()") else:
self.fail("Over-long columns did not generate warnings/exception with execute()")
self.connection.rollback()
self.connection.rollback()
try:
data = [ [ generator(i,j) for j in range(len(columndefs)) ] try:
for i in range(self.rows) ] data = [ [ generator(i,j) for j in range(len(columndefs)) ]
self.cursor.executemany(insert_statement, data) for i in range(self.rows) ]
except self.connection.DataError: self.cursor.executemany(insert_statement, data)
pass except self.connection.DataError:
else: pass
self.fail("Over-long columns did not generate warnings/exception with executemany()") else:
self.fail("Over-long columns did not generate warnings/exception with executemany()")
self.connection.rollback()
self.cursor.execute('drop table %s' % (self.table)) self.connection.rollback()
self.cursor.execute('drop table %s' % (self.table))
def test_CHAR(self):
# Character data def test_CHAR(self):
def generator(row,col): # Character data
return ('%i' % ((row+col) % 10)) * 255 def generator(row,col):
self.check_data_integrity( return ('%i' % ((row+col) % 10)) * 255
('col1 char(255)','col2 char(255)'), self.check_data_integrity(
generator) ('col1 char(255)','col2 char(255)'),
generator)
def test_INT(self):
# Number data def test_INT(self):
def generator(row,col): # Number data
return row*row def generator(row,col):
self.check_data_integrity( return row*row
('col1 INT',), self.check_data_integrity(
generator) ('col1 INT',),
generator)
def test_DECIMAL(self):
# DECIMAL def test_DECIMAL(self):
def generator(row,col): # DECIMAL
from decimal import Decimal def generator(row,col):
return Decimal("%d.%02d" % (row, col)) from decimal import Decimal
self.check_data_integrity( return Decimal("%d.%02d" % (row, col))
('col1 DECIMAL(5,2)',), self.check_data_integrity(
generator) ('col1 DECIMAL(5,2)',),
generator)
def test_DATE(self):
ticks = time() def test_DATE(self):
def generator(row,col): ticks = time()
return self.db_module.DateFromTicks(ticks+row*86400-col*1313) def generator(row,col):
self.check_data_integrity( return self.db_module.DateFromTicks(ticks+row*86400-col*1313)
('col1 DATE',), self.check_data_integrity(
generator) ('col1 DATE',),
generator)
def test_TIME(self):
ticks = time() def test_TIME(self):
def generator(row,col): ticks = time()
return self.db_module.TimeFromTicks(ticks+row*86400-col*1313) def generator(row,col):
self.check_data_integrity( return self.db_module.TimeFromTicks(ticks+row*86400-col*1313)
('col1 TIME',), self.check_data_integrity(
generator) ('col1 TIME',),
generator)
def test_DATETIME(self):
ticks = time() def test_DATETIME(self):
def generator(row,col): ticks = time()
return self.db_module.TimestampFromTicks(ticks+row*86400-col*1313) def generator(row,col):
self.check_data_integrity( return self.db_module.TimestampFromTicks(ticks+row*86400-col*1313)
('col1 DATETIME',), self.check_data_integrity(
generator) ('col1 DATETIME',),
generator)
def test_TIMESTAMP(self):
ticks = time() def test_TIMESTAMP(self):
def generator(row,col): ticks = time()
return self.db_module.TimestampFromTicks(ticks+row*86400-col*1313) def generator(row,col):
self.check_data_integrity( return self.db_module.TimestampFromTicks(ticks+row*86400-col*1313)
('col1 TIMESTAMP',), self.check_data_integrity(
generator) ('col1 TIMESTAMP',),
generator)
def test_fractional_TIMESTAMP(self):
ticks = time() def test_fractional_TIMESTAMP(self):
def generator(row,col): ticks = time()
return self.db_module.TimestampFromTicks(ticks+row*86400-col*1313+row*0.7*col/3.0) def generator(row,col):
self.check_data_integrity( return self.db_module.TimestampFromTicks(ticks+row*86400-col*1313+row*0.7*col/3.0)
('col1 TIMESTAMP',), self.check_data_integrity(
generator) ('col1 TIMESTAMP',),
generator)
def test_LONG(self):
def generator(row,col): def test_LONG(self):
if col == 0: def generator(row,col):
return row if col == 0:
else: return row
return self.BLOBUText # 'BLOB Text ' * 1024 else:
self.check_data_integrity( return self.BLOBUText # 'BLOB Text ' * 1024
('col1 INT','col2 LONG'), self.check_data_integrity(
generator) ('col1 INT','col2 LONG'),
generator)
def test_TEXT(self):
def generator(row,col): def test_TEXT(self):
return self.BLOBUText # 'BLOB Text ' * 1024 def generator(row,col):
self.check_data_integrity( return self.BLOBUText # 'BLOB Text ' * 1024
('col2 TEXT',), self.check_data_integrity(
generator) ('col2 TEXT',),
generator)
def test_LONG_BYTE(self):
def generator(row,col): def test_LONG_BYTE(self):
if col == 0: def generator(row,col):
return row if col == 0:
else: return row
return self.BLOBBinary # 'BLOB\000Binary ' * 1024 else:
self.check_data_integrity( return self.BLOBBinary # 'BLOB\000Binary ' * 1024
('col1 INT','col2 LONG BYTE'), self.check_data_integrity(
generator) ('col1 INT','col2 LONG BYTE'),
generator)
def test_BLOB(self):
def generator(row,col): def test_BLOB(self):
if col == 0: def generator(row,col):
return row if col == 0:
else: return row
return self.BLOBBinary # 'BLOB\000Binary ' * 1024 else:
self.check_data_integrity( return self.BLOBBinary # 'BLOB\000Binary ' * 1024
('col1 INT','col2 BLOB'), self.check_data_integrity(
generator) ('col1 INT','col2 BLOB'),
generator)

25
MySQLdb/tests/configdb.py Normal file
View File

@@ -0,0 +1,25 @@
"""Configure database connection for tests."""
from os import environ, path
tests_path = path.dirname(__file__)
conf_file = environ.get('TESTDB', 'default.cnf')
conf_path = path.join(tests_path, conf_file)
connect_kwargs = dict(
read_default_file = conf_path,
read_default_group = "MySQLdb-tests",
)
def connection_kwargs(kwargs):
db_kwargs = connect_kwargs.copy()
db_kwargs.update(kwargs)
return db_kwargs
def connection_factory(**kwargs):
import MySQLdb
db_kwargs = connection_kwargs(kwargs)
db = MySQLdb.connect(**db_kwargs)
return db

10
MySQLdb/tests/default.cnf Normal file
View File

@@ -0,0 +1,10 @@
# To create your own custom version of this file, read
# http://dev.mysql.com/doc/refman/5.1/en/option-files.html
# and set TESTDB in your environment to the name of the file
[MySQLdb-tests]
host = 127.0.0.1
user = test
database = test
#password =
default-character-set = utf8

205
MySQLdb/tests/test_MySQLdb_capabilities.py
View File

@@ -1,103 +1,102 @@
#!/usr/bin/env python #!/usr/bin/env python
import capabilities import capabilities
import unittest import unittest
import MySQLdb import MySQLdb
import warnings import warnings
warnings.filterwarnings('error') warnings.filterwarnings('error')
class test_MySQLdb(capabilities.DatabaseTest): class test_MySQLdb(capabilities.DatabaseTest):
db_module = MySQLdb db_module = MySQLdb
connect_args = () connect_args = ()
connect_kwargs = dict(db='test', host="127.0.0.1", user="test", #read_default_file='~/.my.cnf', connect_kwargs = dict(use_unicode=True, sql_mode="ANSI,STRICT_TRANS_TABLES,TRADITIONAL")
charset='utf8', sql_mode="ANSI,STRICT_TRANS_TABLES,TRADITIONAL") create_table_extra = "ENGINE=INNODB CHARACTER SET UTF8"
create_table_extra = "ENGINE=INNODB CHARACTER SET UTF8" leak_test = False
leak_test = False
def quote_identifier(self, ident):
def quote_identifier(self, ident): return "`%s`" % ident
return "`%s`" % ident
def test_TIME(self):
def test_TIME(self): from datetime import timedelta
from datetime import timedelta def generator(row,col):
def generator(row,col): return timedelta(0, row*8000)
return timedelta(0, row*8000) self.check_data_integrity(
self.check_data_integrity( ('col1 TIME',),
('col1 TIME',), generator)
generator)
def test_TINYINT(self):
def test_TINYINT(self): # Number data
# Number data def generator(row,col):
def generator(row,col): v = (row*row) % 256
v = (row*row) % 256 if v > 127:
if v > 127: v = v-256
v = v-256 return v
return v self.check_data_integrity(
self.check_data_integrity( ('col1 TINYINT',),
('col1 TINYINT',), generator)
generator)
def test_stored_procedures(self):
def test_stored_procedures(self): db = self.connection
db = self.connection c = self.cursor
c = self.cursor self.create_table(('pos INT', 'tree CHAR(20)'))
self.create_table(('pos INT', 'tree CHAR(20)')) c.executemany("INSERT INTO %s (pos,tree) VALUES (%%s,%%s)" % self.table,
c.executemany("INSERT INTO %s (pos,tree) VALUES (%%s,%%s)" % self.table, list(enumerate('ash birch cedar larch pine'.split())))
list(enumerate('ash birch cedar larch pine'.split()))) db.commit()
db.commit()
c.execute("""
c.execute(""" CREATE PROCEDURE test_sp(IN t VARCHAR(255))
CREATE PROCEDURE test_sp(IN t VARCHAR(255)) BEGIN
BEGIN SELECT pos FROM %s WHERE tree = t;
SELECT pos FROM %s WHERE tree = t; END
END """ % self.table)
""" % self.table) db.commit()
db.commit()
c.callproc('test_sp', ('larch',))
c.callproc('test_sp', ('larch',)) rows = c.fetchall()
rows = c.fetchall() self.assertEquals(len(rows), 1)
self.assertEquals(len(rows), 1) self.assertEquals(rows[0][0], 3)
self.assertEquals(rows[0][0], 3) c.nextset()
c.nextset()
c.execute("DROP PROCEDURE test_sp")
c.execute("DROP PROCEDURE test_sp") c.execute('drop table %s' % (self.table))
c.execute('drop table %s' % (self.table))
def test_small_CHAR(self):
def test_small_CHAR(self): # Character data
# Character data def generator(row,col):
def generator(row,col): i = (row*col+62)%256
i = (row*col+62)%256 if i == 62: return ''
if i == 62: return '' if i == 63: return None
if i == 63: return None return chr(i)
return chr(i) self.check_data_integrity(
self.check_data_integrity( ('col1 char(1)','col2 char(1)'),
('col1 char(1)','col2 char(1)'), generator)
generator)
def test_bug_2671682(self):
def test_bug_2671682(self): from MySQLdb.constants import ER
from MySQLdb.constants import ER try:
try: self.cursor.execute("describe some_non_existent_table");
self.cursor.execute("describe some_non_existent_table"); except self.connection.ProgrammingError, msg:
except self.connection.ProgrammingError, msg: self.assertTrue(msg[0] == ER.NO_SUCH_TABLE)
self.assertTrue(msg[0] == ER.NO_SUCH_TABLE)
def test_bug_3514287(self):
def test_bug_3514287(self): c = self.cursor
c = self.cursor try:
try: c.execute("""create table bug_3541287 (
c.execute("""create table bug_3541287 ( c1 CHAR(10),
c1 CHAR(10), t1 TIMESTAMP)""")
t1 TIMESTAMP)""") c.execute("insert into bug_3541287 (c1,t1) values (%s, NOW())",
c.execute("insert into bug_3541287 (c1,t1) values (%s, NOW())", ("blah",))
("blah",)) finally:
finally: c.execute("drop table if exists bug_3541287")
c.execute("drop table if exists bug_3541287")
def test_ping(self):
def test_ping(self): self.connection.ping()
self.connection.ping()
if __name__ == '__main__':
if __name__ == '__main__': if test_MySQLdb.leak_test:
if test_MySQLdb.leak_test: import gc
import gc gc.enable()
gc.enable() gc.set_debug(gc.DEBUG_LEAK)
gc.set_debug(gc.DEBUG_LEAK) unittest.main()
unittest.main()

407
MySQLdb/tests/test_MySQLdb_dbapi20.py
View File

@@ -1,205 +1,202 @@
#!/usr/bin/env python #!/usr/bin/env python
import dbapi20 import dbapi20
import unittest import unittest
import MySQLdb import MySQLdb
from configdb import connection_kwargs
class test_MySQLdb(dbapi20.DatabaseAPI20Test):
driver = MySQLdb class test_MySQLdb(dbapi20.DatabaseAPI20Test):
connect_args = () driver = MySQLdb
connect_kw_args = dict(db='test', connect_args = ()
host="127.0.0.1", connect_kw_args = connection_kwargs(dict(sql_mode="ANSI,STRICT_TRANS_TABLES,TRADITIONAL"))
user="test", #read_default_file='~/.my.cnf',
charset='utf8', def test_setoutputsize(self): pass
sql_mode="ANSI,STRICT_TRANS_TABLES,TRADITIONAL") def test_setoutputsize_basic(self): pass
def test_nextset(self): pass
def test_setoutputsize(self): pass
def test_setoutputsize_basic(self): pass """The tests on fetchone and fetchall and rowcount bogusly
def test_nextset(self): pass test for an exception if the statement cannot return a
result set. MySQL always returns a result set; it's just that
"""The tests on fetchone and fetchall and rowcount bogusly some things return empty result sets."""
test for an exception if the statement cannot return a
result set. MySQL always returns a result set; it's just that def test_fetchall(self):
some things return empty result sets.""" con = self._connect()
try:
def test_fetchall(self): cur = con.cursor()
con = self._connect() # cursor.fetchall should raise an Error if called
try: # without executing a query that may return rows (such
cur = con.cursor() # as a select)
# cursor.fetchall should raise an Error if called self.assertRaises(self.driver.Error, cur.fetchall)
# without executing a query that may return rows (such
# as a select) self.executeDDL1(cur)
self.assertRaises(self.driver.Error, cur.fetchall) for sql in self._populate():
cur.execute(sql)
self.executeDDL1(cur)
for sql in self._populate(): # cursor.fetchall should raise an Error if called
cur.execute(sql) # after executing a a statement that cannot return rows
## self.assertRaises(self.driver.Error,cur.fetchall)
# cursor.fetchall should raise an Error if called
# after executing a a statement that cannot return rows cur.execute('select name from %sbooze' % self.table_prefix)
## self.assertRaises(self.driver.Error,cur.fetchall) rows = cur.fetchall()
self.assertTrue(cur.rowcount in (-1,len(self.samples)))
cur.execute('select name from %sbooze' % self.table_prefix) self.assertEqual(len(rows),len(self.samples),
rows = cur.fetchall() 'cursor.fetchall did not retrieve all rows'
self.assertTrue(cur.rowcount in (-1,len(self.samples))) )
self.assertEqual(len(rows),len(self.samples), rows = [r[0] for r in rows]
'cursor.fetchall did not retrieve all rows' rows.sort()
) for i in range(0,len(self.samples)):
rows = [r[0] for r in rows] self.assertEqual(rows[i],self.samples[i],
rows.sort() 'cursor.fetchall retrieved incorrect rows'
for i in range(0,len(self.samples)): )
self.assertEqual(rows[i],self.samples[i], rows = cur.fetchall()
'cursor.fetchall retrieved incorrect rows' self.assertEqual(
) len(rows),0,
rows = cur.fetchall() 'cursor.fetchall should return an empty list if called '
self.assertEqual( 'after the whole result set has been fetched'
len(rows),0, )
'cursor.fetchall should return an empty list if called ' self.assertTrue(cur.rowcount in (-1,len(self.samples)))
'after the whole result set has been fetched'
) self.executeDDL2(cur)
self.assertTrue(cur.rowcount in (-1,len(self.samples))) cur.execute('select name from %sbarflys' % self.table_prefix)
rows = cur.fetchall()
self.executeDDL2(cur) self.assertTrue(cur.rowcount in (-1,0))
cur.execute('select name from %sbarflys' % self.table_prefix) self.assertEqual(len(rows),0,
rows = cur.fetchall() 'cursor.fetchall should return an empty list if '
self.assertTrue(cur.rowcount in (-1,0)) 'a select query returns no rows'
self.assertEqual(len(rows),0, )
'cursor.fetchall should return an empty list if '
'a select query returns no rows' finally:
) con.close()
finally: def test_fetchone(self):
con.close() con = self._connect()
try:
def test_fetchone(self): cur = con.cursor()
con = self._connect()
try: # cursor.fetchone should raise an Error if called before
cur = con.cursor() # executing a select-type query
self.assertRaises(self.driver.Error,cur.fetchone)
# cursor.fetchone should raise an Error if called before
# executing a select-type query # cursor.fetchone should raise an Error if called after
self.assertRaises(self.driver.Error,cur.fetchone) # executing a query that cannnot return rows
self.executeDDL1(cur)
# cursor.fetchone should raise an Error if called after ## self.assertRaises(self.driver.Error,cur.fetchone)
# executing a query that cannnot return rows
self.executeDDL1(cur) cur.execute('select name from %sbooze' % self.table_prefix)
## self.assertRaises(self.driver.Error,cur.fetchone) self.assertEqual(cur.fetchone(),None,
'cursor.fetchone should return None if a query retrieves '
cur.execute('select name from %sbooze' % self.table_prefix) 'no rows'
self.assertEqual(cur.fetchone(),None, )
'cursor.fetchone should return None if a query retrieves ' self.assertTrue(cur.rowcount in (-1,0))
'no rows'
) # cursor.fetchone should raise an Error if called after
self.assertTrue(cur.rowcount in (-1,0)) # executing a query that cannnot return rows
cur.execute("insert into %sbooze values ('Victoria Bitter')" % (
# cursor.fetchone should raise an Error if called after self.table_prefix
# executing a query that cannnot return rows ))
cur.execute("insert into %sbooze values ('Victoria Bitter')" % ( ## self.assertRaises(self.driver.Error,cur.fetchone)
self.table_prefix
)) cur.execute('select name from %sbooze' % self.table_prefix)
## self.assertRaises(self.driver.Error,cur.fetchone) r = cur.fetchone()
self.assertEqual(len(r),1,
cur.execute('select name from %sbooze' % self.table_prefix) 'cursor.fetchone should have retrieved a single row'
r = cur.fetchone() )
self.assertEqual(len(r),1, self.assertEqual(r[0],'Victoria Bitter',
'cursor.fetchone should have retrieved a single row' 'cursor.fetchone retrieved incorrect data'
) )
self.assertEqual(r[0],'Victoria Bitter', ## self.assertEqual(cur.fetchone(),None,
'cursor.fetchone retrieved incorrect data' ## 'cursor.fetchone should return None if no more rows available'
) ## )
## self.assertEqual(cur.fetchone(),None, self.assertTrue(cur.rowcount in (-1,1))
## 'cursor.fetchone should return None if no more rows available' finally:
## ) con.close()
self.assertTrue(cur.rowcount in (-1,1))
finally: # Same complaint as for fetchall and fetchone
con.close() def test_rowcount(self):
con = self._connect()
# Same complaint as for fetchall and fetchone try:
def test_rowcount(self): cur = con.cursor()
con = self._connect() self.executeDDL1(cur)
try: ## self.assertEqual(cur.rowcount,-1,
cur = con.cursor() ## 'cursor.rowcount should be -1 after executing no-result '
self.executeDDL1(cur) ## 'statements'
## self.assertEqual(cur.rowcount,-1, ## )
## 'cursor.rowcount should be -1 after executing no-result ' cur.execute("insert into %sbooze values ('Victoria Bitter')" % (
## 'statements' self.table_prefix
## ) ))
cur.execute("insert into %sbooze values ('Victoria Bitter')" % ( ## self.assertTrue(cur.rowcount in (-1,1),
self.table_prefix ## 'cursor.rowcount should == number or rows inserted, or '
)) ## 'set to -1 after executing an insert statement'
## self.assertTrue(cur.rowcount in (-1,1), ## )
## 'cursor.rowcount should == number or rows inserted, or ' cur.execute("select name from %sbooze" % self.table_prefix)
## 'set to -1 after executing an insert statement' self.assertTrue(cur.rowcount in (-1,1),
## ) 'cursor.rowcount should == number of rows returned, or '
cur.execute("select name from %sbooze" % self.table_prefix) 'set to -1 after executing a select statement'
self.assertTrue(cur.rowcount in (-1,1), )
'cursor.rowcount should == number of rows returned, or ' self.executeDDL2(cur)
'set to -1 after executing a select statement' ## self.assertEqual(cur.rowcount,-1,
) ## 'cursor.rowcount not being reset to -1 after executing '
self.executeDDL2(cur) ## 'no-result statements'
## self.assertEqual(cur.rowcount,-1, ## )
## 'cursor.rowcount not being reset to -1 after executing ' finally:
## 'no-result statements' con.close()
## )
finally: def test_callproc(self):
con.close() pass # performed in test_MySQL_capabilities
def test_callproc(self): def help_nextset_setUp(self,cur):
pass # performed in test_MySQL_capabilities ''' Should create a procedure called deleteme
that returns two result sets, first the
def help_nextset_setUp(self,cur): number of rows in booze then "name from booze"
''' Should create a procedure called deleteme '''
that returns two result sets, first the sql="""
number of rows in booze then "name from booze" create procedure deleteme()
''' begin
sql=""" select count(*) from %(tp)sbooze;
create procedure deleteme() select name from %(tp)sbooze;
begin end
select count(*) from %(tp)sbooze; """ % dict(tp=self.table_prefix)
select name from %(tp)sbooze; cur.execute(sql)
end
""" % dict(tp=self.table_prefix) def help_nextset_tearDown(self,cur):
cur.execute(sql) 'If cleaning up is needed after nextSetTest'
cur.execute("drop procedure deleteme")
def help_nextset_tearDown(self,cur):
'If cleaning up is needed after nextSetTest' def test_nextset(self):
cur.execute("drop procedure deleteme") from warnings import warn
con = self._connect()
def test_nextset(self): try:
from warnings import warn cur = con.cursor()
con = self._connect() if not hasattr(cur,'nextset'):
try: return
cur = con.cursor()
if not hasattr(cur,'nextset'): try:
return self.executeDDL1(cur)
sql=self._populate()
try: for sql in self._populate():
self.executeDDL1(cur) cur.execute(sql)
sql=self._populate()
for sql in self._populate(): self.help_nextset_setUp(cur)
cur.execute(sql)
cur.callproc('deleteme')
self.help_nextset_setUp(cur) numberofrows=cur.fetchone()
assert numberofrows[0]== len(self.samples)
cur.callproc('deleteme') assert cur.nextset()
numberofrows=cur.fetchone() names=cur.fetchall()
assert numberofrows[0]== len(self.samples) assert len(names) == len(self.samples)
assert cur.nextset() s=cur.nextset()
names=cur.fetchall() if s:
assert len(names) == len(self.samples) empty = cur.fetchall()
s=cur.nextset() self.assertEquals(len(empty), 0,
if s: "non-empty result set after other result sets")
empty = cur.fetchall() #warn("Incompatibility: MySQL returns an empty result set for the CALL itself",
self.assertEquals(len(empty), 0, # Warning)
"non-empty result set after other result sets") #assert s == None,'No more return sets, should return None'
#warn("Incompatibility: MySQL returns an empty result set for the CALL itself", finally:
# Warning) self.help_nextset_tearDown(cur)
#assert s == None,'No more return sets, should return None'
finally: finally:
self.help_nextset_tearDown(cur) con.close()
finally:
con.close() if __name__ == '__main__':
unittest.main()
if __name__ == '__main__':
unittest.main()