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