source: OpenRLabs-Git/deploy/rlabs-docker/web2py-rlabs/gluon/contrib/pymysql/tests/thirdparty/test_MySQLdb/dbapi20.py

main
Last change on this file was 42bd667, checked in by David Fuertes <dfuertes@…>, 4 years ago

Historial Limpio
  • Property mode set to 100755
File size: 30.7 KB
Line 
1#!/usr/bin/env 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$'
15__version__ = '$Revision$'[11:-2]
16__author__ = 'Stuart Bishop <zen@shangri-la.dropbear.id.au>'
17
18try:
19    import unittest2 as unittest
20except ImportError:
21    import unittest
22
23import time
24
25# $Log$
26# Revision 1.1.2.1  2006/02/25 03:44:32  adustman
27# Generic DB-API unit test module
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 heirarchy.
184        self.assertTrue(issubclass(self.driver.Warning,Exception))
185        self.assertTrue(issubclass(self.driver.Error,Exception))
186        self.assertTrue(
187            issubclass(self.driver.InterfaceError,self.driver.Error)
188            )
189        self.assertTrue(
190            issubclass(self.driver.DatabaseError,self.driver.Error)
191            )
192        self.assertTrue(
193            issubclass(self.driver.OperationalError,self.driver.Error)
194            )
195        self.assertTrue(
196            issubclass(self.driver.IntegrityError,self.driver.Error)
197            )
198        self.assertTrue(
199            issubclass(self.driver.InternalError,self.driver.Error)
200            )
201        self.assertTrue(
202            issubclass(self.driver.ProgrammingError,self.driver.Error)
203            )
204        self.assertTrue(
205            issubclass(self.driver.NotSupportedError,self.driver.Error)
206            )
207
208    def test_ExceptionsAsConnectionAttributes(self):
209        # OPTIONAL EXTENSION
210        # Test for the optional DB API 2.0 extension, where the exceptions
211        # are exposed as attributes on the Connection object
212        # I figure this optional extension will be implemented by any
213        # driver author who is using this test suite, so it is enabled
214        # by default.
215        con = self._connect()
216        drv = self.driver
217        self.assertTrue(con.Warning is drv.Warning)
218        self.assertTrue(con.Error is drv.Error)
219        self.assertTrue(con.InterfaceError is drv.InterfaceError)
220        self.assertTrue(con.DatabaseError is drv.DatabaseError)
221        self.assertTrue(con.OperationalError is drv.OperationalError)
222        self.assertTrue(con.IntegrityError is drv.IntegrityError)
223        self.assertTrue(con.InternalError is drv.InternalError)
224        self.assertTrue(con.ProgrammingError is drv.ProgrammingError)
225        self.assertTrue(con.NotSupportedError is drv.NotSupportedError)
226
227
228    def test_commit(self):
229        con = self._connect()
230        try:
231            # Commit must work, even if it doesn't do anything
232            con.commit()
233        finally:
234            con.close()
235
236    def test_rollback(self):
237        con = self._connect()
238        # If rollback is defined, it should either work or throw
239        # the documented exception
240        if hasattr(con,'rollback'):
241            try:
242                con.rollback()
243            except self.driver.NotSupportedError:
244                pass
245
246    def test_cursor(self):
247        con = self._connect()
248        try:
249            cur = con.cursor()
250        finally:
251            con.close()
252
253    def test_cursor_isolation(self):
254        con = self._connect()
255        try:
256            # Make sure cursors created from the same connection have
257            # the documented transaction isolation level
258            cur1 = con.cursor()
259            cur2 = con.cursor()
260            self.executeDDL1(cur1)
261            cur1.execute("insert into %sbooze values ('Victoria Bitter')" % (
262                self.table_prefix
263                ))
264            cur2.execute("select name from %sbooze" % self.table_prefix)
265            booze = cur2.fetchall()
266            self.assertEqual(len(booze),1)
267            self.assertEqual(len(booze[0]),1)
268            self.assertEqual(booze[0][0],'Victoria Bitter')
269        finally:
270            con.close()
271
272    def test_description(self):
273        con = self._connect()
274        try:
275            cur = con.cursor()
276            self.executeDDL1(cur)
277            self.assertEqual(cur.description,None,
278                'cursor.description should be none after executing a '
279                'statement that can return no rows (such as DDL)'
280                )
281            cur.execute('select name from %sbooze' % self.table_prefix)
282            self.assertEqual(len(cur.description),1,
283                'cursor.description describes too many columns'
284                )
285            self.assertEqual(len(cur.description[0]),7,
286                'cursor.description[x] tuples must have 7 elements'
287                )
288            self.assertEqual(cur.description[0][0].lower(),'name',
289                'cursor.description[x][0] must return column name'
290                )
291            self.assertEqual(cur.description[0][1],self.driver.STRING,
292                'cursor.description[x][1] must return column type. Got %r'
293                    % cur.description[0][1]
294                )
295
296            # Make sure self.description gets reset
297            self.executeDDL2(cur)
298            self.assertEqual(cur.description,None,
299                'cursor.description not being set to None when executing '
300                'no-result statements (eg. DDL)'
301                )
302        finally:
303            con.close()
304
305    def test_rowcount(self):
306        con = self._connect()
307        try:
308            cur = con.cursor()
309            self.executeDDL1(cur)
310            self.assertEqual(cur.rowcount,-1,
311                'cursor.rowcount should be -1 after executing no-result '
312                'statements'
313                )
314            cur.execute("insert into %sbooze values ('Victoria Bitter')" % (
315                self.table_prefix
316                ))
317            self.assertTrue(cur.rowcount in (-1,1),
318                'cursor.rowcount should == number or rows inserted, or '
319                'set to -1 after executing an insert statement'
320                )
321            cur.execute("select name from %sbooze" % self.table_prefix)
322            self.assertTrue(cur.rowcount in (-1,1),
323                'cursor.rowcount should == number of rows returned, or '
324                'set to -1 after executing a select statement'
325                )
326            self.executeDDL2(cur)
327            self.assertEqual(cur.rowcount,-1,
328                'cursor.rowcount not being reset to -1 after executing '
329                'no-result statements'
330                )
331        finally:
332            con.close()
333
334    lower_func = 'lower'
335    def test_callproc(self):
336        con = self._connect()
337        try:
338            cur = con.cursor()
339            if self.lower_func and hasattr(cur,'callproc'):
340                r = cur.callproc(self.lower_func,('FOO',))
341                self.assertEqual(len(r),1)
342                self.assertEqual(r[0],'FOO')
343                r = cur.fetchall()
344                self.assertEqual(len(r),1,'callproc produced no result set')
345                self.assertEqual(len(r[0]),1,
346                    'callproc produced invalid result set'
347                    )
348                self.assertEqual(r[0][0],'foo',
349                    'callproc produced invalid results'
350                    )
351        finally:
352            con.close()
353
354    def test_close(self):
355        con = self._connect()
356        try:
357            cur = con.cursor()
358        finally:
359            con.close()
360
361        # cursor.execute should raise an Error if called after connection
362        # closed
363        self.assertRaises(self.driver.Error,self.executeDDL1,cur)
364
365        # connection.commit should raise an Error if called after connection'
366        # closed.'
367        self.assertRaises(self.driver.Error,con.commit)
368
369        # connection.close should raise an Error if called more than once
370        self.assertRaises(self.driver.Error,con.close)
371
372    def test_execute(self):
373        con = self._connect()
374        try:
375            cur = con.cursor()
376            self._paraminsert(cur)
377        finally:
378            con.close()
379
380    def _paraminsert(self,cur):
381        self.executeDDL1(cur)
382        cur.execute("insert into %sbooze values ('Victoria Bitter')" % (
383            self.table_prefix
384            ))
385        self.assertTrue(cur.rowcount in (-1,1))
386
387        if self.driver.paramstyle == 'qmark':
388            cur.execute(
389                'insert into %sbooze values (?)' % self.table_prefix,
390                ("Cooper's",)
391                )
392        elif self.driver.paramstyle == 'numeric':
393            cur.execute(
394                'insert into %sbooze values (:1)' % self.table_prefix,
395                ("Cooper's",)
396                )
397        elif self.driver.paramstyle == 'named':
398            cur.execute(
399                'insert into %sbooze values (:beer)' % self.table_prefix,
400                {'beer':"Cooper's"}
401                )
402        elif self.driver.paramstyle == 'format':
403            cur.execute(
404                'insert into %sbooze values (%%s)' % self.table_prefix,
405                ("Cooper's",)
406                )
407        elif self.driver.paramstyle == 'pyformat':
408            cur.execute(
409                'insert into %sbooze values (%%(beer)s)' % self.table_prefix,
410                {'beer':"Cooper's"}
411                )
412        else:
413            self.fail('Invalid paramstyle')
414        self.assertTrue(cur.rowcount in (-1,1))
415
416        cur.execute('select name from %sbooze' % self.table_prefix)
417        res = cur.fetchall()
418        self.assertEqual(len(res),2,'cursor.fetchall returned too few rows')
419        beers = [res[0][0],res[1][0]]
420        beers.sort()
421        self.assertEqual(beers[0],"Cooper's",
422            'cursor.fetchall retrieved incorrect data, or data inserted '
423            'incorrectly'
424            )
425        self.assertEqual(beers[1],"Victoria Bitter",
426            'cursor.fetchall retrieved incorrect data, or data inserted '
427            'incorrectly'
428            )
429
430    def test_executemany(self):
431        con = self._connect()
432        try:
433            cur = con.cursor()
434            self.executeDDL1(cur)
435            largs = [ ("Cooper's",) , ("Boag's",) ]
436            margs = [ {'beer': "Cooper's"}, {'beer': "Boag's"} ]
437            if self.driver.paramstyle == 'qmark':
438                cur.executemany(
439                    'insert into %sbooze values (?)' % self.table_prefix,
440                    largs
441                    )
442            elif self.driver.paramstyle == 'numeric':
443                cur.executemany(
444                    'insert into %sbooze values (:1)' % self.table_prefix,
445                    largs
446                    )
447            elif self.driver.paramstyle == 'named':
448                cur.executemany(
449                    'insert into %sbooze values (:beer)' % self.table_prefix,
450                    margs
451                    )
452            elif self.driver.paramstyle == 'format':
453                cur.executemany(
454                    'insert into %sbooze values (%%s)' % self.table_prefix,
455                    largs
456                    )
457            elif self.driver.paramstyle == 'pyformat':
458                cur.executemany(
459                    'insert into %sbooze values (%%(beer)s)' % (
460                        self.table_prefix
461                        ),
462                    margs
463                    )
464            else:
465                self.fail('Unknown paramstyle')
466            self.assertTrue(cur.rowcount in (-1,2),
467                'insert using cursor.executemany set cursor.rowcount to '
468                'incorrect value %r' % cur.rowcount
469                )
470            cur.execute('select name from %sbooze' % self.table_prefix)
471            res = cur.fetchall()
472            self.assertEqual(len(res),2,
473                'cursor.fetchall retrieved incorrect number of rows'
474                )
475            beers = [res[0][0],res[1][0]]
476            beers.sort()
477            self.assertEqual(beers[0],"Boag's",'incorrect data retrieved')
478            self.assertEqual(beers[1],"Cooper's",'incorrect data retrieved')
479        finally:
480            con.close()
481
482    def test_fetchone(self):
483        con = self._connect()
484        try:
485            cur = con.cursor()
486
487            # cursor.fetchone should raise an Error if called before
488            # executing a select-type query
489            self.assertRaises(self.driver.Error,cur.fetchone)
490
491            # cursor.fetchone should raise an Error if called after
492            # executing a query that cannnot return rows
493            self.executeDDL1(cur)
494            self.assertRaises(self.driver.Error,cur.fetchone)
495
496            cur.execute('select name from %sbooze' % self.table_prefix)
497            self.assertEqual(cur.fetchone(),None,
498                'cursor.fetchone should return None if a query retrieves '
499                'no rows'
500                )
501            self.assertTrue(cur.rowcount in (-1,0))
502
503            # cursor.fetchone should raise an Error if called after
504            # executing a query that cannnot return rows
505            cur.execute("insert into %sbooze values ('Victoria Bitter')" % (
506                self.table_prefix
507                ))
508            self.assertRaises(self.driver.Error,cur.fetchone)
509
510            cur.execute('select name from %sbooze' % self.table_prefix)
511            r = cur.fetchone()
512            self.assertEqual(len(r),1,
513                'cursor.fetchone should have retrieved a single row'
514                )
515            self.assertEqual(r[0],'Victoria Bitter',
516                'cursor.fetchone retrieved incorrect data'
517                )
518            self.assertEqual(cur.fetchone(),None,
519                'cursor.fetchone should return None if no more rows available'
520                )
521            self.assertTrue(cur.rowcount in (-1,1))
522        finally:
523            con.close()
524
525    samples = [
526        'Carlton Cold',
527        'Carlton Draft',
528        'Mountain Goat',
529        'Redback',
530        'Victoria Bitter',
531        'XXXX'
532        ]
533
534    def _populate(self):
535        ''' Return a list of sql commands to setup the DB for the fetch
536            tests.
537        '''
538        populate = [
539            "insert into %sbooze values ('%s')" % (self.table_prefix,s)
540                for s in self.samples
541            ]
542        return populate
543
544    def test_fetchmany(self):
545        con = self._connect()
546        try:
547            cur = con.cursor()
548
549            # cursor.fetchmany should raise an Error if called without
550            #issuing a query
551            self.assertRaises(self.driver.Error,cur.fetchmany,4)
552
553            self.executeDDL1(cur)
554            for sql in self._populate():
555                cur.execute(sql)
556
557            cur.execute('select name from %sbooze' % self.table_prefix)
558            r = cur.fetchmany()
559            self.assertEqual(len(r),1,
560                'cursor.fetchmany retrieved incorrect number of rows, '
561                'default of arraysize is one.'
562                )
563            cur.arraysize=10
564            r = cur.fetchmany(3) # Should get 3 rows
565            self.assertEqual(len(r),3,
566                'cursor.fetchmany retrieved incorrect number of rows'
567                )
568            r = cur.fetchmany(4) # Should get 2 more
569            self.assertEqual(len(r),2,
570                'cursor.fetchmany retrieved incorrect number of rows'
571                )
572            r = cur.fetchmany(4) # Should be an empty sequence
573            self.assertEqual(len(r),0,
574                'cursor.fetchmany should return an empty sequence after '
575                'results are exhausted'
576            )
577            self.assertTrue(cur.rowcount in (-1,6))
578
579            # Same as above, using cursor.arraysize
580            cur.arraysize=4
581            cur.execute('select name from %sbooze' % self.table_prefix)
582            r = cur.fetchmany() # Should get 4 rows
583            self.assertEqual(len(r),4,
584                'cursor.arraysize not being honoured by fetchmany'
585                )
586            r = cur.fetchmany() # Should get 2 more
587            self.assertEqual(len(r),2)
588            r = cur.fetchmany() # Should be an empty sequence
589            self.assertEqual(len(r),0)
590            self.assertTrue(cur.rowcount in (-1,6))
591
592            cur.arraysize=6
593            cur.execute('select name from %sbooze' % self.table_prefix)
594            rows = cur.fetchmany() # Should get all rows
595            self.assertTrue(cur.rowcount in (-1,6))
596            self.assertEqual(len(rows),6)
597            self.assertEqual(len(rows),6)
598            rows = [r[0] for r in rows]
599            rows.sort()
600
601            # Make sure we get the right data back out
602            for i in range(0,6):
603                self.assertEqual(rows[i],self.samples[i],
604                    'incorrect data retrieved by cursor.fetchmany'
605                    )
606
607            rows = cur.fetchmany() # Should return an empty list
608            self.assertEqual(len(rows),0,
609                'cursor.fetchmany should return an empty sequence if '
610                'called after the whole result set has been fetched'
611                )
612            self.assertTrue(cur.rowcount in (-1,6))
613
614            self.executeDDL2(cur)
615            cur.execute('select name from %sbarflys' % self.table_prefix)
616            r = cur.fetchmany() # Should get empty sequence
617            self.assertEqual(len(r),0,
618                'cursor.fetchmany should return an empty sequence if '
619                'query retrieved no rows'
620                )
621            self.assertTrue(cur.rowcount in (-1,0))
622
623        finally:
624            con.close()
625
626    def test_fetchall(self):
627        con = self._connect()
628        try:
629            cur = con.cursor()
630            # cursor.fetchall should raise an Error if called
631            # without executing a query that may return rows (such
632            # as a select)
633            self.assertRaises(self.driver.Error, cur.fetchall)
634
635            self.executeDDL1(cur)
636            for sql in self._populate():
637                cur.execute(sql)
638
639            # cursor.fetchall should raise an Error if called
640            # after executing a a statement that cannot return rows
641            self.assertRaises(self.driver.Error,cur.fetchall)
642
643            cur.execute('select name from %sbooze' % self.table_prefix)
644            rows = cur.fetchall()
645            self.assertTrue(cur.rowcount in (-1,len(self.samples)))
646            self.assertEqual(len(rows),len(self.samples),
647                'cursor.fetchall did not retrieve all rows'
648                )
649            rows = [r[0] for r in rows]
650            rows.sort()
651            for i in range(0,len(self.samples)):
652                self.assertEqual(rows[i],self.samples[i],
653                'cursor.fetchall retrieved incorrect rows'
654                )
655            rows = cur.fetchall()
656            self.assertEqual(
657                len(rows),0,
658                'cursor.fetchall should return an empty list if called '
659                'after the whole result set has been fetched'
660                )
661            self.assertTrue(cur.rowcount in (-1,len(self.samples)))
662
663            self.executeDDL2(cur)
664            cur.execute('select name from %sbarflys' % self.table_prefix)
665            rows = cur.fetchall()
666            self.assertTrue(cur.rowcount in (-1,0))
667            self.assertEqual(len(rows),0,
668                'cursor.fetchall should return an empty list if '
669                'a select query returns no rows'
670                )
671
672        finally:
673            con.close()
674
675    def test_mixedfetch(self):
676        con = self._connect()
677        try:
678            cur = con.cursor()
679            self.executeDDL1(cur)
680            for sql in self._populate():
681                cur.execute(sql)
682
683            cur.execute('select name from %sbooze' % self.table_prefix)
684            rows1  = cur.fetchone()
685            rows23 = cur.fetchmany(2)
686            rows4  = cur.fetchone()
687            rows56 = cur.fetchall()
688            self.assertTrue(cur.rowcount in (-1,6))
689            self.assertEqual(len(rows23),2,
690                'fetchmany returned incorrect number of rows'
691                )
692            self.assertEqual(len(rows56),2,
693                'fetchall returned incorrect number of rows'
694                )
695
696            rows = [rows1[0]]
697            rows.extend([rows23[0][0],rows23[1][0]])
698            rows.append(rows4[0])
699            rows.extend([rows56[0][0],rows56[1][0]])
700            rows.sort()
701            for i in range(0,len(self.samples)):
702                self.assertEqual(rows[i],self.samples[i],
703                    'incorrect data retrieved or inserted'
704                    )
705        finally:
706            con.close()
707
708    def help_nextset_setUp(self,cur):
709        ''' Should create a procedure called deleteme
710            that returns two result sets, first the
711            number of rows in booze then "name from booze"
712        '''
713        raise NotImplementedError('Helper not implemented')
714        #sql="""
715        #    create procedure deleteme as
716        #    begin
717        #        select count(*) from booze
718        #        select name from booze
719        #    end
720        #"""
721        #cur.execute(sql)
722
723    def help_nextset_tearDown(self,cur):
724        'If cleaning up is needed after nextSetTest'
725        raise NotImplementedError('Helper not implemented')
726        #cur.execute("drop procedure deleteme")
727
728    def test_nextset(self):
729        con = self._connect()
730        try:
731            cur = con.cursor()
732            if not hasattr(cur,'nextset'):
733                return
734
735            try:
736                self.executeDDL1(cur)
737                sql=self._populate()
738                for sql in self._populate():
739                    cur.execute(sql)
740
741                self.help_nextset_setUp(cur)
742
743                cur.callproc('deleteme')
744                numberofrows=cur.fetchone()
745                assert numberofrows[0]== len(self.samples)
746                assert cur.nextset()
747                names=cur.fetchall()
748                assert len(names) == len(self.samples)
749                s=cur.nextset()
750                assert s == None,'No more return sets, should return None'
751            finally:
752                self.help_nextset_tearDown(cur)
753
754        finally:
755            con.close()
756
757    def test_nextset(self):
758        raise NotImplementedError('Drivers need to override this test')
759
760    def test_arraysize(self):
761        # Not much here - rest of the tests for this are in test_fetchmany
762        con = self._connect()
763        try:
764            cur = con.cursor()
765            self.assertTrue(hasattr(cur,'arraysize'),
766                'cursor.arraysize must be defined'
767                )
768        finally:
769            con.close()
770
771    def test_setinputsizes(self):
772        con = self._connect()
773        try:
774            cur = con.cursor()
775            cur.setinputsizes( (25,) )
776            self._paraminsert(cur) # Make sure cursor still works
777        finally:
778            con.close()
779
780    def test_setoutputsize_basic(self):
781        # Basic test is to make sure setoutputsize doesn't blow up
782        con = self._connect()
783        try:
784            cur = con.cursor()
785            cur.setoutputsize(1000)
786            cur.setoutputsize(2000,0)
787            self._paraminsert(cur) # Make sure the cursor still works
788        finally:
789            con.close()
790
791    def test_setoutputsize(self):
792        # Real test for setoutputsize is driver dependant
793        raise NotImplementedError('Driver need to override this test')
794
795    def test_None(self):
796        con = self._connect()
797        try:
798            cur = con.cursor()
799            self.executeDDL1(cur)
800            cur.execute('insert into %sbooze values (NULL)' % self.table_prefix)
801            cur.execute('select name from %sbooze' % self.table_prefix)
802            r = cur.fetchall()
803            self.assertEqual(len(r),1)
804            self.assertEqual(len(r[0]),1)
805            self.assertEqual(r[0][0],None,'NULL value not returned as None')
806        finally:
807            con.close()
808
809    def test_Date(self):
810        d1 = self.driver.Date(2002,12,25)
811        d2 = self.driver.DateFromTicks(time.mktime((2002,12,25,0,0,0,0,0,0)))
812        # Can we assume this? API doesn't specify, but it seems implied
813        # self.assertEqual(str(d1),str(d2))
814
815    def test_Time(self):
816        t1 = self.driver.Time(13,45,30)
817        t2 = self.driver.TimeFromTicks(time.mktime((2001,1,1,13,45,30,0,0,0)))
818        # Can we assume this? API doesn't specify, but it seems implied
819        # self.assertEqual(str(t1),str(t2))
820
821    def test_Timestamp(self):
822        t1 = self.driver.Timestamp(2002,12,25,13,45,30)
823        t2 = self.driver.TimestampFromTicks(
824            time.mktime((2002,12,25,13,45,30,0,0,0))
825            )
826        # Can we assume this? API doesn't specify, but it seems implied
827        # self.assertEqual(str(t1),str(t2))
828
829    def test_Binary(self):
830        b = self.driver.Binary(b'Something')
831        b = self.driver.Binary(b'')
832
833    def test_STRING(self):
834        self.assertTrue(hasattr(self.driver,'STRING'),
835            'module.STRING must be defined'
836            )
837
838    def test_BINARY(self):
839        self.assertTrue(hasattr(self.driver,'BINARY'),
840            'module.BINARY must be defined.'
841            )
842
843    def test_NUMBER(self):
844        self.assertTrue(hasattr(self.driver,'NUMBER'),
845            'module.NUMBER must be defined.'
846            )
847
848    def test_DATETIME(self):
849        self.assertTrue(hasattr(self.driver,'DATETIME'),
850            'module.DATETIME must be defined.'
851            )
852
853    def test_ROWID(self):
854        self.assertTrue(hasattr(self.driver,'ROWID'),
855            'module.ROWID must be defined.'
856            )
Note: See TracBrowser for help on using the repository browser.