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