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()
_______________________________________________
DB-SIG maillist - DB-SIG@python.org
http://mail.python.org/mailman/listinfo/db-sig
