Menu
▾
▴
[Pypgsql-checkins] CVS: pypgsql/test/regression pgconnection.py,NONE,1.1
|
From: Billy G. A. <bal...@us...> - 2001-09-30 06:34:08
|
Update of /cvsroot/pypgsql/pypgsql/test/regression
In directory usw-pr-cvs1:/tmp/cvs-serv32391/test/regression
Added Files:
pgconnection.py
Log Message:
29SEP2001 bga Fixed numerous bugs found during the development of the regres0
sion test cases for pgconnection.c.
--- The PgConnection's attributes are now set to a value of None
when finish() is called.
--- Added regression tests for the pgconnection.c module.
--- NEW FILE: pgconnection.py ---
#!/usr/local/bin/python
#ident "@(#) $Id: pgconnection.py,v 1.1 2001/09/30 06:34:04 ballie01 Exp $"
# vi:set sw=4 ts=8 showmode ai:
#-----------------------------------------------------------------------+
# Name: pgconnection.py |
# |
# Description: pgconnection.py contains (white box) regression test |
# cases for the PgConnection object. |
# |
# Note: These test cases requires that a test database named |
# pypgsql exists and that the person running the test has |
# full rights to the database. |
#=======================================================================|
# Copyright 2001 by Billy G. Allie. |
# All rights reserved. |
# |
# Permission to use, copy, modify, and distribute this software and its |
# documentation for any purpose and without fee is hereby granted, pro- |
# vided that the above copyright notice appear in all copies and that |
# both that copyright notice and this permission notice appear in sup- |
# porting documentation, and that the copyright owner's name not be |
# used in advertising or publicity pertaining to distribution of the |
# software without specific, written prior permission. |
# |
# THE AUTHOR(S) DISCLAIMS ALL WARRANTIES WITH REGARD TO THIS SOFTWARE, |
# INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS. IN |
# NO EVENT SHALL THE AUTHOR(S) BE LIABLE FOR ANY SPECIAL, INDIRECT OR |
# CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS |
# OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE |
# OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE |
# USE OR PERFORMANCE OF THIS SOFTWARE. |
#=======================================================================|
# Revision History: |
# |
# Date Ini Description |
# --------- --- ------------------------------------------------------- |
# 17SEP2001 bga Initial release by Billy G. Allie. |
#-----------------------------------------------------------------------+
import sys
import unittest
import select
import types
import string
import libpq
version = sys.version_info
version = ((((version[0] * 100) + version[1]) * 100) + version[2])
# Get the default values for connection variables.
conndefs = {}
defs = libpq.PQconndefaults()
for i in defs:
conndefs[i[0]] = i[3]
del defs
del i
if conndefs['host'] == None:
conndefs['host'] = 'localhost'
conndefs['port'] = int(conndefs['port'])
members = ["host", "port", "db", "tty", "user", "password", "backendPID",
"socket", "notices", "version", "status", "errorMessage", "isBusy",
"isnonblocking"]
memtypes =[types.StringType, types.IntType, types.StringType, types.StringType,
types.StringType, types.StringType, types.IntType, types.IntType,
types.ListType, libpq.PgVersionType, types.IntType, types.StringType,
types.IntType, types.IntType]
methods = ["consumeInput", "endcopy", "finish", "getResult", "getline",
"getlineAsync", "lo_creat", "lo_import", "lo_unlink", "notifies",
"putline", "query", "requestCancel", "reset", "sendQuery", "trace",
"untrace", "connectPoll", "flush", "setnonblocking"]
# Get a connection and a version string to be used globally.
cnx = libpq.PQconnectdb("dbname=pypgsql")
vstr = "%(major)d.%(minor)d.%(level)d" % cnx.version
class PgConnectionTestCases(unittest.TestCase):
def setUp(self):
global cnx, vstr, conndefs, members, memtypes, methods
self.cnx = cnx
self.vstr = vstr
self.cdef = conndefs
self.members = members
self.memtypes = memtypes
self.methods = methods
def CheckMemberValues(self, cnx, expected):
for i in range(len(self.members)):
exec 'v = cnx.%s' % self.members[i]
self.assertEquals(v, expected[i],
'PgConnection.%s is %s, it should be %s!' % \
(self.members[i], v, expected[i]))
def CheckForMethods(self, cnx):
for i in self.methods:
self.assert_(i in cnx.__methods__,
'PgConnection is missing method "%s".' % i)
exec 'm = cnx.%s' % i
self.assert_(callable(m),
'PgConnection method %s() is not callable.' % i)
def CheckMemberTypes(self, cnx, expected):
for i in range(len(self.members)):
exec 'v = cnx.%s' % self.members[i]
self.assertEquals(type(v), expected[i],
'PgConnection.%s is %s, it should be %s!' % \
(self.members[i], type(v), expected[i]))
def CheckFinish(self):
"""
Test finish()
"""
try:
cnx.finish()
expected = [None, None, None, None, None, None, None, None, None,
None, None, None, None, None]
self.CheckMemberValues(cnx, expected)
except Exception, msg:
self.fail(msg)
def CheckCopyOutAsyncProcessing(self):
"""
Test getlineAsync(), endcopy()
"""
try:
res = cnx.query("copy copytest to stdin")
self.assertEquals(res.resultStatus, libpq.COPY_OUT,
"Copy Out command did not return expect results.")
out = []
inp = ''
while inp != '\\.':
inp = cnx.getlineAsync()
if inp != None:
out.append(inp)
cnx.endcopy()
self.assertEquals(out[0], '3\tHello World\t4.5',
"Copy Out did not work")
self.assertEquals(out[1], '4\tGoodbye World\t7.11',
"Copy Out did not work")
del res
except Exception, msg:
self.fail(msg)
def CheckCopyOutProcessing(self):
"""
Test getline(), endcopy()
"""
try:
res = cnx.query("copy copytest to stdin")
self.assertEquals(res.resultStatus, libpq.COPY_OUT,
"Copy Out command did not return expect results.")
out = []
inp = ''
while inp != '\\.':
inp = cnx.getline()
out.append(inp)
cnx.endcopy()
self.assertEquals(out[0], '3\tHello World\t4.5',
"Copy Out did not work")
self.assertEquals(out[1], '4\tGoodbye World\t7.11',
"Copy Out did not work")
del res
except Exception, msg:
self.fail(msg)
def CheckCopyInProcessing(self):
"""
Test putline(), endcopy()
"""
try:
try:
ret = cnx.query("drop table copytest")
except Exception, msg:
pass
res = cnx.query("create table copytest (a int, b text, c float8)")
res = cnx.query("copy copytest from stdin")
self.assertEquals(res.resultStatus, libpq.COPY_IN,
"Copy In command did not return expect results.")
cnx.putline('3\tHello World\t4.5\n')
cnx.putline('4\tGoodbye World\t7.11\n')
cnx.putline('\\.\n')
cnx.endcopy()
res = cnx.query("select * from copytest")
self.assertEquals(res.ntuples, 2, "Copy In did not work")
self.assertEquals(res.getvalue(0,0), 3, "Copy In did not work")
self.assertEquals(res.getvalue(0,1), "Hello World",
"Copy In did not work")
self.assertEquals(res.getvalue(0,2), 4.5, "Copy In did not work")
self.assertEquals(res.getvalue(1,0), 4, "Copy In did not work")
self.assertEquals(res.getvalue(1,1), "Goodbye World",
"Copy In did not work")
self.assertEquals(res.getvalue(1,2), 7.11, "Copy In did not work")
del res
except Exception, msg:
self.fail(msg)
def CheckAsyncProcessing(self):
"""
Test sendQuery, consumeInput, and getResult.
"""
try:
cnx.sendQuery("""select * from pg_database
where datname='template1'""");
while 1:
ready_sockets = select.select([cnx.socket], [], [], 1.0)[0]
if len(ready_sockets) != 0:
cnx.consumeInput()
if cnx.isBusy == 0:
break
res = cnx.getResult()
self.assertEquals(type(res), libpq.PgResultType,
'getResult did not return a PgResult.')
self.assertEquals(res.ntuples, 1,
'Query return wrong number of tuples.')
self.assertEquals(res.nfields, 7,
'Query return wrong number of fields.')
if res.getvalue(0,0) != 'template1' or \
res.getvalue(0,2) != 0 or \
res.getvalue(0,3) != 1 or \
res.getvalue(0,4) != 1:
self.fail('Query returned unexpected data')
except Exception, msg:
self.fail(msg)
def CheckMethodFunctionality(self):
"""
Check setnonblocking(), reset()
"""
self.assertEquals(cnx.isnonblocking, 0,
"isnonblocking should be 0. It isn't.")
cnx.setnonblocking(1)
self.assertEquals(cnx.isnonblocking, 1,
"isnonblocking should be 1. It isn't.")
cnx.setnonblocking(0)
self.assertEquals(cnx.isnonblocking, 0,
"isnonblocking should be 0. It isn't.")
cnx.reset()
expected = self.memtypes[:]
self.CheckMemberTypes(cnx, expected)
# Note: There is no way we can determine what the value of cnx.socket
# and cnx.backendPID are programtically. Also errorMessage will
# contain a string at this time (from previous errors).
# Fill in the expected values from the connection.
cdef = self.cdef
expected = [cdef['host'], cdef['port'], 'pypgsql', cdef['tty'],
cdef['user'], cdef['password'], self.cnx.backendPID,
self.cnx.socket, [], self.vstr, 0, cnx.errorMessage, 0, 0]
self.CheckMemberValues(cnx, expected)
def CheckMethodSanity3(self):
"""
Check response when methods are called with erronious arguments.
"""
self.CheckForMethods(self.cnx)
self.failUnlessRaises(ValueError, self.cnx.lo_creat, 's')
self.failUnlessRaises(libpq.OperationalError,
self.cnx.lo_import, 'NoSuchFile')
self.failUnlessRaises(IOError, self.cnx.lo_unlink, 0)
# putline can not have an erronious argument if the type is correct
self.failUnlessRaises(TypeError, self.cnx.putline, 'a\x00b')
self.failUnlessRaises(TypeError, self.cnx.query, 0)
self.failUnlessRaises(TypeError, self.cnx.sendQuery, 0)
# trace can not have an erronious argument if the type is correct
# setnonblocking can not have an erronious argument if the type is
# correct.
def CheckMethodSanity2(self):
"""
Check response when methods are called with the wrong type of argument[s]
"""
self.CheckForMethods(self.cnx)
self.failUnlessRaises(TypeError, self.cnx.lo_creat, [])
self.failUnlessRaises(TypeError, self.cnx.lo_import, [])
self.failUnlessRaises(TypeError, self.cnx.lo_unlink, [])
self.failUnlessRaises(TypeError, self.cnx.putline, [])
self.failUnlessRaises(TypeError, self.cnx.query, [])
self.failUnlessRaises(TypeError, self.cnx.sendQuery, [])
self.failUnlessRaises(TypeError, self.cnx.trace, [])
self.failUnlessRaises(TypeError, self.cnx.setnonblocking, [])
def CheckMethodSanity1(self):
"""
Check response when methods are called with the wrong number of arguments
"""
self.CheckForMethods(self.cnx)
self.failUnlessRaises(libpq.InterfaceError, self.cnx.consumeInput, 1)
self.failUnlessRaises(libpq.InterfaceError, self.cnx.endcopy, 1)
self.failUnlessRaises(libpq.InterfaceError, self.cnx.finish, 1)
self.failUnlessRaises(libpq.InterfaceError, self.cnx.getResult, 1)
self.failUnlessRaises(libpq.InterfaceError, self.cnx.getline, 1)
self.failUnlessRaises(libpq.InterfaceError, self.cnx.getlineAsync, 1)
self.failUnlessRaises(TypeError, self.cnx.lo_creat)
self.failUnlessRaises(TypeError, self.cnx.lo_creat, 1, 2)
self.failUnlessRaises(TypeError, self.cnx.lo_import)
self.failUnlessRaises(TypeError, self.cnx.lo_import, 1, 2)
self.failUnlessRaises(TypeError, self.cnx.lo_unlink)
self.failUnlessRaises(TypeError, self.cnx.lo_unlink, 1, 2)
self.failUnlessRaises(libpq.InterfaceError, self.cnx.notifies, 1)
self.failUnlessRaises(TypeError, self.cnx.putline)
self.failUnlessRaises(TypeError, self.cnx.putline, 1, 2)
self.failUnlessRaises(TypeError, self.cnx.query)
self.failUnlessRaises(TypeError, self.cnx.query, 1, 2)
self.failUnlessRaises(libpq.InterfaceError, self.cnx.requestCancel, 1)
self.failUnlessRaises(libpq.InterfaceError, self.cnx.reset, 1)
self.failUnlessRaises(TypeError, self.cnx.sendQuery)
self.failUnlessRaises(TypeError, self.cnx.sendQuery, 1, 2)
self.failUnlessRaises(TypeError, self.cnx.trace)
self.failUnlessRaises(TypeError, self.cnx.trace, 1, 2)
self.failUnlessRaises(libpq.InterfaceError, self.cnx.untrace, 1)
self.failUnlessRaises(libpq.InterfaceError, self.cnx.connectPoll, 1)
self.failUnlessRaises(libpq.InterfaceError, self.cnx.flush, 1)
self.failUnlessRaises(TypeError, self.cnx.setnonblocking)
self.failUnlessRaises(TypeError, self.cnx.setnonblocking, 1, 2)
# Check for existance of removed method, lo_get()
try:
tattr = self.cnx.lo_get
self.fail("PgConnection method 'lo_get' exists!")
except AttributeError, msg:
pass
def CheckMemberSanity(self):
"""
Check the object for valid member types and values.
"""
self.CheckForMethods(self.cnx)
expected = self.memtypes[:]
expected[11] = types.NoneType
self.CheckMemberTypes(cnx, expected)
# Note: There is no way we can determine what the value of cnx.socket
# and cnx.backendPID are programtically. Fill in the expected
# values from the connection.
cdef = self.cdef
expected = [cdef['host'], cdef['port'], 'pypgsql', cdef['tty'],
cdef['user'], cdef['password'], self.cnx.backendPID,
self.cnx.socket, [], self.vstr, 0, None, 0, 0]
self.CheckMemberValues(cnx, expected)
# Test the hidden attribute, toggleShowQuery.
self.assertEquals(cnx.toggleShowQuery, 'On',
'Turning on ShowQuery failed.')
self.assertEquals(cnx.toggleShowQuery, 'Off',
'Turning off ShowQuery failed.')
if __name__ == "__main__":
TestSuite = unittest.TestSuite()
TestSuite.addTest(PgConnectionTestCases("CheckMemberSanity"))
TestSuite.addTest(PgConnectionTestCases("CheckMethodSanity1"))
TestSuite.addTest(PgConnectionTestCases("CheckMethodSanity2"))
TestSuite.addTest(PgConnectionTestCases("CheckMethodSanity3"))
TestSuite.addTest(PgConnectionTestCases("CheckMethodFunctionality"))
TestSuite.addTest(PgConnectionTestCases("CheckAsyncProcessing"))
TestSuite.addTest(PgConnectionTestCases("CheckCopyInProcessing"))
TestSuite.addTest(PgConnectionTestCases("CheckCopyOutProcessing"))
TestSuite.addTest(PgConnectionTestCases("CheckCopyOutAsyncProcessing"))
TestSuite.addTest(PgConnectionTestCases("CheckFinish"))
runner = unittest.TextTestRunner()
runner.run(TestSuite)
|
