source: trunk/tests/dbapi20.py

Last change on this file was 969, checked in by cito, 3 months ago

Shebang should not be followed by a blank

It's a myth that it is needed because some old versions of Unix expect it.
However, some editors do not like a blank here.

  • Property svn:keywords set to Id
File size: 30.8 KB
Line 
1#!/usr/bin/python
2''' Python DB API 2.0 driver compliance unit test suite.
3
4    This software is Public Domain and may be used without restrictions.
5
6 "Now we have booze and barflies entering the discussion, plus rumours of
7  DBAs on drugs... and I won't tell you what flashes through my mind each
8  time I read the subject line with 'Anal Compliance' in it.  All around
9  this is turning out to be a thoroughly unwholesome unit test."
10
11    -- Ian Bicking
12'''
13
14__rcs_id__  = '$Id: dbapi20.py 969 2019-04-19 13:35:23Z cito $'
15__version__ = '$Revision: 1.5 $'[11:-2]
16__author__ = 'Stuart Bishop <zen@shangri-la.dropbear.id.au>'
17
18try:
19    import unittest2 as unittest  # for Python < 2.7
20except ImportError:
21    import unittest
22
23import time
24
25# $Log: not supported by cvs2svn $
26# Revision 1.4  2008/11/01 18:37:55  cito
27# Updated the dbapi20 test module. Exposed the exceptions as attributes of the connection.
28#
29# Revision 1.10  2003/10/09 03:14:14  zenzen
30# Add test for DB API 2.0 optional extension, where database exceptions
31# are exposed as attributes on the Connection object.
32#
33# Revision 1.9  2003/08/13 01:16:36  zenzen
34# Minor tweak from Stefan Fleiter
35#
36# Revision 1.8  2003/04/10 00:13:25  zenzen
37# Changes, as per suggestions by M.-A. Lemburg
38# - Add a table prefix, to ensure namespace collisions can always be avoided
39#
40# Revision 1.7  2003/02/26 23:33:37  zenzen
41# Break out DDL into helper functions, as per request by David Rushby
42#
43# Revision 1.6  2003/02/21 03:04:33  zenzen
44# Stuff from Henrik Ekelund:
45#     added test_None
46#     added test_nextset & hooks
47#
48# Revision 1.5  2003/02/17 22:08:43  zenzen
49# Implement suggestions and code from Henrik Eklund - test that cursor.arraysize
50# defaults to 1 & generic cursor.callproc test added
51#
52# Revision 1.4  2003/02/15 00:16:33  zenzen
53# Changes, as per suggestions and bug reports by M.-A. Lemburg,
54# Matthew T. Kromer, Federico Di Gregorio and Daniel Dittmar
55# - Class renamed
56# - Now a subclass of TestCase, to avoid requiring the driver stub
57#   to use multiple inheritance
58# - Reversed the polarity of buggy test in test_description
59# - Test exception heirarchy correctly
60# - self.populate is now self._populate(), so if a driver stub
61#   overrides self.ddl1 this change propogates
62# - VARCHAR columns now have a width, which will hopefully make the
63#   DDL even more portible (this will be reversed if it causes more problems)
64# - cursor.rowcount being checked after various execute and fetchXXX methods
65# - Check for fetchall and fetchmany returning empty lists after results
66#   are exhausted (already checking for empty lists if select retrieved
67#   nothing
68# - Fix bugs in test_setoutputsize_basic and test_setinputsizes
69#
70
71class DatabaseAPI20Test(unittest.TestCase):
72    ''' Test a database self.driver for DB API 2.0 compatibility.
73        This implementation tests Gadfly, but the TestCase
74        is structured so that other self.drivers can subclass this
75        test case to ensure compiliance with the DB-API. It is
76        expected that this TestCase may be expanded in the future
77        if ambiguities or edge conditions are discovered.
78
79        The 'Optional Extensions' are not yet being tested.
80
81        self.drivers should subclass this test, overriding setUp, tearDown,
82        self.driver, connect_args and connect_kw_args. Class specification
83        should be as follows:
84
85        import dbapi20
86        class mytest(dbapi20.DatabaseAPI20Test):
87           [...]
88
89        Don't 'import DatabaseAPI20Test from dbapi20', or you will
90        confuse the unit tester - just 'import dbapi20'.
91    '''
92
93    # The self.driver module. This should be the module where the 'connect'
94    # method is to be found
95    driver = None
96    connect_args = () # List of arguments to pass to connect
97    connect_kw_args = {} # Keyword arguments for connect
98    table_prefix = 'dbapi20test_' # If you need to specify a prefix for tables
99
100    ddl1 = 'create table %sbooze (name varchar(20))' % table_prefix
101    ddl2 = 'create table %sbarflys (name varchar(20))' % table_prefix
102    xddl1 = 'drop table %sbooze' % table_prefix
103    xddl2 = 'drop table %sbarflys' % table_prefix
104
105    lowerfunc = 'lower' # Name of stored procedure to convert string->lowercase
106
107    # Some drivers may need to override these helpers, for example adding
108    # a 'commit' after the execute.
109    def executeDDL1(self,cursor):
110        cursor.execute(self.ddl1)
111
112    def executeDDL2(self,cursor):
113        cursor.execute(self.ddl2)
114
115    def setUp(self):
116        """self.drivers should override this method to perform required setup
117            if any is necessary, such as creating the database.
118        """
119        pass
120
121    def tearDown(self):
122        """self.drivers should override this method to perform required cleanup
123            if any is necessary, such as deleting the test database.
124            The default drops the tables that may be created.
125        """
126        con = self._connect()
127        try:
128            cur = con.cursor()
129            for ddl in (self.xddl1,self.xddl2):
130                try:
131                    cur.execute(ddl)
132                    con.commit()
133                except self.driver.Error:
134                    # Assume table didn't exist. Other tests will check if
135                    # execute is busted.
136                    pass
137        finally:
138            con.close()
139
140    def _connect(self):
141        try:
142            return self.driver.connect(
143                *self.connect_args,**self.connect_kw_args
144                )
145        except AttributeError:
146            self.fail("No connect method found in self.driver module")
147
148    def test_connect(self):
149        con = self._connect()
150        con.close()
151
152    def test_apilevel(self):
153        try:
154            # Must exist
155            apilevel = self.driver.apilevel
156            # Must equal 2.0
157            self.assertEqual(apilevel,'2.0')
158        except AttributeError:
159            self.fail("Driver doesn't define apilevel")
160
161    def test_threadsafety(self):
162        try:
163            # Must exist
164            threadsafety = self.driver.threadsafety
165            # Must be a valid value
166            self.assertTrue(threadsafety in (0,1,2,3))
167        except AttributeError:
168            self.fail("Driver doesn't define threadsafety")
169
170    def test_paramstyle(self):
171        try:
172            # Must exist
173            paramstyle = self.driver.paramstyle
174            # Must be a valid value
175            self.assertTrue(paramstyle in (
176                'qmark','numeric','named','format','pyformat'
177                ))
178        except AttributeError:
179            self.fail("Driver doesn't define paramstyle")
180
181    def test_Exceptions(self):
182        """Make sure required exceptions exist, and are in the
183        defined hierarchy.
184        """
185        self.assertTrue(issubclass(self.driver.Warning,Exception))
186        self.assertTrue(issubclass(self.driver.Error,Exception))
187        self.assertTrue(
188            issubclass(self.driver.InterfaceError,self.driver.Error)
189            )
190        self.assertTrue(
191            issubclass(self.driver.DatabaseError,self.driver.Error)
192            )
193        self.assertTrue(
194            issubclass(self.driver.OperationalError,self.driver.Error)
195            )
196        self.assertTrue(
197            issubclass(self.driver.IntegrityError,self.driver.Error)
198            )
199        self.assertTrue(
200            issubclass(self.driver.InternalError,self.driver.Error)
201            )
202        self.assertTrue(
203            issubclass(self.driver.ProgrammingError,self.driver.Error)
204            )
205        self.assertTrue(
206            issubclass(self.driver.NotSupportedError,self.driver.Error)
207            )
208
209    def test_ExceptionsAsConnectionAttributes(self):
210        """Optional extension
211
212        Test for the optional DB API 2.0 extension, where the exceptions
213        are exposed as attributes on the Connection object
214        I figure this optional extension will be implemented by any
215        driver author who is using this test suite, so it is enabled
216        by default.
217        """
218        con = self._connect()
219        drv = self.driver
220        self.assertTrue(con.Warning is drv.Warning)
221        self.assertTrue(con.Error is drv.Error)
222        self.assertTrue(con.InterfaceError is drv.InterfaceError)
223        self.assertTrue(con.DatabaseError is drv.DatabaseError)
224        self.assertTrue(con.OperationalError is drv.OperationalError)
225        self.assertTrue(con.IntegrityError is drv.IntegrityError)
226        self.assertTrue(con.InternalError is drv.InternalError)
227        self.assertTrue(con.ProgrammingError is drv.ProgrammingError)
228        self.assertTrue(con.NotSupportedError is drv.NotSupportedError)
229
230    def test_commit(self):
231        con = self._connect()
232        try:
233            # Commit must work, even if it doesn't do anything
234            con.commit()
235        finally:
236            con.close()
237
238    def test_rollback(self):
239        con = self._connect()
240        # If rollback is defined, it should either work or throw
241        # the documented exception
242        if hasattr(con,'rollback'):
243            try:
244                con.rollback()
245            except self.driver.NotSupportedError:
246                pass
247
248    def test_cursor(self):
249        con = self._connect()
250        try:
251            cur = con.cursor()
252        finally:
253            con.close()
254
255    def test_cursor_isolation(self):
256        con = self._connect()
257        try:
258            # Make sure cursors created from the same connection have
259            # the documented transaction isolation level
260            cur1 = con.cursor()
261            cur2 = con.cursor()
262            self.executeDDL1(cur1)
263            cur1.execute("insert into %sbooze values ('Victoria Bitter')" % (
264                self.table_prefix
265                ))
266            cur2.execute("select name from %sbooze" % self.table_prefix)
267            booze = cur2.fetchall()
268            self.assertEqual(len(booze),1)
269            self.assertEqual(len(booze[0]),1)
270            self.assertEqual(booze[0][0],'Victoria Bitter')
271        finally:
272            con.close()
273
274    def test_description(self):
275        con = self._connect()
276        try:
277            cur = con.cursor()
278            self.executeDDL1(cur)
279            self.assertEqual(cur.description,None,
280                'cursor.description should be none after executing a '
281                'statement that can return no rows (such as DDL)'
282                )
283            cur.execute('select name from %sbooze' % self.table_prefix)
284            self.assertEqual(len(cur.description),1,
285                'cursor.description describes too many columns'
286                )
287            self.assertEqual(len(cur.description[0]),7,
288                'cursor.description[x] tuples must have 7 elements'
289                )
290            self.assertEqual(cur.description[0][0].lower(),'name',
291                'cursor.description[x][0] must return column name'
292                )
293            self.assertEqual(cur.description[0][1],self.driver.STRING,
294                'cursor.description[x][1] must return column type. Got %r'
295                    % cur.description[0][1]
296                )
297
298            # Make sure self.description gets reset
299            self.executeDDL2(cur)
300            self.assertEqual(cur.description,None,
301                'cursor.description not being set to None when executing '
302                'no-result statements (eg. DDL)'
303                )
304        finally:
305            con.close()
306
307    def test_rowcount(self):
308        con = self._connect()
309        try:
310            cur = con.cursor()
311            self.executeDDL1(cur)
312            self.assertEqual(cur.rowcount,-1,
313                'cursor.rowcount should be -1 after executing no-result '
314                'statements'
315                )
316            cur.execute("insert into %sbooze values ('Victoria Bitter')" % (
317                self.table_prefix
318                ))
319            self.assertTrue(cur.rowcount in (-1,1),
320                'cursor.rowcount should == number or rows inserted, or '
321                'set to -1 after executing an insert statement'
322                )
323            cur.execute("select name from %sbooze" % self.table_prefix)
324            self.assertTrue(cur.rowcount in (-1,1),
325                'cursor.rowcount should == number of rows returned, or '
326                'set to -1 after executing a select statement'
327                )
328            self.executeDDL2(cur)
329            self.assertEqual(cur.rowcount,-1,
330                'cursor.rowcount not being reset to -1 after executing '
331                'no-result statements'
332                )
333        finally:
334            con.close()
335
336    lower_func = 'lower'
337    def test_callproc(self):
338        con = self._connect()
339        try:
340            cur = con.cursor()
341            if self.lower_func and hasattr(cur,'callproc'):
342                r = cur.callproc(self.lower_func,('FOO',))
343                self.assertEqual(len(r),1)
344                self.assertEqual(r[0],'FOO')
345                r = cur.fetchall()
346                self.assertEqual(len(r),1,'callproc produced no result set')
347                self.assertEqual(len(r[0]),1,
348                    'callproc produced invalid result set'
349                    )
350                self.assertEqual(r[0][0],'foo',
351                    'callproc produced invalid results'
352                    )
353        finally:
354            con.close()
355
356    def test_close(self):
357        con = self._connect()
358        try:
359            cur = con.cursor()
360        finally:
361            con.close()
362
363        # cursor.execute should raise an Error if called after connection
364        # closed
365        self.assertRaises(self.driver.Error,self.executeDDL1,cur)
366
367        # connection.commit should raise an Error if called after connection'
368        # closed.'
369        self.assertRaises(self.driver.Error,con.commit)
370
371        # connection.close should raise an Error if called more than once
372        self.assertRaises(self.driver.Error,con.close)
373
374    def test_execute(self):
375        con = self._connect()
376        try:
377            cur = con.cursor()
378            self._paraminsert(cur)
379        finally:
380            con.close()
381
382    def _paraminsert(self,cur):
383        self.executeDDL1(cur)
384        cur.execute("insert into %sbooze values ('Victoria Bitter')" % (
385            self.table_prefix
386            ))
387        self.assertTrue(cur.rowcount in (-1,1))
388
389        if self.driver.paramstyle == 'qmark':
390            cur.execute(
391                'insert into %sbooze values (?)' % self.table_prefix,
392                ("Cooper's",)
393                )
394        elif self.driver.paramstyle == 'numeric':
395            cur.execute(
396                'insert into %sbooze values (:1)' % self.table_prefix,
397                ("Cooper's",)
398                )
399        elif self.driver.paramstyle == 'named':
400            cur.execute(
401                'insert into %sbooze values (:beer)' % self.table_prefix,
402                {'beer':"Cooper's"}
403                )
404        elif self.driver.paramstyle == 'format':
405            cur.execute(
406                'insert into %sbooze values (%%s)' % self.table_prefix,
407                ("Cooper's",)
408                )
409        elif self.driver.paramstyle == 'pyformat':
410            cur.execute(
411                'insert into %sbooze values (%%(beer)s)' % self.table_prefix,
412                {'beer':"Cooper's"}
413                )
414        else:
415            self.fail('Invalid paramstyle')
416        self.assertTrue(cur.rowcount in (-1,1))
417
418        cur.execute('select name from %sbooze' % self.table_prefix)
419        res = cur.fetchall()
420        self.assertEqual(len(res),2,'cursor.fetchall returned too few rows')
421        beers = [res[0][0],res[1][0]]
422        beers.sort()
423        self.assertEqual(beers[0],"Cooper's",
424            'cursor.fetchall retrieved incorrect data, or data inserted '
425            'incorrectly'
426            )
427        self.assertEqual(beers[1],"Victoria Bitter",
428            'cursor.fetchall retrieved incorrect data, or data inserted '
429            'incorrectly'
430            )
431
432    def test_executemany(self):
433        con = self._connect()
434        try:
435            cur = con.cursor()
436            self.executeDDL1(cur)
437            largs = [ ("Cooper's",) , ("Boag's",) ]
438            margs = [ {'beer': "Cooper's"}, {'beer': "Boag's"} ]
439            if self.driver.paramstyle == 'qmark':
440                cur.executemany(
441                    'insert into %sbooze values (?)' % self.table_prefix,
442                    largs
443                    )
444            elif self.driver.paramstyle == 'numeric':
445                cur.executemany(
446                    'insert into %sbooze values (:1)' % self.table_prefix,
447                    largs
448                    )
449            elif self.driver.paramstyle == 'named':
450                cur.executemany(
451                    'insert into %sbooze values (:beer)' % self.table_prefix,
452                    margs
453                    )
454            elif self.driver.paramstyle == 'format':
455                cur.executemany(
456                    'insert into %sbooze values (%%s)' % self.table_prefix,
457                    largs
458                    )
459            elif self.driver.paramstyle == 'pyformat':
460                cur.executemany(
461                    'insert into %sbooze values (%%(beer)s)' % (
462                        self.table_prefix
463                        ),
464                    margs
465                    )
466            else:
467                self.fail('Unknown paramstyle')
468            self.assertTrue(cur.rowcount in (-1,2),
469                'insert using cursor.executemany set cursor.rowcount to '
470                'incorrect value %r' % cur.rowcount
471                )
472            cur.execute('select name from %sbooze' % self.table_prefix)
473            res = cur.fetchall()
474            self.assertEqual(len(res),2,
475                'cursor.fetchall retrieved incorrect number of rows'
476                )
477            beers = [res[0][0],res[1][0]]
478            beers.sort()
479            self.assertEqual(beers[0],"Boag's",'incorrect data retrieved')
480            self.assertEqual(beers[1],"Cooper's",'incorrect data retrieved')
481        finally:
482            con.close()
483
484    def test_fetchone(self):
485        con = self._connect()
486        try:
487            cur = con.cursor()
488
489            # cursor.fetchone should raise an Error if called before
490            # executing a select-type query
491            self.assertRaises(self.driver.Error,cur.fetchone)
492
493            # cursor.fetchone should raise an Error if called after
494            # executing a query that cannnot return rows
495            self.executeDDL1(cur)
496            self.assertRaises(self.driver.Error,cur.fetchone)
497
498            cur.execute('select name from %sbooze' % self.table_prefix)
499            self.assertEqual(cur.fetchone(),None,
500                'cursor.fetchone should return None if a query retrieves '
501                'no rows'
502                )
503            self.assertTrue(cur.rowcount in (-1,0))
504
505            # cursor.fetchone should raise an Error if called after
506            # executing a query that cannnot return rows
507            cur.execute("insert into %sbooze values ('Victoria Bitter')" % (
508                self.table_prefix
509                ))
510            self.assertRaises(self.driver.Error,cur.fetchone)
511
512            cur.execute('select name from %sbooze' % self.table_prefix)
513            r = cur.fetchone()
514            self.assertEqual(len(r),1,
515                'cursor.fetchone should have retrieved a single row'
516                )
517            self.assertEqual(r[0],'Victoria Bitter',
518                'cursor.fetchone retrieved incorrect data'
519                )
520            self.assertEqual(cur.fetchone(),None,
521                'cursor.fetchone should return None if no more rows available'
522                )
523            self.assertTrue(cur.rowcount in (-1,1))
524        finally:
525            con.close()
526
527    samples = [
528        'Carlton Cold',
529        'Carlton Draft',
530        'Mountain Goat',
531        'Redback',
532        'Victoria Bitter',
533        'XXXX'
534        ]
535
536    def _populate(self):
537        """Return a list of sql commands to setup the DB for the fetch
538            tests.
539        """
540        populate = [
541            "insert into %sbooze values ('%s')" % (self.table_prefix,s)
542                for s in self.samples
543            ]
544        return populate
545
546    def test_fetchmany(self):
547        con = self._connect()
548        try:
549            cur = con.cursor()
550
551            # cursor.fetchmany should raise an Error if called without
552            #issuing a query
553            self.assertRaises(self.driver.Error,cur.fetchmany,4)
554
555            self.executeDDL1(cur)
556            for sql in self._populate():
557                cur.execute(sql)
558
559            cur.execute('select name from %sbooze' % self.table_prefix)
560            r = cur.fetchmany()
561            self.assertEqual(len(r),1,
562                'cursor.fetchmany retrieved incorrect number of rows, '
563                'default of arraysize is one.'
564                )
565            cur.arraysize=10
566            r = cur.fetchmany(3) # Should get 3 rows
567            self.assertEqual(len(r),3,
568                'cursor.fetchmany retrieved incorrect number of rows'
569                )
570            r = cur.fetchmany(4) # Should get 2 more
571            self.assertEqual(len(r),2,
572                'cursor.fetchmany retrieved incorrect number of rows'
573                )
574            r = cur.fetchmany(4) # Should be an empty sequence
575            self.assertEqual(len(r),0,
576                'cursor.fetchmany should return an empty sequence after '
577                'results are exhausted'
578            )
579            self.assertTrue(cur.rowcount in (-1,6))
580
581            # Same as above, using cursor.arraysize
582            cur.arraysize=4
583            cur.execute('select name from %sbooze' % self.table_prefix)
584            r = cur.fetchmany() # Should get 4 rows
585            self.assertEqual(len(r),4,
586                'cursor.arraysize not being honoured by fetchmany'
587                )
588            r = cur.fetchmany() # Should get 2 more
589            self.assertEqual(len(r),2)
590            r = cur.fetchmany() # Should be an empty sequence
591            self.assertEqual(len(r),0)
592            self.assertTrue(cur.rowcount in (-1,6))
593
594            cur.arraysize=6
595            cur.execute('select name from %sbooze' % self.table_prefix)
596            rows = cur.fetchmany() # Should get all rows
597            self.assertTrue(cur.rowcount in (-1,6))
598            self.assertEqual(len(rows),6)
599            self.assertEqual(len(rows),6)
600            rows = [r[0] for r in rows]
601            rows.sort()
602
603            # Make sure we get the right data back out
604            for i in range(0,6):
605                self.assertEqual(rows[i],self.samples[i],
606                    'incorrect data retrieved by cursor.fetchmany'
607                    )
608
609            rows = cur.fetchmany() # Should return an empty list
610            self.assertEqual(len(rows),0,
611                'cursor.fetchmany should return an empty sequence if '
612                'called after the whole result set has been fetched'
613                )
614            self.assertTrue(cur.rowcount in (-1,6))
615
616            self.executeDDL2(cur)
617            cur.execute('select name from %sbarflys' % self.table_prefix)
618            r = cur.fetchmany() # Should get empty sequence
619            self.assertEqual(len(r),0,
620                'cursor.fetchmany should return an empty sequence if '
621                'query retrieved no rows'
622                )
623            self.assertTrue(cur.rowcount in (-1,0))
624
625        finally:
626            con.close()
627
628    def test_fetchall(self):
629        con = self._connect()
630        try:
631            cur = con.cursor()
632            # cursor.fetchall should raise an Error if called
633            # without executing a query that may return rows (such
634            # as a select)
635            self.assertRaises(self.driver.Error, cur.fetchall)
636
637            self.executeDDL1(cur)
638            for sql in self._populate():
639                cur.execute(sql)
640
641            # cursor.fetchall should raise an Error if called
642            # after executing a a statement that cannot return rows
643            self.assertRaises(self.driver.Error,cur.fetchall)
644
645            cur.execute('select name from %sbooze' % self.table_prefix)
646            rows = cur.fetchall()
647            self.assertTrue(cur.rowcount in (-1,len(self.samples)))
648            self.assertEqual(len(rows),len(self.samples),
649                'cursor.fetchall did not retrieve all rows'
650                )
651            rows = [r[0] for r in rows]
652            rows.sort()
653            for i in range(0,len(self.samples)):
654                self.assertEqual(rows[i],self.samples[i],
655                'cursor.fetchall retrieved incorrect rows'
656                )
657            rows = cur.fetchall()
658            self.assertEqual(
659                len(rows),0,
660                'cursor.fetchall should return an empty list if called '
661                'after the whole result set has been fetched'
662                )
663            self.assertTrue(cur.rowcount in (-1,len(self.samples)))
664
665            self.executeDDL2(cur)
666            cur.execute('select name from %sbarflys' % self.table_prefix)
667            rows = cur.fetchall()
668            self.assertTrue(cur.rowcount in (-1,0))
669            self.assertEqual(len(rows),0,
670                'cursor.fetchall should return an empty list if '
671                'a select query returns no rows'
672                )
673
674        finally:
675            con.close()
676
677    def test_mixedfetch(self):
678        con = self._connect()
679        try:
680            cur = con.cursor()
681            self.executeDDL1(cur)
682            for sql in self._populate():
683                cur.execute(sql)
684
685            cur.execute('select name from %sbooze' % self.table_prefix)
686            rows1  = cur.fetchone()
687            rows23 = cur.fetchmany(2)
688            rows4  = cur.fetchone()
689            rows56 = cur.fetchall()
690            self.assertTrue(cur.rowcount in (-1,6))
691            self.assertEqual(len(rows23),2,
692                'fetchmany returned incorrect number of rows'
693                )
694            self.assertEqual(len(rows56),2,
695                'fetchall returned incorrect number of rows'
696                )
697
698            rows = [rows1[0]]
699            rows.extend([rows23[0][0],rows23[1][0]])
700            rows.append(rows4[0])
701            rows.extend([rows56[0][0],rows56[1][0]])
702            rows.sort()
703            for i in range(0,len(self.samples)):
704                self.assertEqual(rows[i],self.samples[i],
705                    'incorrect data retrieved or inserted'
706                    )
707        finally:
708            con.close()
709
710    def help_nextset_setUp(self, cur):
711        """Should create a procedure called deleteme
712            that returns two result sets, first the
713            number of rows in booze then "name from booze"
714        """
715        if False:
716            sql = """
717                create procedure deleteme as
718                begin
719                    select count(*) from booze
720                    select name from booze
721                end
722            """
723            cur.execute(sql)
724        else:
725            raise NotImplementedError('Helper not implemented')
726
727    def help_nextset_tearDown(self, cur):
728        """If cleaning up is needed after nextSetTest"""
729        if False:
730            cur.execute("drop procedure deleteme")
731        else:
732
733            raise NotImplementedError('Helper not implemented')
734
735    def test_nextset(self):
736        con = self._connect()
737        try:
738            cur = con.cursor()
739            if not hasattr(cur,'nextset'):
740                return
741
742            try:
743                self.executeDDL1(cur)
744                sql=self._populate()
745                for sql in self._populate():
746                    cur.execute(sql)
747
748                self.help_nextset_setUp(cur)
749
750                cur.callproc('deleteme')
751                numberofrows=cur.fetchone()
752                assert numberofrows[0]== len(self.samples)
753                assert cur.nextset()
754                names=cur.fetchall()
755                assert len(names) == len(self.samples)
756                s=cur.nextset()
757                assert s == None,'No more return sets, should return None'
758            finally:
759                self.help_nextset_tearDown(cur)
760
761        finally:
762            con.close()
763
764    def test_arraysize(self):
765        """Not much here - rest of the tests for this are in test_fetchmany"""
766        con = self._connect()
767        try:
768            cur = con.cursor()
769            self.assertTrue(hasattr(cur,'arraysize'),
770                'cursor.arraysize must be defined'
771                )
772        finally:
773            con.close()
774
775    def test_setinputsizes(self):
776        con = self._connect()
777        try:
778            cur = con.cursor()
779            cur.setinputsizes( (25,) )
780            self._paraminsert(cur) # Make sure cursor still works
781        finally:
782            con.close()
783
784    def test_setoutputsize_basic(self):
785        """Basic test is to make sure setoutputsize doesn't blow up"""
786        con = self._connect()
787        try:
788            cur = con.cursor()
789            cur.setoutputsize(1000)
790            cur.setoutputsize(2000,0)
791            self._paraminsert(cur) # Make sure the cursor still works
792        finally:
793            con.close()
794
795    def test_setoutputsize(self):
796        """Real test for setoutputsize is driver dependant"""
797        raise NotImplementedError('Driver needs to override this test')
798
799    def test_None(self):
800        con = self._connect()
801        try:
802            cur = con.cursor()
803            self.executeDDL1(cur)
804            cur.execute('insert into %sbooze values (NULL)' % self.table_prefix)
805            cur.execute('select name from %sbooze' % self.table_prefix)
806            r = cur.fetchall()
807            self.assertEqual(len(r),1)
808            self.assertEqual(len(r[0]),1)
809            self.assertEqual(r[0][0],None,'NULL value not returned as None')
810        finally:
811            con.close()
812
813    def test_Date(self):
814        d1 = self.driver.Date(2002,12,25)
815        d2 = self.driver.DateFromTicks(time.mktime((2002,12,25,0,0,0,0,0,0)))
816        # Can we assume this? API doesn't specify, but it seems implied
817        # self.assertEqual(str(d1),str(d2))
818
819    def test_Time(self):
820        t1 = self.driver.Time(13,45,30)
821        t2 = self.driver.TimeFromTicks(time.mktime((2001,1,1,13,45,30,0,0,0)))
822        # Can we assume this? API doesn't specify, but it seems implied
823        # self.assertEqual(str(t1),str(t2))
824
825    def test_Timestamp(self):
826        t1 = self.driver.Timestamp(2002,12,25,13,45,30)
827        t2 = self.driver.TimestampFromTicks(
828            time.mktime((2002,12,25,13,45,30,0,0,0))
829            )
830        # Can we assume this? API doesn't specify, but it seems implied
831        # self.assertEqual(str(t1),str(t2))
832
833    def test_Binary(self):
834        b = self.driver.Binary(b'Something')
835        b = self.driver.Binary(b'')
836
837    def test_STRING(self):
838        self.assertTrue(hasattr(self.driver,'STRING'),
839            'module.STRING must be defined'
840            )
841
842    def test_BINARY(self):
843        self.assertTrue(hasattr(self.driver,'BINARY'),
844            'module.BINARY must be defined.'
845            )
846
847    def test_NUMBER(self):
848        self.assertTrue(hasattr(self.driver,'NUMBER'),
849            'module.NUMBER must be defined.'
850            )
851
852    def test_DATETIME(self):
853        self.assertTrue(hasattr(self.driver,'DATETIME'),
854            'module.DATETIME must be defined.'
855            )
856
857    def test_ROWID(self):
858        self.assertTrue(hasattr(self.driver,'ROWID'),
859            'module.ROWID must be defined.'
860            )
861
Note: See TracBrowser for help on using the repository browser.