python/mysqlclient
1
0
Fork 0
mirror of https://github.com/PyMySQL/mysqlclient.git synced 2025-08-14 18:12:35 +08:00
Code Issues Packages Projects Releases Wiki Activity

Format

Browse Source
This commit is contained in:
Inada Naoki
2018-12-06 19:33:31 +09:00
parent c754b25080
commit a33e1c3836
10 changed files with 585 additions and 599 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
MySQLdb/_mysql_exceptions.py
View File

@ -13,42 +13,35 @@ except ImportError:
class MySQLError(StandardError): class MySQLError(StandardError):
"""Exception related to operation with MySQL.""" """Exception related to operation with MySQL."""
class Warning(Warning, MySQLError): class Warning(Warning, MySQLError):
"""Exception raised for important warnings like data truncations """Exception raised for important warnings like data truncations
while inserting, etc.""" while inserting, etc."""
class Error(MySQLError): class Error(MySQLError):
"""Exception that is the base class of all other error exceptions """Exception that is the base class of all other error exceptions
(not Warning).""" (not Warning)."""
class InterfaceError(Error): class InterfaceError(Error):
"""Exception raised for errors that are related to the database """Exception raised for errors that are related to the database
interface rather than the database itself.""" interface rather than the database itself."""
class DatabaseError(Error): class DatabaseError(Error):
"""Exception raised for errors that are related to the """Exception raised for errors that are related to the
database.""" database."""
class DataError(DatabaseError): class DataError(DatabaseError):
"""Exception raised for errors that are due to problems with the """Exception raised for errors that are due to problems with the
processed data like division by zero, numeric value out of range, processed data like division by zero, numeric value out of range,
etc.""" etc."""
class OperationalError(DatabaseError): class OperationalError(DatabaseError):
"""Exception raised for errors that are related to the database's """Exception raised for errors that are related to the database's
operation and not necessarily under the control of the programmer, operation and not necessarily under the control of the programmer,
e.g. an unexpected disconnect occurs, the data source name is not e.g. an unexpected disconnect occurs, the data source name is not
@ -57,31 +50,25 @@ class OperationalError(DatabaseError):
class IntegrityError(DatabaseError): class IntegrityError(DatabaseError):
"""Exception raised when the relational integrity of the database """Exception raised when the relational integrity of the database
is affected, e.g. a foreign key check fails, duplicate key, is affected, e.g. a foreign key check fails, duplicate key,
etc.""" etc."""
class InternalError(DatabaseError): class InternalError(DatabaseError):
"""Exception raised when the database encounters an internal """Exception raised when the database encounters an internal
error, e.g. the cursor is not valid anymore, the transaction is error, e.g. the cursor is not valid anymore, the transaction is
out of sync, etc.""" out of sync, etc."""
class ProgrammingError(DatabaseError): class ProgrammingError(DatabaseError):
"""Exception raised for programming errors, e.g. table not found """Exception raised for programming errors, e.g. table not found
or already exists, syntax error in the SQL statement, wrong number or already exists, syntax error in the SQL statement, wrong number
of parameters specified, etc.""" of parameters specified, etc."""
class NotSupportedError(DatabaseError): class NotSupportedError(DatabaseError):
"""Exception raised in case a method or database API was used """Exception raised in case a method or database API was used
which is not supported by the database, e.g. requesting a which is not supported by the database, e.g. requesting a
.rollback() on a connection that does not support transaction or .rollback() on a connection that does not support transaction or
has transactions turned off.""" has transactions turned off."""

14
doc/FAQ.rst
View File

@ -26,7 +26,7 @@ probably the issue, but it shouldn't happen any more.
ImportError ImportError
----------- -----------
ImportError: No module named _mysql ImportError: No module named _mysql
If you see this, it's likely you did some wrong when installing If you see this, it's likely you did some wrong when installing
MySQLdb; re-read (or read) README. _mysql is the low-level C module MySQLdb; re-read (or read) README. _mysql is the low-level C module
@ -42,7 +42,7 @@ still have to edit a configuration file so that the setup knows where
to find MySQL and what libraries to include. to find MySQL and what libraries to include.
ImportError: libmysqlclient_r.so.14: cannot open shared object file: No such file or directory ImportError: libmysqlclient_r.so.14: cannot open shared object file: No such file or directory
The number after .so may vary, but this means you have a version of The number after .so may vary, but this means you have a version of
MySQLdb compiled against one version of MySQL, and are now trying to MySQLdb compiled against one version of MySQL, and are now trying to
@ -67,7 +67,7 @@ Solutions:
<http://docs.oracle.com/cd/E19253-01/817-1984/chapter6-63352/>`_. <http://docs.oracle.com/cd/E19253-01/817-1984/chapter6-63352/>`_.
ImportError: ld.so.1: python: fatal: libmtmalloc.so.1: DF_1_NOOPEN tagged object may not be dlopen()'ed ImportError: ld.so.1: python: fatal: libmtmalloc.so.1: DF_1_NOOPEN tagged object may not be dlopen()'ed
This is a weird one from Solaris. What does it mean? I have no idea. This is a weird one from Solaris. What does it mean? I have no idea.
However, things like this can happen if there is some sort of a compiler However, things like this can happen if there is some sort of a compiler
@ -79,9 +79,9 @@ different vendors.
Solution: Rebuild Python or MySQL (or maybe both) from source. Solution: Rebuild Python or MySQL (or maybe both) from source.
ImportError: dlopen(./_mysql.so, 2): Symbol not found: _sprintf$LDBLStub ImportError: dlopen(./_mysql.so, 2): Symbol not found: _sprintf$LDBLStub
Referenced from: ./_mysql.so Referenced from: ./_mysql.so
Expected in: dynamic lookup Expected in: dynamic lookup
This is one from Mac OS X. It seems to have been a compiler mismatch, This is one from Mac OS X. It seems to have been a compiler mismatch,
but this time between two different versions of GCC. It seems nearly but this time between two different versions of GCC. It seems nearly
@ -110,7 +110,7 @@ rolled back, and they cause pending transactions to commit.
Other Errors Other Errors
------------ ------------
OperationalError: (1251, 'Client does not support authentication protocol requested by server; consider upgrading MySQL client') OperationalError: (1251, 'Client does not support authentication protocol requested by server; consider upgrading MySQL client')
This means your server and client libraries are not the same version. This means your server and client libraries are not the same version.
More specifically, it probably means you have a 4.1 or newer server More specifically, it probably means you have a 4.1 or newer server

118
doc/MySQLdb.constants.rst
View File

@ -1,59 +1,59 @@
constants Package constants Package
================= =================
:mod:`constants` Package :mod:`constants` Package
------------------------ ------------------------
.. automodule:: MySQLdb.constants .. automodule:: MySQLdb.constants
:members: :members:
:undoc-members: :undoc-members:
:show-inheritance: :show-inheritance:
:mod:`CLIENT` Module :mod:`CLIENT` Module
-------------------- --------------------
.. automodule:: MySQLdb.constants.CLIENT .. automodule:: MySQLdb.constants.CLIENT
:members: :members:
:undoc-members: :undoc-members:
:show-inheritance: :show-inheritance:
:mod:`CR` Module :mod:`CR` Module
---------------- ----------------
.. automodule:: MySQLdb.constants.CR .. automodule:: MySQLdb.constants.CR
:members: :members:
:undoc-members: :undoc-members:
:show-inheritance: :show-inheritance:
:mod:`ER` Module :mod:`ER` Module
---------------- ----------------
.. automodule:: MySQLdb.constants.ER .. automodule:: MySQLdb.constants.ER
:members: :members:
:undoc-members: :undoc-members:
:show-inheritance: :show-inheritance:
:mod:`FIELD_TYPE` Module :mod:`FIELD_TYPE` Module
------------------------ ------------------------
.. automodule:: MySQLdb.constants.FIELD_TYPE .. automodule:: MySQLdb.constants.FIELD_TYPE
:members: :members:
:undoc-members: :undoc-members:
:show-inheritance: :show-inheritance:
:mod:`FLAG` Module :mod:`FLAG` Module
------------------ ------------------
.. automodule:: MySQLdb.constants.FLAG .. automodule:: MySQLdb.constants.FLAG
:members: :members:
:undoc-members: :undoc-members:
:show-inheritance: :show-inheritance:
:mod:`REFRESH` Module :mod:`REFRESH` Module
--------------------- ---------------------
.. automodule:: MySQLdb.constants.REFRESH .. automodule:: MySQLdb.constants.REFRESH
:members: :members:
:undoc-members: :undoc-members:
:show-inheritance: :show-inheritance:

14
doc/_mysql_exceptions.rst
View File

@ -1,7 +1,7 @@
_mysql_exceptions Module _mysql_exceptions Module
======================== ========================
.. automodule:: _mysql_exceptions .. automodule:: _mysql_exceptions
:members: :members:
:undoc-members: :undoc-members:
:show-inheritance: :show-inheritance:

14
doc/modules.rst
View File

@ -1,7 +1,7 @@
MySQLdb MySQLdb
======= =======
.. toctree:: .. toctree::
:maxdepth: 4 :maxdepth: 4
MySQLdb MySQLdb

1
setup_windows.py
View File

@ -47,4 +47,3 @@ def get_config():
if __name__ == "__main__": if __name__ == "__main__":
sys.stderr.write("""You shouldn't be running this directly; it is used by setup.py.""") sys.stderr.write("""You shouldn't be running this directly; it is used by setup.py.""")

596
tests/capabilities.py
View File

@ -1,298 +1,298 @@
#!/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 from configdb import connection_factory
from MySQLdb.compat import unichr from MySQLdb.compat import unichr
class DatabaseTest(unittest.TestCase): class DatabaseTest(unittest.TestCase):
db_module = None db_module = None
connect_args = () connect_args = ()
connect_kwargs = dict() connect_kwargs = dict()
create_table_extra = '' create_table_extra = ''
rows = 10 rows = 10
debug = False debug = False
def setUp(self): def setUp(self):
import gc import gc
db = connection_factory(**self.connect_kwargs) db = connection_factory(**self.connect_kwargs)
self.connection = db self.connection = db
self.cursor = db.cursor() self.cursor = db.cursor()
self.BLOBUText = u''.join([unichr(i) for i in range(16384)]) self.BLOBUText = u''.join([unichr(i) for i in range(16384)])
self.BLOBBinary = self.db_module.Binary((u''.join([unichr(i) for i in range(256)] * 16)).encode('latin1')) self.BLOBBinary = self.db_module.Binary((u''.join([unichr(i) for i in range(256)] * 16)).encode('latin1'))
leak_test = True leak_test = True
def tearDown(self): def tearDown(self):
if self.leak_test: if self.leak_test:
import gc import gc
del self.cursor del self.cursor
orphans = gc.collect() orphans = gc.collect()
self.failIf(orphans, "%d orphaned objects found after deleting cursor" % orphans) self.failIf(orphans, "%d orphaned objects found after deleting cursor" % orphans)
del self.connection del self.connection
orphans = gc.collect() orphans = gc.collect()
self.failIf(orphans, "%d orphaned objects found after deleting connection" % orphans) self.failIf(orphans, "%d orphaned objects found after deleting connection" % orphans)
def table_exists(self, name): def table_exists(self, name):
try: try:
self.cursor.execute('select * from %s where 1=0' % name) self.cursor.execute('select * from %s where 1=0' % name)
except: except:
return False return False
else: else:
return True return True
def quote_identifier(self, ident): def quote_identifier(self, ident):
return '"%s"' % ident return '"%s"' % ident
def new_table_name(self): def new_table_name(self):
i = id(self.cursor) i = id(self.cursor)
while True: while True:
name = self.quote_identifier('tb%08x' % i) name = self.quote_identifier('tb%08x' % i)
if not self.table_exists(name): if not self.table_exists(name):
return name return name
i = i + 1 i = i + 1
def create_table(self, columndefs): def create_table(self, columndefs):
""" Create a table using a list of column definitions given in """ Create a table using a list of column definitions given in
columndefs. columndefs.
generator must be a function taking arguments (row_number, generator must be a function taking arguments (row_number,
col_number) returning a suitable data object for insertion col_number) returning a suitable data object for insertion
into the table. into the table.
""" """
self.table = self.new_table_name() self.table = self.new_table_name()
self.cursor.execute('CREATE TABLE %s (%s) %s' % self.cursor.execute('CREATE TABLE %s (%s) %s' %
(self.table, (self.table,
',\n'.join(columndefs), ',\n'.join(columndefs),
self.create_table_extra)) self.create_table_extra))
def check_data_integrity(self, columndefs, generator): def check_data_integrity(self, columndefs, generator):
# insert # insert
self.create_table(columndefs) self.create_table(columndefs)
insert_statement = ('INSERT INTO %s VALUES (%s)' % insert_statement = ('INSERT INTO %s VALUES (%s)' %
(self.table, (self.table,
','.join(['%s'] * len(columndefs)))) ','.join(['%s'] * len(columndefs))))
data = [ [ generator(i,j) for j in range(len(columndefs)) ] data = [ [ generator(i,j) for j in range(len(columndefs)) ]
for i in range(self.rows) ] for i in range(self.rows) ]
self.cursor.executemany(insert_statement, data) self.cursor.executemany(insert_statement, data)
self.connection.commit() self.connection.commit()
# verify # verify
self.cursor.execute('select * from %s' % self.table) self.cursor.execute('select * from %s' % self.table)
l = self.cursor.fetchall() l = self.cursor.fetchall()
self.assertEqual(len(l), self.rows) self.assertEqual(len(l), self.rows)
try: try:
for i in range(self.rows): for i in range(self.rows):
for j in range(len(columndefs)): for j in range(len(columndefs)):
self.assertEqual(l[i][j], generator(i,j)) self.assertEqual(l[i][j], generator(i,j))
finally: finally:
if not self.debug: if not self.debug:
self.cursor.execute('drop table %s' % (self.table)) self.cursor.execute('drop table %s' % (self.table))
def test_transactions(self): def test_transactions(self):
columndefs = ( 'col1 INT', 'col2 VARCHAR(255)') columndefs = ( 'col1 INT', 'col2 VARCHAR(255)')
def generator(row, col): def generator(row, col):
if col == 0: return row if col == 0: return row
else: return ('%i' % (row%10))*255 else: return ('%i' % (row%10))*255
self.create_table(columndefs) self.create_table(columndefs)
insert_statement = ('INSERT INTO %s VALUES (%s)' % insert_statement = ('INSERT INTO %s VALUES (%s)' %
(self.table, (self.table,
','.join(['%s'] * len(columndefs)))) ','.join(['%s'] * len(columndefs))))
data = [ [ generator(i,j) for j in range(len(columndefs)) ] data = [ [ generator(i,j) for j in range(len(columndefs)) ]
for i in range(self.rows) ] for i in range(self.rows) ]
self.cursor.executemany(insert_statement, data) self.cursor.executemany(insert_statement, data)
# verify # verify
self.connection.commit() self.connection.commit()
self.cursor.execute('select * from %s' % self.table) self.cursor.execute('select * from %s' % self.table)
l = self.cursor.fetchall() l = self.cursor.fetchall()
self.assertEqual(len(l), self.rows) self.assertEqual(len(l), self.rows)
for i in range(self.rows): for i in range(self.rows):
for j in range(len(columndefs)): for j in range(len(columndefs)):
self.assertEqual(l[i][j], generator(i,j)) self.assertEqual(l[i][j], generator(i,j))
delete_statement = 'delete from %s where col1=%%s' % self.table delete_statement = 'delete from %s where col1=%%s' % self.table
self.cursor.execute(delete_statement, (0,)) self.cursor.execute(delete_statement, (0,))
self.cursor.execute('select col1 from %s where col1=%s' % \ self.cursor.execute('select col1 from %s where col1=%s' % \
(self.table, 0)) (self.table, 0))
l = self.cursor.fetchall() l = self.cursor.fetchall()
self.assertFalse(l, "DELETE didn't work") self.assertFalse(l, "DELETE didn't work")
self.connection.rollback() self.connection.rollback()
self.cursor.execute('select col1 from %s where col1=%s' % \ self.cursor.execute('select col1 from %s where col1=%s' % \
(self.table, 0)) (self.table, 0))
l = self.cursor.fetchall() l = self.cursor.fetchall()
self.assertTrue(len(l) == 1, "ROLLBACK didn't work") self.assertTrue(len(l) == 1, "ROLLBACK didn't work")
self.cursor.execute('drop table %s' % (self.table)) self.cursor.execute('drop table %s' % (self.table))
def test_truncation(self): def test_truncation(self):
columndefs = ( 'col1 INT', 'col2 VARCHAR(255)') columndefs = ( 'col1 INT', 'col2 VARCHAR(255)')
def generator(row, col): def generator(row, col):
if col == 0: return row if col == 0: return row
else: return ('%i' % (row%10))*((255-self.rows//2)+row) else: return ('%i' % (row%10))*((255-self.rows//2)+row)
self.create_table(columndefs) self.create_table(columndefs)
insert_statement = ('INSERT INTO %s VALUES (%s)' % insert_statement = ('INSERT INTO %s VALUES (%s)' %
(self.table, (self.table,
','.join(['%s'] * len(columndefs)))) ','.join(['%s'] * len(columndefs))))
try: try:
self.cursor.execute(insert_statement, (0, '0'*256)) self.cursor.execute(insert_statement, (0, '0'*256))
except self.connection.DataError: except self.connection.DataError:
pass pass
else: else:
self.fail("Over-long column did not generate warnings/exception with single insert") self.fail("Over-long column did not generate warnings/exception with single insert")
self.connection.rollback() self.connection.rollback()
try: try:
for i in range(self.rows): for i in range(self.rows):
data = [] data = []
for j in range(len(columndefs)): for j in range(len(columndefs)):
data.append(generator(i,j)) data.append(generator(i,j))
self.cursor.execute(insert_statement,tuple(data)) self.cursor.execute(insert_statement,tuple(data))
except self.connection.DataError: except self.connection.DataError:
pass pass
else: else:
self.fail("Over-long columns did not generate warnings/exception with execute()") self.fail("Over-long columns did not generate warnings/exception with execute()")
self.connection.rollback() self.connection.rollback()
try: try:
data = [ [ generator(i,j) for j in range(len(columndefs)) ] data = [ [ generator(i,j) for j in range(len(columndefs)) ]
for i in range(self.rows) ] for i in range(self.rows) ]
self.cursor.executemany(insert_statement, data) self.cursor.executemany(insert_statement, data)
except self.connection.DataError: except self.connection.DataError:
pass pass
else: else:
self.fail("Over-long columns did not generate warnings/exception with executemany()") self.fail("Over-long columns did not generate warnings/exception with executemany()")
self.connection.rollback() self.connection.rollback()
self.cursor.execute('drop table %s' % (self.table)) self.cursor.execute('drop table %s' % (self.table))
def test_CHAR(self): def test_CHAR(self):
# Character data # Character data
def generator(row,col): def generator(row,col):
return ('%i' % ((row+col) % 10)) * 255 return ('%i' % ((row+col) % 10)) * 255
self.check_data_integrity( self.check_data_integrity(
('col1 char(255)','col2 char(255)'), ('col1 char(255)','col2 char(255)'),
generator) generator)
def test_INT(self): def test_INT(self):
# Number data # Number data
def generator(row,col): def generator(row,col):
return row*row return row*row
self.check_data_integrity( self.check_data_integrity(
('col1 INT',), ('col1 INT',),
generator) generator)
def test_DECIMAL(self): def test_DECIMAL(self):
# DECIMAL # DECIMAL
from decimal import Decimal from decimal import Decimal
def generator(row,col): def generator(row,col):
return Decimal("%d.%02d" % (row, col)) return Decimal("%d.%02d" % (row, col))
self.check_data_integrity( self.check_data_integrity(
('col1 DECIMAL(5,2)',), ('col1 DECIMAL(5,2)',),
generator) generator)
val = Decimal('1.11111111111111119E-7') val = Decimal('1.11111111111111119E-7')
self.cursor.execute('SELECT %s', (val,)) self.cursor.execute('SELECT %s', (val,))
result = self.cursor.fetchone()[0] result = self.cursor.fetchone()[0]
self.assertEqual(result, val) self.assertEqual(result, val)
self.assertIsInstance(result, Decimal) self.assertIsInstance(result, Decimal)
self.cursor.execute('SELECT %s + %s', (Decimal('0.1'), Decimal('0.2'))) self.cursor.execute('SELECT %s + %s', (Decimal('0.1'), Decimal('0.2')))
result = self.cursor.fetchone()[0] result = self.cursor.fetchone()[0]
self.assertEqual(result, Decimal('0.3')) self.assertEqual(result, Decimal('0.3'))
self.assertIsInstance(result, Decimal) self.assertIsInstance(result, Decimal)
def test_DATE(self): def test_DATE(self):
ticks = time() ticks = time()
def generator(row,col): def generator(row,col):
return self.db_module.DateFromTicks(ticks+row*86400-col*1313) return self.db_module.DateFromTicks(ticks+row*86400-col*1313)
self.check_data_integrity( self.check_data_integrity(
('col1 DATE',), ('col1 DATE',),
generator) generator)
def test_TIME(self): def test_TIME(self):
ticks = time() ticks = time()
def generator(row,col): def generator(row,col):
return self.db_module.TimeFromTicks(ticks+row*86400-col*1313) return self.db_module.TimeFromTicks(ticks+row*86400-col*1313)
self.check_data_integrity( self.check_data_integrity(
('col1 TIME',), ('col1 TIME',),
generator) generator)
def test_DATETIME(self): def test_DATETIME(self):
ticks = time() ticks = time()
def generator(row,col): def generator(row,col):
return self.db_module.TimestampFromTicks(ticks+row*86400-col*1313) return self.db_module.TimestampFromTicks(ticks+row*86400-col*1313)
self.check_data_integrity( self.check_data_integrity(
('col1 DATETIME',), ('col1 DATETIME',),
generator) generator)
def test_TIMESTAMP(self): def test_TIMESTAMP(self):
ticks = time() ticks = time()
def generator(row,col): def generator(row,col):
return self.db_module.TimestampFromTicks(ticks+row*86400-col*1313) return self.db_module.TimestampFromTicks(ticks+row*86400-col*1313)
self.check_data_integrity( self.check_data_integrity(
('col1 TIMESTAMP',), ('col1 TIMESTAMP',),
generator) generator)
def test_fractional_TIMESTAMP(self): def test_fractional_TIMESTAMP(self):
ticks = time() ticks = time()
def generator(row,col): def generator(row,col):
return self.db_module.TimestampFromTicks(ticks+row*86400-col*1313+row*0.7*col/3.0) return self.db_module.TimestampFromTicks(ticks+row*86400-col*1313+row*0.7*col/3.0)
self.check_data_integrity( self.check_data_integrity(
('col1 TIMESTAMP',), ('col1 TIMESTAMP',),
generator) generator)
def test_LONG(self): def test_LONG(self):
def generator(row,col): def generator(row,col):
if col == 0: if col == 0:
return row return row
else: else:
return self.BLOBUText # 'BLOB Text ' * 1024 return self.BLOBUText # 'BLOB Text ' * 1024
self.check_data_integrity( self.check_data_integrity(
('col1 INT','col2 LONG'), ('col1 INT','col2 LONG'),
generator) generator)
def test_TEXT(self): def test_TEXT(self):
def generator(row,col): def generator(row,col):
return self.BLOBUText # 'BLOB Text ' * 1024 return self.BLOBUText # 'BLOB Text ' * 1024
self.check_data_integrity( self.check_data_integrity(
('col2 TEXT',), ('col2 TEXT',),
generator) generator)
def test_LONG_BYTE(self): def test_LONG_BYTE(self):
def generator(row,col): def generator(row,col):
if col == 0: if col == 0:
return row return row
else: else:
return self.BLOBBinary # 'BLOB\000Binary ' * 1024 return self.BLOBBinary # 'BLOB\000Binary ' * 1024
self.check_data_integrity( self.check_data_integrity(
('col1 INT','col2 LONG BYTE'), ('col1 INT','col2 LONG BYTE'),
generator) generator)
def test_BLOB(self): def test_BLOB(self):
def generator(row,col): def generator(row,col):
if col == 0: if col == 0:
return row return row
else: else:
return self.BLOBBinary # 'BLOB\000Binary ' * 1024 return self.BLOBBinary # 'BLOB\000Binary ' * 1024
self.check_data_integrity( self.check_data_integrity(
('col1 INT','col2 BLOB'), ('col1 INT','col2 BLOB'),
generator) generator)
def test_DOUBLE(self): def test_DOUBLE(self):
for val in (18014398509481982.0, 0.1): for val in (18014398509481982.0, 0.1):
self.cursor.execute('SELECT %s', (val,)); self.cursor.execute('SELECT %s', (val,));
result = self.cursor.fetchone()[0] result = self.cursor.fetchone()[0]
self.assertEqual(result, val) self.assertEqual(result, val)
self.assertIsInstance(result, float) self.assertIsInstance(result, float)

32
tests/dbapi20.py
View File

@ -1,6 +1,6 @@
#!/usr/bin/env python #!/usr/bin/env python
''' Python DB API 2.0 driver compliance unit test suite. ''' Python DB API 2.0 driver compliance unit test suite.
This software is Public Domain and may be used without restrictions. This software is Public Domain and may be used without restrictions.
"Now we have booze and barflies entering the discussion, plus rumours of "Now we have booze and barflies entering the discussion, plus rumours of
@ -67,8 +67,8 @@ import time
class DatabaseAPI20Test(unittest.TestCase): class DatabaseAPI20Test(unittest.TestCase):
''' Test a database self.driver for DB API 2.0 compatibility. ''' Test a database self.driver for DB API 2.0 compatibility.
This implementation tests Gadfly, but the TestCase This implementation tests Gadfly, but the TestCase
is structured so that other self.drivers can subclass this is structured so that other self.drivers can subclass this
test case to ensure compiliance with the DB-API. It is test case to ensure compiliance with the DB-API. It is
expected that this TestCase may be expanded in the future expected that this TestCase may be expanded in the future
if ambiguities or edge conditions are discovered. if ambiguities or edge conditions are discovered.
@ -78,9 +78,9 @@ class DatabaseAPI20Test(unittest.TestCase):
self.driver, connect_args and connect_kw_args. Class specification self.driver, connect_args and connect_kw_args. Class specification
should be as follows: should be as follows:
import dbapi20 import dbapi20
class mytest(dbapi20.DatabaseAPI20Test): class mytest(dbapi20.DatabaseAPI20Test):
[...] [...]
Don't 'import DatabaseAPI20Test from dbapi20', or you will Don't 'import DatabaseAPI20Test from dbapi20', or you will
confuse the unit tester - just 'import dbapi20'. confuse the unit tester - just 'import dbapi20'.
@ -99,7 +99,7 @@ class DatabaseAPI20Test(unittest.TestCase):
xddl2 = 'drop table %sbarflys' % table_prefix xddl2 = 'drop table %sbarflys' % table_prefix
lowerfunc = 'lower' # Name of stored procedure to convert string->lowercase lowerfunc = 'lower' # Name of stored procedure to convert string->lowercase
# Some drivers may need to override these helpers, for example adding # Some drivers may need to override these helpers, for example adding
# a 'commit' after the execute. # a 'commit' after the execute.
def executeDDL1(self,cursor): def executeDDL1(self,cursor):
@ -123,10 +123,10 @@ class DatabaseAPI20Test(unittest.TestCase):
try: try:
cur = con.cursor() cur = con.cursor()
for ddl in (self.xddl1,self.xddl2): for ddl in (self.xddl1,self.xddl2):
try: try:
cur.execute(ddl) cur.execute(ddl)
con.commit() con.commit()
except self.driver.Error: except self.driver.Error:
# Assume table didn't exist. Other tests will check if # Assume table didn't exist. Other tests will check if
# execute is busted. # execute is busted.
pass pass
@ -238,7 +238,7 @@ class DatabaseAPI20Test(unittest.TestCase):
con.rollback() con.rollback()
except self.driver.NotSupportedError: except self.driver.NotSupportedError:
pass pass
def test_cursor(self): def test_cursor(self):
con = self._connect() con = self._connect()
try: try:
@ -392,7 +392,7 @@ class DatabaseAPI20Test(unittest.TestCase):
) )
elif self.driver.paramstyle == 'named': elif self.driver.paramstyle == 'named':
cur.execute( cur.execute(
'insert into %sbooze values (:beer)' % self.table_prefix, 'insert into %sbooze values (:beer)' % self.table_prefix,
{'beer':"Cooper's"} {'beer':"Cooper's"}
) )
elif self.driver.paramstyle == 'format': elif self.driver.paramstyle == 'format':
@ -532,7 +532,7 @@ class DatabaseAPI20Test(unittest.TestCase):
tests. tests.
''' '''
populate = [ populate = [
"insert into %sbooze values ('%s')" % (self.table_prefix,s) "insert into %sbooze values ('%s')" % (self.table_prefix,s)
for s in self.samples for s in self.samples
] ]
return populate return populate
@ -593,7 +593,7 @@ class DatabaseAPI20Test(unittest.TestCase):
self.assertEqual(len(rows),6) self.assertEqual(len(rows),6)
rows = [r[0] for r in rows] rows = [r[0] for r in rows]
rows.sort() rows.sort()
# Make sure we get the right data back out # Make sure we get the right data back out
for i in range(0,6): for i in range(0,6):
self.assertEqual(rows[i],self.samples[i], self.assertEqual(rows[i],self.samples[i],
@ -664,10 +664,10 @@ class DatabaseAPI20Test(unittest.TestCase):
'cursor.fetchall should return an empty list if ' 'cursor.fetchall should return an empty list if '
'a select query returns no rows' 'a select query returns no rows'
) )
finally: finally:
con.close() con.close()
def test_mixedfetch(self): def test_mixedfetch(self):
con = self._connect() con = self._connect()
try: try:
@ -703,7 +703,7 @@ class DatabaseAPI20Test(unittest.TestCase):
def help_nextset_setUp(self,cur): def help_nextset_setUp(self,cur):
''' Should create a procedure called deleteme ''' Should create a procedure called deleteme
that returns two result sets, first the that returns two result sets, first the
number of rows in booze then "name from booze" number of rows in booze then "name from booze"
''' '''
raise NotImplementedError('Helper not implemented') raise NotImplementedError('Helper not implemented')

372
tests/test_MySQLdb_capabilities.py
View File

@ -1,186 +1,186 @@
#!/usr/bin/env python #!/usr/bin/env python
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
import capabilities import capabilities
from datetime import timedelta from datetime import timedelta
from contextlib import closing from contextlib import closing
import unittest import unittest
import MySQLdb import MySQLdb
from MySQLdb.compat import unicode from MySQLdb.compat import unicode
from MySQLdb import cursors from MySQLdb import cursors
from configdb import connection_factory from configdb import connection_factory
import warnings import warnings
warnings.filterwarnings('ignore') warnings.filterwarnings('ignore')
class test_MySQLdb(capabilities.DatabaseTest): class test_MySQLdb(capabilities.DatabaseTest):
db_module = MySQLdb db_module = MySQLdb
connect_args = () connect_args = ()
connect_kwargs = dict(use_unicode=True, sql_mode="ANSI,STRICT_TRANS_TABLES,TRADITIONAL") connect_kwargs = dict(use_unicode=True, 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):
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 Lärche pine'.split()))) list(enumerate('ash birch cedar Lärche 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', ('Lärche',)) c.callproc('test_sp', ('Lärche',))
rows = c.fetchall() rows = c.fetchall()
self.assertEqual(len(rows), 1) self.assertEqual(len(rows), 1)
self.assertEqual(rows[0][0], 3) self.assertEqual(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_BIT(self): def test_BIT(self):
c = self.cursor c = self.cursor
try: try:
c.execute("""create table test_BIT ( c.execute("""create table test_BIT (
b3 BIT(3), b3 BIT(3),
b7 BIT(10), b7 BIT(10),
b64 BIT(64))""") b64 BIT(64))""")
one64 = '1'*64 one64 = '1'*64
c.execute( c.execute(
"insert into test_BIT (b3, b7, b64)" "insert into test_BIT (b3, b7, b64)"
" VALUES (b'011', b'1111111111', b'%s')" " VALUES (b'011', b'1111111111', b'%s')"
% one64) % one64)
c.execute("SELECT b3, b7, b64 FROM test_BIT") c.execute("SELECT b3, b7, b64 FROM test_BIT")
row = c.fetchone() row = c.fetchone()
self.assertEqual(row[0], b'\x03') self.assertEqual(row[0], b'\x03')
self.assertEqual(row[1], b'\x03\xff') self.assertEqual(row[1], b'\x03\xff')
self.assertEqual(row[2], b'\xff'*8) self.assertEqual(row[2], b'\xff'*8)
finally: finally:
c.execute("drop table if exists test_BIT") c.execute("drop table if exists test_BIT")
def test_MULTIPOLYGON(self): def test_MULTIPOLYGON(self):
c = self.cursor c = self.cursor
try: try:
c.execute("""create table test_MULTIPOLYGON ( c.execute("""create table test_MULTIPOLYGON (
id INTEGER PRIMARY KEY, id INTEGER PRIMARY KEY,
border MULTIPOLYGON)""") border MULTIPOLYGON)""")
c.execute( c.execute(
"insert into test_MULTIPOLYGON (id, border)" "insert into test_MULTIPOLYGON (id, border)"
" VALUES (1, GeomFromText('MULTIPOLYGON(((1 1, 1 -1, -1 -1, -1 1, 1 1)),((1 1, 3 1, 3 3, 1 3, 1 1)))'))" " VALUES (1, GeomFromText('MULTIPOLYGON(((1 1, 1 -1, -1 -1, -1 1, 1 1)),((1 1, 3 1, 3 3, 1 3, 1 1)))'))"
) )
c.execute("SELECT id, AsText(border) FROM test_MULTIPOLYGON") c.execute("SELECT id, AsText(border) FROM test_MULTIPOLYGON")
row = c.fetchone() row = c.fetchone()
self.assertEqual(row[0], 1) self.assertEqual(row[0], 1)
self.assertEqual(row[1], 'MULTIPOLYGON(((1 1,1 -1,-1 -1,-1 1,1 1)),((1 1,3 1,3 3,1 3,1 1)))') self.assertEqual(row[1], 'MULTIPOLYGON(((1 1,1 -1,-1 -1,-1 1,1 1)),((1 1,3 1,3 3,1 3,1 1)))')
c.execute("SELECT id, AsWKB(border) FROM test_MULTIPOLYGON") c.execute("SELECT id, AsWKB(border) FROM test_MULTIPOLYGON")
row = c.fetchone() row = c.fetchone()
self.assertEqual(row[0], 1) self.assertEqual(row[0], 1)
self.assertNotEqual(len(row[1]), 0) self.assertNotEqual(len(row[1]), 0)
c.execute("SELECT id, border FROM test_MULTIPOLYGON") c.execute("SELECT id, border FROM test_MULTIPOLYGON")
row = c.fetchone() row = c.fetchone()
self.assertEqual(row[0], 1) self.assertEqual(row[0], 1)
self.assertNotEqual(len(row[1]), 0) self.assertNotEqual(len(row[1]), 0)
finally: finally:
c.execute("drop table if exists test_MULTIPOLYGON") c.execute("drop table if exists test_MULTIPOLYGON")
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 as msg: except self.connection.ProgrammingError as msg:
self.assertTrue(str(ER.NO_SUCH_TABLE) in str(msg)) self.assertTrue(str(ER.NO_SUCH_TABLE) in str(msg))
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()
def test_reraise_exception(self): def test_reraise_exception(self):
c = self.cursor c = self.cursor
try: try:
c.execute("SELECT x FROM not_existing_table") c.execute("SELECT x FROM not_existing_table")
except MySQLdb.ProgrammingError as e: except MySQLdb.ProgrammingError as e:
self.assertEqual(e.args[0], 1146) self.assertEqual(e.args[0], 1146)
return return
self.fail("Should raise ProgrammingError") self.fail("Should raise ProgrammingError")
def test_binary_prefix(self): def test_binary_prefix(self):
# verify prefix behaviour when enabled, disabled and for default (disabled) # verify prefix behaviour when enabled, disabled and for default (disabled)
for binary_prefix in (True, False, None): for binary_prefix in (True, False, None):
kwargs = self.connect_kwargs.copy() kwargs = self.connect_kwargs.copy()
# needs to be set to can guarantee CHARSET response for normal strings # needs to be set to can guarantee CHARSET response for normal strings
kwargs['charset'] = 'utf8' kwargs['charset'] = 'utf8'
if binary_prefix != None: if binary_prefix != None:
kwargs['binary_prefix'] = binary_prefix kwargs['binary_prefix'] = binary_prefix
with closing(connection_factory(**kwargs)) as conn: with closing(connection_factory(**kwargs)) as conn:
with closing(conn.cursor()) as c: with closing(conn.cursor()) as c:
c.execute('SELECT CHARSET(%s)', (MySQLdb.Binary(b'raw bytes'),)) c.execute('SELECT CHARSET(%s)', (MySQLdb.Binary(b'raw bytes'),))
self.assertEqual(c.fetchall()[0][0], 'binary' if binary_prefix else 'utf8') self.assertEqual(c.fetchall()[0][0], 'binary' if binary_prefix else 'utf8')
# normal strings should not get prefix # normal strings should not get prefix
c.execute('SELECT CHARSET(%s)', ('str',)) c.execute('SELECT CHARSET(%s)', ('str',))
self.assertEqual(c.fetchall()[0][0], 'utf8') self.assertEqual(c.fetchall()[0][0], 'utf8')
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()

10
tests/test_MySQLdb_dbapi20.py
View File

@ -20,7 +20,7 @@ class test_MySQLdb(dbapi20.DatabaseAPI20Test):
test for an exception if the statement cannot return a test for an exception if the statement cannot return a
result set. MySQL always returns a result set; it's just that result set. MySQL always returns a result set; it's just that
some things return empty result sets.""" some things return empty result sets."""
def test_fetchall(self): def test_fetchall(self):
con = self._connect() con = self._connect()
try: try:
@ -66,10 +66,10 @@ class test_MySQLdb(dbapi20.DatabaseAPI20Test):
'cursor.fetchall should return an empty list if ' 'cursor.fetchall should return an empty list if '
'a select query returns no rows' 'a select query returns no rows'
) )
finally: finally:
con.close() con.close()
def test_fetchone(self): def test_fetchone(self):
con = self._connect() con = self._connect()
try: try:
@ -148,7 +148,7 @@ class test_MySQLdb(dbapi20.DatabaseAPI20Test):
def help_nextset_setUp(self,cur): def help_nextset_setUp(self,cur):
''' Should create a procedure called deleteme ''' Should create a procedure called deleteme
that returns two result sets, first the that returns two result sets, first the
number of rows in booze then "name from booze" number of rows in booze then "name from booze"
''' '''
sql=""" sql="""
@ -200,6 +200,6 @@ class test_MySQLdb(dbapi20.DatabaseAPI20Test):
finally: finally:
con.close() con.close()
if __name__ == '__main__': if __name__ == '__main__':
unittest.main() unittest.main()