| #!/usr/bin/env python |
| |
| # Licensed to the Apache Software Foundation (ASF) under one or more |
| # contributor license agreements. See the NOTICE file distributed with |
| # this work for additional information regarding copyright ownership. |
| # The ASF licenses this file to You under the Apache License, Version 2.0 |
| # (the "License"); you may not use this file except in compliance with |
| # the License. You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # See the License for the specific language governing permissions and |
| # limitations under the License. |
| |
| ''' Python DB API 2.0 driver compliance unit test suite. |
| |
| This software is Public Domain and may be used without restrictions. |
| |
| "Now we have booze and barflies entering the discussion, plus rumours of |
| DBAs on drugs... and I won't tell you what flashes through my mind each |
| time I read the subject line with 'Anal Compliance' in it. All around |
| this is turning out to be a thoroughly unwholesome unit test." |
| |
| -- Ian Bicking |
| ''' |
| |
| __version__ = '1.14.3' |
| |
| import unittest |
| import time |
| import sys |
| |
| if sys.version[0] >= '3': #python 3.x |
| _BaseException = Exception |
| def _failUnless(self, expr, msg=None): |
| self.assertTrue(expr, msg) |
| else: #python 2.x |
| from exceptions import StandardError as _BaseException |
| def _failUnless(self, expr, msg=None): |
| self.failUnless(expr, msg) ## deprecated since Python 2.6 |
| |
| def str2bytes(sval): |
| if sys.version_info < (3,0) and isinstance(sval, str): |
| sval = sval.decode("latin1") |
| return sval.encode("latin1") #python 3 make unicode into bytes |
| |
| class DatabaseAPI20Test(unittest.TestCase): |
| ''' Test a database self.driver for DB API 2.0 compatibility. |
| This implementation tests Gadfly, but the TestCase |
| is structured so that other self.drivers can subclass this |
| test case to ensure compiliance with the DB-API. It is |
| expected that this TestCase may be expanded in the future |
| if ambiguities or edge conditions are discovered. |
| |
| The 'Optional Extensions' are not yet being tested. |
| |
| self.drivers should subclass this test, overriding setUp, tearDown, |
| self.driver, connect_args and connect_kw_args. Class specification |
| should be as follows: |
| |
| import dbapi20 |
| class mytest(dbapi20.DatabaseAPI20Test): |
| [...] |
| |
| Don't 'import DatabaseAPI20Test from dbapi20', or you will |
| confuse the unit tester - just 'import dbapi20'. |
| ''' |
| |
| # The self.driver module. This should be the module where the 'connect' |
| # method is to be found |
| driver = None |
| connect_args = () # List of arguments to pass to connect |
| connect_kw_args = {} # Keyword arguments for connect |
| table_prefix = 'dbapi20test_' # If you need to specify a prefix for tables |
| |
| ddl1 = 'create table %sbooze (name varchar(20))' % table_prefix |
| ddl2 = 'create table %sbarflys (name varchar(20), drink varchar(30))' % table_prefix |
| xddl1 = 'drop table %sbooze' % table_prefix |
| xddl2 = 'drop table %sbarflys' % table_prefix |
| insert = 'insert' |
| |
| lowerfunc = 'lower' # Name of stored procedure to convert string->lowercase |
| |
| # Some drivers may need to override these helpers, for example adding |
| # a 'commit' after the execute. |
| def executeDDL1(self,cursor): |
| cursor.execute(self.ddl1) |
| |
| def executeDDL2(self,cursor): |
| cursor.execute(self.ddl2) |
| |
| def setUp(self): |
| ''' self.drivers should override this method to perform required setup |
| if any is necessary, such as creating the database. |
| ''' |
| pass |
| |
| def tearDown(self): |
| ''' self.drivers should override this method to perform required cleanup |
| if any is necessary, such as deleting the test database. |
| The default drops the tables that may be created. |
| ''' |
| try: |
| con = self._connect() |
| try: |
| cur = con.cursor() |
| for ddl in (self.xddl1,self.xddl2): |
| try: |
| cur.execute(ddl) |
| con.commit() |
| except self.driver.Error: |
| # Assume table didn't exist. Other tests will check if |
| # execute is busted. |
| pass |
| finally: |
| con.close() |
| except _BaseException: |
| pass |
| |
| def _connect(self): |
| try: |
| r = self.driver.connect( |
| *self.connect_args,**self.connect_kw_args |
| ) |
| except AttributeError: |
| self.fail("No connect method found in self.driver module") |
| return r |
| |
| def test_connect(self): |
| con = self._connect() |
| con.close() |
| |
| def test_apilevel(self): |
| try: |
| # Must exist |
| apilevel = self.driver.apilevel |
| # Must equal 2.0 |
| self.assertEqual(apilevel,'2.0') |
| except AttributeError: |
| self.fail("Driver doesn't define apilevel") |
| |
| def test_threadsafety(self): |
| try: |
| # Must exist |
| threadsafety = self.driver.threadsafety |
| # Must be a valid value |
| _failUnless(self, threadsafety in (0,1,2,3)) |
| except AttributeError: |
| self.fail("Driver doesn't define threadsafety") |
| |
| def test_paramstyle(self): |
| try: |
| # Must exist |
| paramstyle = self.driver.paramstyle |
| # Must be a valid value |
| _failUnless(self, paramstyle in ( |
| 'qmark','numeric','named','format','pyformat' |
| )) |
| except AttributeError: |
| self.fail("Driver doesn't define paramstyle") |
| |
| def test_Exceptions(self): |
| # Make sure required exceptions exist, and are in the |
| # defined heirarchy. |
| if sys.version[0] == '3': #under Python 3 StardardError no longer exists |
| self.assertTrue(issubclass(self.driver.Warning,Exception)) |
| self.assertTrue(issubclass(self.driver.Error,Exception)) |
| else: |
| self.failUnless(issubclass(self.driver.Warning,StandardError)) |
| self.failUnless(issubclass(self.driver.Error,StandardError)) |
| |
| _failUnless(self, |
| issubclass(self.driver.InterfaceError,self.driver.Error) |
| ) |
| _failUnless(self, |
| issubclass(self.driver.DatabaseError,self.driver.Error) |
| ) |
| _failUnless(self, |
| issubclass(self.driver.OperationalError,self.driver.Error) |
| ) |
| _failUnless(self, |
| issubclass(self.driver.IntegrityError,self.driver.Error) |
| ) |
| _failUnless(self, |
| issubclass(self.driver.InternalError,self.driver.Error) |
| ) |
| _failUnless(self, |
| issubclass(self.driver.ProgrammingError,self.driver.Error) |
| ) |
| _failUnless(self, |
| issubclass(self.driver.NotSupportedError,self.driver.Error) |
| ) |
| |
| def test_ExceptionsAsConnectionAttributes(self): |
| # OPTIONAL EXTENSION |
| # Test for the optional DB API 2.0 extension, where the exceptions |
| # are exposed as attributes on the Connection object |
| # I figure this optional extension will be implemented by any |
| # driver author who is using this test suite, so it is enabled |
| # by default. |
| con = self._connect() |
| drv = self.driver |
| _failUnless(self,con.Warning is drv.Warning) |
| _failUnless(self,con.Error is drv.Error) |
| _failUnless(self,con.InterfaceError is drv.InterfaceError) |
| _failUnless(self,con.DatabaseError is drv.DatabaseError) |
| _failUnless(self,con.OperationalError is drv.OperationalError) |
| _failUnless(self,con.IntegrityError is drv.IntegrityError) |
| _failUnless(self,con.InternalError is drv.InternalError) |
| _failUnless(self,con.ProgrammingError is drv.ProgrammingError) |
| _failUnless(self,con.NotSupportedError is drv.NotSupportedError) |
| |
| |
| def test_commit(self): |
| con = self._connect() |
| try: |
| # Commit must work, even if it doesn't do anything |
| con.commit() |
| finally: |
| con.close() |
| |
| def test_rollback(self): |
| con = self._connect() |
| # If rollback is defined, it should either work or throw |
| # the documented exception |
| if hasattr(con,'rollback'): |
| try: |
| con.rollback() |
| except self.driver.NotSupportedError: |
| pass |
| |
| def test_cursor(self): |
| con = self._connect() |
| try: |
| cur = con.cursor() |
| finally: |
| con.close() |
| |
| def test_cursor_isolation(self): |
| con = self._connect() |
| try: |
| # Make sure cursors created from the same connection have |
| # the documented transaction isolation level |
| cur1 = con.cursor() |
| cur2 = con.cursor() |
| self.executeDDL1(cur1) |
| cur1.execute("%s into %sbooze values ('Victoria Bitter')" % ( |
| self.insert, self.table_prefix |
| )) |
| cur2.execute("select name from %sbooze" % self.table_prefix) |
| booze = cur2.fetchall() |
| self.assertEqual(len(booze),1) |
| self.assertEqual(len(booze[0]),1) |
| self.assertEqual(booze[0][0],'Victoria Bitter') |
| finally: |
| con.close() |
| |
| def test_description(self): |
| con = self._connect() |
| try: |
| cur = con.cursor() |
| self.executeDDL1(cur) |
| self.assertEqual(cur.description,None, |
| 'cursor.description should be none after executing a ' |
| 'statement that can return no rows (such as DDL)' |
| ) |
| cur.execute('select name from %sbooze' % self.table_prefix) |
| self.assertEqual(len(cur.description),1, |
| 'cursor.description describes too many columns' |
| ) |
| self.assertEqual(len(cur.description[0]),7, |
| 'cursor.description[x] tuples must have 7 elements' |
| ) |
| self.assertEqual(cur.description[0][0].lower(),'name', |
| 'cursor.description[x][0] must return column name' |
| ) |
| self.assertEqual(cur.description[0][1],self.driver.STRING, |
| 'cursor.description[x][1] must return column type. Got %r' |
| % cur.description[0][1] |
| ) |
| |
| # Make sure self.description gets reset |
| self.executeDDL2(cur) |
| self.assertEqual(cur.description,None, |
| 'cursor.description not being set to None when executing ' |
| 'no-result statements (eg. DDL)' |
| ) |
| finally: |
| con.close() |
| |
| def test_rowcount(self): |
| con = self._connect() |
| try: |
| cur = con.cursor() |
| self.executeDDL1(cur) |
| _failUnless(self,cur.rowcount in (-1,0), # Bug #543885 |
| 'cursor.rowcount should be -1 or 0 after executing no-result ' |
| 'statements' |
| ) |
| cur.execute("%s into %sbooze values ('Victoria Bitter')" % ( |
| self.insert, self.table_prefix |
| )) |
| _failUnless(self,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) |
| _failUnless(self,cur.rowcount in (-1,1), |
| 'cursor.rowcount should == number of rows returned, or ' |
| 'set to -1 after executing a select statement' |
| ) |
| self.executeDDL2(cur) |
| _failUnless(self,cur.rowcount in (-1,0), # Bug #543885 |
| 'cursor.rowcount should be -1 or 0 after executing no-result ' |
| 'statements' |
| ) |
| finally: |
| con.close() |
| |
| lower_func = 'lower' |
| def test_callproc(self): |
| con = self._connect() |
| try: |
| cur = con.cursor() |
| if self.lower_func and hasattr(cur,'callproc'): |
| r = cur.callproc(self.lower_func,('FOO',)) |
| self.assertEqual(len(r),1) |
| self.assertEqual(r[0],'FOO') |
| r = cur.fetchall() |
| self.assertEqual(len(r),1,'callproc produced no result set') |
| self.assertEqual(len(r[0]),1, |
| 'callproc produced invalid result set' |
| ) |
| self.assertEqual(r[0][0],'foo', |
| 'callproc produced invalid results' |
| ) |
| finally: |
| con.close() |
| |
| def test_close(self): |
| con = self._connect() |
| try: |
| cur = con.cursor() |
| finally: |
| con.close() |
| |
| # cursor.execute should raise an Error if called after connection |
| # closed |
| self.assertRaises(self.driver.Error,self.executeDDL1,cur) |
| |
| # connection.commit should raise an Error if called after connection' |
| # closed.' |
| self.assertRaises(self.driver.Error,con.commit) |
| |
| def test_non_idempotent_close(self): |
| con = self._connect() |
| con.close() |
| # connection.close should raise an Error if called more than once |
| #!!! reasonable persons differ about the usefulness of this test and this feature !!! |
| self.assertRaises(self.driver.Error,con.close) |
| |
| def test_execute(self): |
| con = self._connect() |
| try: |
| cur = con.cursor() |
| self._paraminsert(cur) |
| finally: |
| con.close() |
| |
| def _paraminsert(self,cur): |
| self.executeDDL2(cur) |
| cur.execute("%s into %sbarflys values ('Victoria Bitter', 'thi%%s :may ca%%(u)se? troub:1e')" % ( |
| self.insert, self.table_prefix |
| )) |
| _failUnless(self,cur.rowcount in (-1,1)) |
| |
| if self.driver.paramstyle == 'qmark': |
| cur.execute( |
| "%s into %sbarflys values (?, 'thi%%s :may ca%%(u)se? troub:1e')" % (self.insert, self.table_prefix), |
| ("Cooper's",) |
| ) |
| elif self.driver.paramstyle == 'numeric': |
| cur.execute( |
| "%s into %sbarflys values (:1, 'thi%%s :may ca%%(u)se? troub:1e')" % (self.insert, self.table_prefix), |
| ("Cooper's",) |
| ) |
| elif self.driver.paramstyle == 'named': |
| cur.execute( |
| "%s into %sbarflys values (:beer, 'thi%%s :may ca%%(u)se? troub:1e')" % (self.insert, self.table_prefix), |
| {'beer':"Cooper's"} |
| ) |
| elif self.driver.paramstyle == 'format': |
| cur.execute( |
| "%s into %sbarflys values (%%s, 'thi%%%%s :may ca%%%%(u)se? troub:1e')" % (self.insert, self.table_prefix), |
| ("Cooper's",) |
| ) |
| elif self.driver.paramstyle == 'pyformat': |
| cur.execute( |
| "%s into %sbarflys values (%%(beer)s, 'thi%%%%s :may ca%%%%(u)se? troub:1e')" % (self.insert, self.table_prefix), |
| {'beer':"Cooper's"} |
| ) |
| else: |
| self.fail('Invalid paramstyle') |
| _failUnless(self,cur.rowcount in (-1,1)) |
| |
| cur.execute('select name, drink from %sbarflys' % self.table_prefix) |
| res = cur.fetchall() |
| self.assertEqual(len(res),2,'cursor.fetchall returned too few rows') |
| beers = [res[0][0],res[1][0]] |
| beers.sort() |
| self.assertEqual(beers[0],"Cooper's", |
| 'cursor.fetchall retrieved incorrect data, or data inserted ' |
| 'incorrectly' |
| ) |
| self.assertEqual(beers[1],"Victoria Bitter", |
| 'cursor.fetchall retrieved incorrect data, or data inserted ' |
| 'incorrectly' |
| ) |
| trouble = "thi%s :may ca%(u)se? troub:1e" |
| self.assertEqual(res[0][1], trouble, |
| 'cursor.fetchall retrieved incorrect data, or data inserted ' |
| 'incorrectly. Got=%s, Expected=%s' % (repr(res[0][1]), repr(trouble))) |
| self.assertEqual(res[1][1], trouble, |
| 'cursor.fetchall retrieved incorrect data, or data inserted ' |
| 'incorrectly. Got=%s, Expected=%s' % (repr(res[1][1]), repr(trouble) |
| )) |
| |
| def test_executemany(self): |
| con = self._connect() |
| try: |
| cur = con.cursor() |
| self.executeDDL1(cur) |
| largs = [ ("Cooper's",) , ("Boag's",) ] |
| margs = [ {'beer': "Cooper's"}, {'beer': "Boag's"} ] |
| if self.driver.paramstyle == 'qmark': |
| cur.executemany( |
| '%s into %sbooze values (?)' % (self.insert, self.table_prefix), |
| largs |
| ) |
| elif self.driver.paramstyle == 'numeric': |
| cur.executemany( |
| '%s into %sbooze values (:1)' % (self.insert, self.table_prefix), |
| largs |
| ) |
| elif self.driver.paramstyle == 'named': |
| cur.executemany( |
| '%s into %sbooze values (:beer)' % (self.insert, self.table_prefix), |
| margs |
| ) |
| elif self.driver.paramstyle == 'format': |
| cur.executemany( |
| '%s into %sbooze values (%%s)' % (self.insert, self.table_prefix), |
| largs |
| ) |
| elif self.driver.paramstyle == 'pyformat': |
| cur.executemany( |
| '%s into %sbooze values (%%(beer)s)' % ( |
| self.insert, self.table_prefix |
| ), |
| margs |
| ) |
| else: |
| self.fail('Unknown paramstyle') |
| _failUnless(self,cur.rowcount in (-1,2), |
| 'insert using cursor.executemany set cursor.rowcount to ' |
| 'incorrect value %r' % cur.rowcount |
| ) |
| cur.execute('select name from %sbooze' % self.table_prefix) |
| res = cur.fetchall() |
| self.assertEqual(len(res),2, |
| 'cursor.fetchall retrieved incorrect number of rows' |
| ) |
| beers = [res[0][0],res[1][0]] |
| beers.sort() |
| self.assertEqual(beers[0],"Boag's",'incorrect data retrieved') |
| self.assertEqual(beers[1],"Cooper's",'incorrect data retrieved') |
| 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' |
| ) |
| _failUnless(self,cur.rowcount in (-1,0)) |
| |
| # cursor.fetchone should raise an Error if called after |
| # executing a query that cannnot return rows |
| cur.execute("%s into %sbooze values ('Victoria Bitter')" % ( |
| self.insert, 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' |
| ) |
| _failUnless(self,cur.rowcount in (-1,1)) |
| finally: |
| con.close() |
| |
| samples = [ |
| 'Carlton Cold', |
| 'Carlton Draft', |
| 'Mountain Goat', |
| 'Redback', |
| 'Victoria Bitter', |
| 'XXXX' |
| ] |
| |
| def _populate(self): |
| ''' Return a list of sql commands to setup the DB for the fetch |
| tests. |
| ''' |
| populate = [ |
| "%s into %sbooze values ('%s')" % (self.insert, self.table_prefix, s) |
| for s in self.samples |
| ] |
| return populate |
| |
| def test_fetchmany(self): |
| con = self._connect() |
| try: |
| cur = con.cursor() |
| |
| # cursor.fetchmany should raise an Error if called without |
| #issuing a query |
| self.assertRaises(self.driver.Error,cur.fetchmany,4) |
| |
| self.executeDDL1(cur) |
| for sql in self._populate(): |
| cur.execute(sql) |
| |
| cur.execute('select name from %sbooze' % self.table_prefix) |
| r = cur.fetchmany() |
| self.assertEqual(len(r),1, |
| 'cursor.fetchmany retrieved incorrect number of rows, ' |
| 'default of arraysize is one.' |
| ) |
| cur.arraysize=10 |
| r = cur.fetchmany(3) # Should get 3 rows |
| self.assertEqual(len(r),3, |
| 'cursor.fetchmany retrieved incorrect number of rows' |
| ) |
| r = cur.fetchmany(4) # Should get 2 more |
| self.assertEqual(len(r),2, |
| 'cursor.fetchmany retrieved incorrect number of rows' |
| ) |
| r = cur.fetchmany(4) # Should be an empty sequence |
| self.assertEqual(len(r),0, |
| 'cursor.fetchmany should return an empty sequence after ' |
| 'results are exhausted' |
| ) |
| _failUnless(self,cur.rowcount in (-1,6)) |
| |
| # Same as above, using cursor.arraysize |
| cur.arraysize=4 |
| cur.execute('select name from %sbooze' % self.table_prefix) |
| r = cur.fetchmany() # Should get 4 rows |
| self.assertEqual(len(r),4, |
| 'cursor.arraysize not being honoured by fetchmany' |
| ) |
| r = cur.fetchmany() # Should get 2 more |
| self.assertEqual(len(r),2) |
| r = cur.fetchmany() # Should be an empty sequence |
| self.assertEqual(len(r),0) |
| _failUnless(self,cur.rowcount in (-1,6)) |
| |
| cur.arraysize=6 |
| cur.execute('select name from %sbooze' % self.table_prefix) |
| rows = cur.fetchmany() # Should get all rows |
| _failUnless(self,cur.rowcount in (-1,6)) |
| self.assertEqual(len(rows),6) |
| self.assertEqual(len(rows),6) |
| rows = [r[0] for r in rows] |
| rows.sort() |
| |
| # Make sure we get the right data back out |
| for i in range(0,6): |
| self.assertEqual(rows[i],self.samples[i], |
| 'incorrect data retrieved by cursor.fetchmany' |
| ) |
| |
| rows = cur.fetchmany() # Should return an empty list |
| self.assertEqual(len(rows),0, |
| 'cursor.fetchmany should return an empty sequence if ' |
| 'called after the whole result set has been fetched' |
| ) |
| _failUnless(self,cur.rowcount in (-1,6)) |
| |
| self.executeDDL2(cur) |
| cur.execute('select name from %sbarflys' % self.table_prefix) |
| r = cur.fetchmany() # Should get empty sequence |
| self.assertEqual(len(r),0, |
| 'cursor.fetchmany should return an empty sequence if ' |
| 'query retrieved no rows' |
| ) |
| _failUnless(self,cur.rowcount in (-1,0)) |
| |
| finally: |
| con.close() |
| |
| 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() |
| _failUnless(self,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' |
| ) |
| _failUnless(self,cur.rowcount in (-1,len(self.samples))) |
| |
| self.executeDDL2(cur) |
| cur.execute('select name from %sbarflys' % self.table_prefix) |
| rows = cur.fetchall() |
| _failUnless(self,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_mixedfetch(self): |
| con = self._connect() |
| try: |
| cur = con.cursor() |
| self.executeDDL1(cur) |
| for sql in self._populate(): |
| cur.execute(sql) |
| |
| cur.execute('select name from %sbooze' % self.table_prefix) |
| rows1 = cur.fetchone() |
| rows23 = cur.fetchmany(2) |
| rows4 = cur.fetchone() |
| rows56 = cur.fetchall() |
| _failUnless(self,cur.rowcount in (-1,6)) |
| self.assertEqual(len(rows23),2, |
| 'fetchmany returned incorrect number of rows' |
| ) |
| self.assertEqual(len(rows56),2, |
| 'fetchall returned incorrect number of rows' |
| ) |
| |
| rows = [rows1[0]] |
| rows.extend([rows23[0][0],rows23[1][0]]) |
| rows.append(rows4[0]) |
| rows.extend([rows56[0][0],rows56[1][0]]) |
| rows.sort() |
| for i in range(0,len(self.samples)): |
| self.assertEqual(rows[i],self.samples[i], |
| 'incorrect data retrieved or inserted' |
| ) |
| finally: |
| con.close() |
| |
| 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" |
| ''' |
| raise NotImplementedError('Helper not implemented') |
| #sql=""" |
| # create procedure deleteme as |
| # begin |
| # select count(*) from booze |
| # select name from booze |
| # end |
| #""" |
| #cur.execute(sql) |
| |
| def help_nextset_tearDown(self,cur): |
| 'If cleaning up is needed after nextSetTest' |
| raise NotImplementedError('Helper not implemented') |
| #cur.execute("drop procedure deleteme") |
| |
| def test_nextset(self): |
| 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() |
| assert s == None,'No more return sets, should return None' |
| finally: |
| self.help_nextset_tearDown(cur) |
| |
| finally: |
| con.close() |
| |
| def test_nextset(self): |
| raise NotImplementedError('Drivers need to override this test') |
| |
| def test_arraysize(self): |
| # Not much here - rest of the tests for this are in test_fetchmany |
| con = self._connect() |
| try: |
| cur = con.cursor() |
| _failUnless(self,hasattr(cur,'arraysize'), |
| 'cursor.arraysize must be defined' |
| ) |
| finally: |
| con.close() |
| |
| def test_setinputsizes(self): |
| con = self._connect() |
| try: |
| cur = con.cursor() |
| cur.setinputsizes( (25,) ) |
| self._paraminsert(cur) # Make sure cursor still works |
| finally: |
| con.close() |
| |
| def test_setoutputsize_basic(self): |
| # Basic test is to make sure setoutputsize doesn't blow up |
| con = self._connect() |
| try: |
| cur = con.cursor() |
| cur.setoutputsize(1000) |
| cur.setoutputsize(2000,0) |
| self._paraminsert(cur) # Make sure the cursor still works |
| finally: |
| con.close() |
| |
| def test_setoutputsize(self): |
| # Real test for setoutputsize is driver dependant |
| raise NotImplementedError('Driver needed to override this test') |
| |
| def test_None(self): |
| con = self._connect() |
| try: |
| cur = con.cursor() |
| self.executeDDL1(cur) |
| cur.execute("%s into %sbarflys values ('a', NULL)" % (self.insert, self.table_prefix)) |
| cur.execute('select drink from %sbarflys' % self.table_prefix) |
| r = cur.fetchall() |
| self.assertEqual(len(r),1) |
| self.assertEqual(len(r[0]),1) |
| self.assertEqual(r[0][0],None,'NULL value not returned as None') |
| finally: |
| con.close() |
| |
| def test_Date(self): |
| d1 = self.driver.Date(2002,12,25) |
| d2 = self.driver.DateFromTicks(time.mktime((2002,12,25,0,0,0,0,0,0))) |
| # Can we assume this? API doesn't specify, but it seems implied |
| # self.assertEqual(str(d1),str(d2)) |
| |
| def test_Time(self): |
| t1 = self.driver.Time(13,45,30) |
| t2 = self.driver.TimeFromTicks(time.mktime((2001,1,1,13,45,30,0,0,0))) |
| # Can we assume this? API doesn't specify, but it seems implied |
| # self.assertEqual(str(t1),str(t2)) |
| |
| def test_Timestamp(self): |
| t1 = self.driver.Timestamp(2002,12,25,13,45,30) |
| t2 = self.driver.TimestampFromTicks( |
| time.mktime((2002,12,25,13,45,30,0,0,0)) |
| ) |
| # Can we assume this? API doesn't specify, but it seems implied |
| # self.assertEqual(str(t1),str(t2)) |
| |
| def test_Binary(self): |
| b = self.driver.Binary(str2bytes('Something')) |
| b = self.driver.Binary(str2bytes('')) |
| |
| def test_STRING(self): |
| _failUnless(self, hasattr(self.driver,'STRING'), |
| 'module.STRING must be defined' |
| ) |
| |
| def test_BINARY(self): |
| _failUnless(self, hasattr(self.driver,'BINARY'), |
| 'module.BINARY must be defined.' |
| ) |
| |
| def test_NUMBER(self): |
| _failUnless(self, hasattr(self.driver,'NUMBER'), |
| 'module.NUMBER must be defined.' |
| ) |
| |
| def test_DATETIME(self): |
| _failUnless(self, hasattr(self.driver,'DATETIME'), |
| 'module.DATETIME must be defined.' |
| ) |
| |
| def test_ROWID(self): |
| _failUnless(self, hasattr(self.driver,'ROWID'), |
| 'module.ROWID must be defined.' |
| ) |
