Update UnitTest methods deprecated in Python 2.7

This commit is contained in:
adustman
2012-09-07 15:32:48 +00:00
parent 3070f46d7d
commit 81dbbc5174
5 changed files with 56 additions and 55 deletions

View File

@ -127,12 +127,12 @@ class DatabaseTest(unittest.TestCase):
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.failIf(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.failUnless(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):

View File

@ -159,7 +159,7 @@ class DatabaseAPI20Test(unittest.TestCase):
# Must exist # Must exist
threadsafety = self.driver.threadsafety threadsafety = self.driver.threadsafety
# Must be a valid value # Must be a valid value
self.failUnless(threadsafety in (0,1,2,3)) self.assertTrue(threadsafety in (0,1,2,3))
except AttributeError: except AttributeError:
self.fail("Driver doesn't define threadsafety") self.fail("Driver doesn't define threadsafety")
@ -168,7 +168,7 @@ class DatabaseAPI20Test(unittest.TestCase):
# Must exist # Must exist
paramstyle = self.driver.paramstyle paramstyle = self.driver.paramstyle
# Must be a valid value # Must be a valid value
self.failUnless(paramstyle in ( self.assertTrue(paramstyle in (
'qmark','numeric','named','format','pyformat' 'qmark','numeric','named','format','pyformat'
)) ))
except AttributeError: except AttributeError:
@ -177,27 +177,27 @@ class DatabaseAPI20Test(unittest.TestCase):
def test_Exceptions(self): def test_Exceptions(self):
# Make sure required exceptions exist, and are in the # Make sure required exceptions exist, and are in the
# defined heirarchy. # defined heirarchy.
self.failUnless(issubclass(self.driver.Warning,StandardError)) self.assertTrue(issubclass(self.driver.Warning,StandardError))
self.failUnless(issubclass(self.driver.Error,StandardError)) self.assertTrue(issubclass(self.driver.Error,StandardError))
self.failUnless( self.assertTrue(
issubclass(self.driver.InterfaceError,self.driver.Error) issubclass(self.driver.InterfaceError,self.driver.Error)
) )
self.failUnless( self.assertTrue(
issubclass(self.driver.DatabaseError,self.driver.Error) issubclass(self.driver.DatabaseError,self.driver.Error)
) )
self.failUnless( self.assertTrue(
issubclass(self.driver.OperationalError,self.driver.Error) issubclass(self.driver.OperationalError,self.driver.Error)
) )
self.failUnless( self.assertTrue(
issubclass(self.driver.IntegrityError,self.driver.Error) issubclass(self.driver.IntegrityError,self.driver.Error)
) )
self.failUnless( self.assertTrue(
issubclass(self.driver.InternalError,self.driver.Error) issubclass(self.driver.InternalError,self.driver.Error)
) )
self.failUnless( self.assertTrue(
issubclass(self.driver.ProgrammingError,self.driver.Error) issubclass(self.driver.ProgrammingError,self.driver.Error)
) )
self.failUnless( self.assertTrue(
issubclass(self.driver.NotSupportedError,self.driver.Error) issubclass(self.driver.NotSupportedError,self.driver.Error)
) )
@ -210,15 +210,15 @@ class DatabaseAPI20Test(unittest.TestCase):
# by default. # by default.
con = self._connect() con = self._connect()
drv = self.driver drv = self.driver
self.failUnless(con.Warning is drv.Warning) self.assertTrue(con.Warning is drv.Warning)
self.failUnless(con.Error is drv.Error) self.assertTrue(con.Error is drv.Error)
self.failUnless(con.InterfaceError is drv.InterfaceError) self.assertTrue(con.InterfaceError is drv.InterfaceError)
self.failUnless(con.DatabaseError is drv.DatabaseError) self.assertTrue(con.DatabaseError is drv.DatabaseError)
self.failUnless(con.OperationalError is drv.OperationalError) self.assertTrue(con.OperationalError is drv.OperationalError)
self.failUnless(con.IntegrityError is drv.IntegrityError) self.assertTrue(con.IntegrityError is drv.IntegrityError)
self.failUnless(con.InternalError is drv.InternalError) self.assertTrue(con.InternalError is drv.InternalError)
self.failUnless(con.ProgrammingError is drv.ProgrammingError) self.assertTrue(con.ProgrammingError is drv.ProgrammingError)
self.failUnless(con.NotSupportedError is drv.NotSupportedError) self.assertTrue(con.NotSupportedError is drv.NotSupportedError)
def test_commit(self): def test_commit(self):
@ -310,12 +310,12 @@ class DatabaseAPI20Test(unittest.TestCase):
cur.execute("insert into %sbooze values ('Victoria Bitter')" % ( cur.execute("insert into %sbooze values ('Victoria Bitter')" % (
self.table_prefix self.table_prefix
)) ))
self.failUnless(cur.rowcount in (-1,1), self.assertTrue(cur.rowcount in (-1,1),
'cursor.rowcount should == number or rows inserted, or ' 'cursor.rowcount should == number or rows inserted, or '
'set to -1 after executing an insert statement' 'set to -1 after executing an insert statement'
) )
cur.execute("select name from %sbooze" % self.table_prefix) cur.execute("select name from %sbooze" % self.table_prefix)
self.failUnless(cur.rowcount in (-1,1), self.assertTrue(cur.rowcount in (-1,1),
'cursor.rowcount should == number of rows returned, or ' 'cursor.rowcount should == number of rows returned, or '
'set to -1 after executing a select statement' 'set to -1 after executing a select statement'
) )
@ -378,7 +378,7 @@ class DatabaseAPI20Test(unittest.TestCase):
cur.execute("insert into %sbooze values ('Victoria Bitter')" % ( cur.execute("insert into %sbooze values ('Victoria Bitter')" % (
self.table_prefix self.table_prefix
)) ))
self.failUnless(cur.rowcount in (-1,1)) self.assertTrue(cur.rowcount in (-1,1))
if self.driver.paramstyle == 'qmark': if self.driver.paramstyle == 'qmark':
cur.execute( cur.execute(
@ -407,7 +407,7 @@ class DatabaseAPI20Test(unittest.TestCase):
) )
else: else:
self.fail('Invalid paramstyle') self.fail('Invalid paramstyle')
self.failUnless(cur.rowcount in (-1,1)) self.assertTrue(cur.rowcount in (-1,1))
cur.execute('select name from %sbooze' % self.table_prefix) cur.execute('select name from %sbooze' % self.table_prefix)
res = cur.fetchall() res = cur.fetchall()
@ -459,7 +459,7 @@ class DatabaseAPI20Test(unittest.TestCase):
) )
else: else:
self.fail('Unknown paramstyle') self.fail('Unknown paramstyle')
self.failUnless(cur.rowcount in (-1,2), self.assertTrue(cur.rowcount in (-1,2),
'insert using cursor.executemany set cursor.rowcount to ' 'insert using cursor.executemany set cursor.rowcount to '
'incorrect value %r' % cur.rowcount 'incorrect value %r' % cur.rowcount
) )
@ -494,7 +494,7 @@ class DatabaseAPI20Test(unittest.TestCase):
'cursor.fetchone should return None if a query retrieves ' 'cursor.fetchone should return None if a query retrieves '
'no rows' 'no rows'
) )
self.failUnless(cur.rowcount in (-1,0)) self.assertTrue(cur.rowcount in (-1,0))
# cursor.fetchone should raise an Error if called after # cursor.fetchone should raise an Error if called after
# executing a query that cannnot return rows # executing a query that cannnot return rows
@ -514,7 +514,7 @@ class DatabaseAPI20Test(unittest.TestCase):
self.assertEqual(cur.fetchone(),None, self.assertEqual(cur.fetchone(),None,
'cursor.fetchone should return None if no more rows available' 'cursor.fetchone should return None if no more rows available'
) )
self.failUnless(cur.rowcount in (-1,1)) self.assertTrue(cur.rowcount in (-1,1))
finally: finally:
con.close() con.close()
@ -570,7 +570,7 @@ class DatabaseAPI20Test(unittest.TestCase):
'cursor.fetchmany should return an empty sequence after ' 'cursor.fetchmany should return an empty sequence after '
'results are exhausted' 'results are exhausted'
) )
self.failUnless(cur.rowcount in (-1,6)) self.assertTrue(cur.rowcount in (-1,6))
# Same as above, using cursor.arraysize # Same as above, using cursor.arraysize
cur.arraysize=4 cur.arraysize=4
@ -583,12 +583,12 @@ class DatabaseAPI20Test(unittest.TestCase):
self.assertEqual(len(r),2) self.assertEqual(len(r),2)
r = cur.fetchmany() # Should be an empty sequence r = cur.fetchmany() # Should be an empty sequence
self.assertEqual(len(r),0) self.assertEqual(len(r),0)
self.failUnless(cur.rowcount in (-1,6)) self.assertTrue(cur.rowcount in (-1,6))
cur.arraysize=6 cur.arraysize=6
cur.execute('select name from %sbooze' % self.table_prefix) cur.execute('select name from %sbooze' % self.table_prefix)
rows = cur.fetchmany() # Should get all rows rows = cur.fetchmany() # Should get all rows
self.failUnless(cur.rowcount in (-1,6)) self.assertTrue(cur.rowcount in (-1,6))
self.assertEqual(len(rows),6) self.assertEqual(len(rows),6)
self.assertEqual(len(rows),6) self.assertEqual(len(rows),6)
rows = [r[0] for r in rows] rows = [r[0] for r in rows]
@ -605,7 +605,7 @@ class DatabaseAPI20Test(unittest.TestCase):
'cursor.fetchmany should return an empty sequence if ' 'cursor.fetchmany should return an empty sequence if '
'called after the whole result set has been fetched' 'called after the whole result set has been fetched'
) )
self.failUnless(cur.rowcount in (-1,6)) self.assertTrue(cur.rowcount in (-1,6))
self.executeDDL2(cur) self.executeDDL2(cur)
cur.execute('select name from %sbarflys' % self.table_prefix) cur.execute('select name from %sbarflys' % self.table_prefix)
@ -614,7 +614,7 @@ class DatabaseAPI20Test(unittest.TestCase):
'cursor.fetchmany should return an empty sequence if ' 'cursor.fetchmany should return an empty sequence if '
'query retrieved no rows' 'query retrieved no rows'
) )
self.failUnless(cur.rowcount in (-1,0)) self.assertTrue(cur.rowcount in (-1,0))
finally: finally:
con.close() con.close()
@ -638,7 +638,7 @@ class DatabaseAPI20Test(unittest.TestCase):
cur.execute('select name from %sbooze' % self.table_prefix) cur.execute('select name from %sbooze' % self.table_prefix)
rows = cur.fetchall() rows = cur.fetchall()
self.failUnless(cur.rowcount in (-1,len(self.samples))) self.assertTrue(cur.rowcount in (-1,len(self.samples)))
self.assertEqual(len(rows),len(self.samples), self.assertEqual(len(rows),len(self.samples),
'cursor.fetchall did not retrieve all rows' 'cursor.fetchall did not retrieve all rows'
) )
@ -654,12 +654,12 @@ class DatabaseAPI20Test(unittest.TestCase):
'cursor.fetchall should return an empty list if called ' 'cursor.fetchall should return an empty list if called '
'after the whole result set has been fetched' 'after the whole result set has been fetched'
) )
self.failUnless(cur.rowcount in (-1,len(self.samples))) self.assertTrue(cur.rowcount in (-1,len(self.samples)))
self.executeDDL2(cur) self.executeDDL2(cur)
cur.execute('select name from %sbarflys' % self.table_prefix) cur.execute('select name from %sbarflys' % self.table_prefix)
rows = cur.fetchall() rows = cur.fetchall()
self.failUnless(cur.rowcount in (-1,0)) self.assertTrue(cur.rowcount in (-1,0))
self.assertEqual(len(rows),0, self.assertEqual(len(rows),0,
'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'
@ -681,7 +681,7 @@ class DatabaseAPI20Test(unittest.TestCase):
rows23 = cur.fetchmany(2) rows23 = cur.fetchmany(2)
rows4 = cur.fetchone() rows4 = cur.fetchone()
rows56 = cur.fetchall() rows56 = cur.fetchall()
self.failUnless(cur.rowcount in (-1,6)) self.assertTrue(cur.rowcount in (-1,6))
self.assertEqual(len(rows23),2, self.assertEqual(len(rows23),2,
'fetchmany returned incorrect number of rows' 'fetchmany returned incorrect number of rows'
) )
@ -758,7 +758,7 @@ class DatabaseAPI20Test(unittest.TestCase):
con = self._connect() con = self._connect()
try: try:
cur = con.cursor() cur = con.cursor()
self.failUnless(hasattr(cur,'arraysize'), self.assertTrue(hasattr(cur,'arraysize'),
'cursor.arraysize must be defined' 'cursor.arraysize must be defined'
) )
finally: finally:
@ -827,27 +827,27 @@ class DatabaseAPI20Test(unittest.TestCase):
b = self.driver.Binary('') b = self.driver.Binary('')
def test_STRING(self): def test_STRING(self):
self.failUnless(hasattr(self.driver,'STRING'), self.assertTrue(hasattr(self.driver,'STRING'),
'module.STRING must be defined' 'module.STRING must be defined'
) )
def test_BINARY(self): def test_BINARY(self):
self.failUnless(hasattr(self.driver,'BINARY'), self.assertTrue(hasattr(self.driver,'BINARY'),
'module.BINARY must be defined.' 'module.BINARY must be defined.'
) )
def test_NUMBER(self): def test_NUMBER(self):
self.failUnless(hasattr(self.driver,'NUMBER'), self.assertTrue(hasattr(self.driver,'NUMBER'),
'module.NUMBER must be defined.' 'module.NUMBER must be defined.'
) )
def test_DATETIME(self): def test_DATETIME(self):
self.failUnless(hasattr(self.driver,'DATETIME'), self.assertTrue(hasattr(self.driver,'DATETIME'),
'module.DATETIME must be defined.' 'module.DATETIME must be defined.'
) )
def test_ROWID(self): def test_ROWID(self):
self.failUnless(hasattr(self.driver,'ROWID'), self.assertTrue(hasattr(self.driver,'ROWID'),
'module.ROWID must be defined.' 'module.ROWID must be defined.'
) )

View File

@ -10,7 +10,7 @@ class test_MySQLdb(capabilities.DatabaseTest):
db_module = MySQLdb db_module = MySQLdb
connect_args = () connect_args = ()
connect_kwargs = dict(db='test', read_default_file='~/.my.cnf', connect_kwargs = dict(db='test', host="127.0.0.1", user="test", #read_default_file='~/.my.cnf',
charset='utf8', 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
@ -78,7 +78,7 @@ class test_MySQLdb(capabilities.DatabaseTest):
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.failUnless(msg[0] == ER.NO_SUCH_TABLE) self.assertTrue(msg[0] == ER.NO_SUCH_TABLE)
def test_ping(self): def test_ping(self):
self.connection.ping() self.connection.ping()

View File

@ -7,7 +7,8 @@ class test_MySQLdb(dbapi20.DatabaseAPI20Test):
driver = MySQLdb driver = MySQLdb
connect_args = () connect_args = ()
connect_kw_args = dict(db='test', connect_kw_args = dict(db='test',
read_default_file='~/.my.cnf', host="127.0.0.1",
user="test", #read_default_file='~/.my.cnf',
charset='utf8', charset='utf8',
sql_mode="ANSI,STRICT_TRANS_TABLES,TRADITIONAL") sql_mode="ANSI,STRICT_TRANS_TABLES,TRADITIONAL")
@ -39,7 +40,7 @@ class test_MySQLdb(dbapi20.DatabaseAPI20Test):
cur.execute('select name from %sbooze' % self.table_prefix) cur.execute('select name from %sbooze' % self.table_prefix)
rows = cur.fetchall() rows = cur.fetchall()
self.failUnless(cur.rowcount in (-1,len(self.samples))) self.assertTrue(cur.rowcount in (-1,len(self.samples)))
self.assertEqual(len(rows),len(self.samples), self.assertEqual(len(rows),len(self.samples),
'cursor.fetchall did not retrieve all rows' 'cursor.fetchall did not retrieve all rows'
) )
@ -55,12 +56,12 @@ class test_MySQLdb(dbapi20.DatabaseAPI20Test):
'cursor.fetchall should return an empty list if called ' 'cursor.fetchall should return an empty list if called '
'after the whole result set has been fetched' 'after the whole result set has been fetched'
) )
self.failUnless(cur.rowcount in (-1,len(self.samples))) self.assertTrue(cur.rowcount in (-1,len(self.samples)))
self.executeDDL2(cur) self.executeDDL2(cur)
cur.execute('select name from %sbarflys' % self.table_prefix) cur.execute('select name from %sbarflys' % self.table_prefix)
rows = cur.fetchall() rows = cur.fetchall()
self.failUnless(cur.rowcount in (-1,0)) self.assertTrue(cur.rowcount in (-1,0))
self.assertEqual(len(rows),0, self.assertEqual(len(rows),0,
'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'
@ -88,7 +89,7 @@ class test_MySQLdb(dbapi20.DatabaseAPI20Test):
'cursor.fetchone should return None if a query retrieves ' 'cursor.fetchone should return None if a query retrieves '
'no rows' 'no rows'
) )
self.failUnless(cur.rowcount in (-1,0)) self.assertTrue(cur.rowcount in (-1,0))
# cursor.fetchone should raise an Error if called after # cursor.fetchone should raise an Error if called after
# executing a query that cannnot return rows # executing a query that cannnot return rows
@ -108,7 +109,7 @@ class test_MySQLdb(dbapi20.DatabaseAPI20Test):
## self.assertEqual(cur.fetchone(),None, ## self.assertEqual(cur.fetchone(),None,
## 'cursor.fetchone should return None if no more rows available' ## 'cursor.fetchone should return None if no more rows available'
## ) ## )
self.failUnless(cur.rowcount in (-1,1)) self.assertTrue(cur.rowcount in (-1,1))
finally: finally:
con.close() con.close()
@ -125,12 +126,12 @@ class test_MySQLdb(dbapi20.DatabaseAPI20Test):
cur.execute("insert into %sbooze values ('Victoria Bitter')" % ( cur.execute("insert into %sbooze values ('Victoria Bitter')" % (
self.table_prefix self.table_prefix
)) ))
## self.failUnless(cur.rowcount in (-1,1), ## self.assertTrue(cur.rowcount in (-1,1),
## 'cursor.rowcount should == number or rows inserted, or ' ## 'cursor.rowcount should == number or rows inserted, or '
## 'set to -1 after executing an insert statement' ## 'set to -1 after executing an insert statement'
## ) ## )
cur.execute("select name from %sbooze" % self.table_prefix) cur.execute("select name from %sbooze" % self.table_prefix)
self.failUnless(cur.rowcount in (-1,1), self.assertTrue(cur.rowcount in (-1,1),
'cursor.rowcount should == number of rows returned, or ' 'cursor.rowcount should == number of rows returned, or '
'set to -1 after executing a select statement' 'set to -1 after executing a select statement'
) )

View File

@ -44,7 +44,7 @@ class CoreAPI(unittest.TestCase):
"""Test _mysql interaction internals.""" """Test _mysql interaction internals."""
def setUp(self): def setUp(self):
self.conn = _mysql.connect(db='test', read_default_file="~/.my.cnf") self.conn = _mysql.connect(db='test', host='127.0.0.1', user='test') #read_default_file="~/.my.cnf")
def tearDown(self): def tearDown(self):
self.conn.close() self.conn.close()