[DB-SIG] DB-API unittests

Andrew Chambers andychambers2002 at yahoo.co.uk
Thu Dec 1 03:22:52 CET 2005

Hi All,

I've been doing some work on some unittests for the Python DB-API.

So far, I've created tests for...

   - Attempt to connect using invalid dbname should raise Error
   - Attempt to divide by zero should raise DataError
   - Attempt to violate foreign/primery key should raise IntegrityError
   - Attempt to create identical tables should raise ProgrammingError
   - Connection object should have minimum set of methods
   - Cursor object should have minimum set of methods
   - Simple select query should result in expected description/rowcount
   - Type_codes in description attr should evaluate to corresponding

It only works for postgresql drivers since it connects to template1
which should be present on most postgresql databases 

It raises a few points (these may have been discussed here before but
googling the archives didn't reveal them.

- Evidently there is ambiguity about the way that dates should be entered.
  pgdb - curs.execute("select '%s'" % Date(2005, 01, 01))
  psycopg - curs.execute("select %s" % Date(2005, 01, 01))
  Don't know about the others - I don't have them installed

- Exceptions.  My interpretation of the API seems to deviate from that
  of psycopg and pgdb.  Does anyone have any views on which is correct?
  I wrote the tests without having any existing implementation in mind 
  just going from the spec.  Are the tests at fault?

I'd also be interested in what else you think should be tested.  I've
started to write some performance tests that simply query the pg_type
table for a fixed time period (30 secs).  These need more work though.
I'll post if anyone asks.

Finally, I couldn't think of a way to run these tests across a number of
modules without duplicating the code (although to be honest I haven't
given it much thought yet.  If anyone has any ideas I'd be interested to 
hear those too.

#!/usr/bin/env python

import unittest
import time
from <db_module_of_your_choice> import *

class ConnectionTestCase(unittest.TestCase):

	def testConnection(self):
		"""Connection Object returned should have minimum operations"""
		conn = connect(database="template1")
		self.assert_(set(('close', 'commit', 
		              'cursor', 'rollback')).issubset(set(dir(conn))))

	def testConnectFail(self):
		"""Should raise error on connecting with bad parameters"""
		self.assertRaises(DatabaseError, connect, database="fun'y name")

class CursorTestCase(unittest.TestCase):

	def setUp(self):
		self.conn = connect(database="template1")
		self.curs = self.conn.cursor()
	def tearDown(self):
		del self.curs
		del self.conn
	def testHasAttrs(self):
		"""Cursor object should have listed attributes\n
		self.curs.execute("select 1, 2;")
		required = ('arraysize', 'callproc', 'close', 
		            'description', 'execute', 'executemany', 
			    'fetchall', 'fetchmany', 'fetchone', 
			    'rowcount', 'setinputsizes', 'setoutputsize')
		actual   = (dir(self.curs))
		diff     = set(required).difference(set(actual))
		self.assert_(len(diff) == 0,
			"Absentees: %s" % diff)

	def testSelect(self):
		"""Data returned should be as expected
		For the query below, 
		rowcount = 4
		description = [(n, int4, 4, None, None, None, None), 
		               (t, 'text', -1, None, None, None, None)]
		sql = """select 1 as n, 'text' as t
			 select 2, 'two'
			 select 3, 'three'
		required_rows = 3
		required_desc = [('n', 'int4', 4, None, None, None, None),
                                 ('t', 'text', -1, None, None, None, None)]
		self.assert_(self.curs.rowcount == required_rows, 
		     "Actual rows: %s" % self.curs.rowcount)
		self.assert_(self.curs.description == required_desc,
		     "Actual description: %s" % self.curs.description)
	def testSelectParams(self):
		"""Query using parameters in simple pyformat
		sql = """select 1 as %s, '%s' as t
			 select %i, 'two'
			 select 3, 'three'
		self.curs.execute(sql, ('one', 'text', 2))
		required_rows = 3
		self.assert_(self.curs.rowcount == required_rows,
		     "Actual rows: %s" % self.curs.rowcount)
	def testSelectTypes(self):
		"""Description types should evaluate equal to module constants"""
		sql = """select '%(string)s', %(number)i, %(datetime)s"""
		self.curs.execute(sql, {'string' : "String", 
		                        'number' : 42, 
					'datetime' : Date(2005, 01, 01)})
		desc = self.curs.description
		self.assertEqual(desc[0][1], STRING, str(desc))
		self.assertEqual(desc[1][1], NUMBER)
		self.assertEqual(desc[2][1], DATETIME)

	def testInsertOK(self):
		"""Successfully insert valid data into test table
		create = """create table pygres_ins_test (
		              order_no  int  PRIMARY KEY,
			      supp_addr text,
			      order_date date
		insert = """insert into pygres_ins_test values (%i, '%s', %s)"""
		destroy = "drop table pygres_ins_test;"
		self.curs.execute(insert, (1, "Any'town", Date(2005, 11, 11)))
	def testProgrammingError(self):
		"""Attempt to create identical tables should raise ProgrammingError
		create = """create table dup (
			a	integer	
		self.assertRaises(ProgrammingError, self.curs.execute, create)
	def testIntegrityErrorPK(self):
		"""Attempt to violate primary key should raise IntegrityError
		create = """create table order (
			order_id	integer      PRIMARY KEY,
			order_desc	text
		insert = "insert into order values (%i, '%s')"
		self.curs.execute(insert, (1, 'First Order'))
		                  self.curs.execute, insert, (1, 'Second Order'))
	def testIntegrityErrorFK(self):
		"""Attempt to violate foreign key should raise IntegrityError"""
		create_1 = """
		        create table order (
			order_id	integer		PRIMARY KEY,
			order_desc	text
		create_2 = """
			create table order_item (
			order_id	integer REFERENCES order(order_id),
			product_id	integer,
			quantity	numeric,
			PRIMARY KEY(order_id, product_id)
		insert_ord = "insert into order values (%i, '%s')"
		insert_ord_line = "insert into order_item values (%i, %i, %i)"
		self.curs.execute(insert_ord, (1, "First Order"))
		                  self.curs.execute, insert_ord_line, (2, 2, 2))
		self.curs.execute("drop table order cascade;")
		self.curs.execute("drop table order_item;")

	def testDataError(self):
		"""Attempt to divide by zero should raise DataError
		sql = "select (10 / 0)"
		self.assertRaises(DataError, self.curs.execute, sql)
class PerformanceTestCase(unittest.TestCase):

	def __connect(self):
		self.conn = connect(database="template1")
		self.curs = self.conn.cursor()

	def __close(self):
		del self.curs
		del self.conn

	def testRetrievals(self):
		rows = 0
		start = time.time()
		while True:
			self.curs.execute("select * from pg_type")
			while True:
				row = self.curs.fetchone()
				rows += 1
				if time.time() - start > 30:
					print rows

def connectionSuite():
	"""Returns the Connection test suite"""
	return unittest.makeSuite(ConnectionTestCase)

def cursorSuite():
	return unittest.makeSuite(CursorTestCase)

if __name__ == '__main__':

More information about the DB-SIG mailing list