[DB-SIG] DB-API unittests

[Andrew Chambers]

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
     Singletons.

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
		         union
			 select 2, 'two'
			 union
			 select 3, 'three'
		"""
		required_rows = 3
		required_desc = [('n', 'int4', 4, None, None, None, None),
                                 ('t', 'text', -1, None, None, None, None)]
		self.curs.execute(sql)
		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
		         union
			 select %i, 'two'
			 union
			 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(create)
		self.curs.execute(insert, (1, "Any'town", Date(2005, 11, 11)))
		self.curs.execute(destroy)
	
	def testProgrammingError(self):
		"""Attempt to create identical tables should raise ProgrammingError
		"""
		create = """create table dup (
			a	integer	
		);
		"""
		self.curs.execute(create)
		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(create)
		self.curs.execute(insert, (1, 'First Order'))
		self.assertRaises(IntegrityError,
		                  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(create_1)
		self.curs.execute(insert_ord, (1, "First Order"))
		self.conn.commit()
		self.curs.execute(create_2)
		self.conn.commit()
		self.assertRaises(IntegrityError, 
		                  self.curs.execute, insert_ord_line, (2, 2, 2))
		self.conn.commit()
		self.curs.execute("drop table order cascade;")
		self.curs.execute("drop table order_item;")
		self.conn.commit()

	
	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):
		self.__connect()	
		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
					self.__close()
					return
				


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

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

if __name__ == '__main__':
	unittest.main()