|
|
@@ -1,857 +0,0 @@
|
|
|
-#!/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.'
|
|
|
- )
|