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