[Jython-checkins] CVS: jython/Lib/test/zxjdbc sptest.py,NONE,1.1 test.xml,1.1,1.2 zxtest.py,1.1,1.2

[brian z. <bz...@us...>]

Update of /cvsroot/jython/jython/Lib/test/zxjdbc
In directory usw-pr-cvs1:/tmp/cvs-serv25475

Modified Files:
	test.xml zxtest.py 
Added Files:
	sptest.py 
Log Message:
initial tests for procedures

--- NEW FILE: sptest.py ---

# Jython Database Specification API 2.0
#
# $Id: sptest.py,v 1.1 2001/12/17 03:45:38 bzimmer Exp $
#
# Copyright (c) 2001 brian zimmer <bz...@zi...>

import sys
from zxtest import zxJDBCTest
from com.ziclix.python.sql import Procedure

class SPTest(zxJDBCTest):
	"""
	These tests are very specific to Oracle.  Eventually support for other engines will
	be available.
	"""

	def setUp(self):
		zxJDBCTest.setUp(self)

		c = self.cursor()

		try:
			try:
				c.execute("drop table plsqltest")
			except:
				self.db.rollback()
			try:
				c.execute("create table plsqltest (x char(20))")
				c.execute("create or replace procedure procnone is begin insert into plsqltest values ('testing'); end;")
				c.execute("create or replace procedure procin (y char) is begin insert into plsqltest values (y); end;")
				c.execute("create or replace procedure procout (y out char) is begin y := 'tested'; end;")
				c.execute("create or replace procedure procinout (y out varchar, z in varchar) is begin insert into plsqltest values (z); y := 'tested'; end;")
				c.execute("create or replace function funcnone return char is begin return 'tested'; end;")
				c.execute("create or replace function funcin (y char) return char is begin return y || y; end;")
				c.execute("create or replace function funcout (y out char) return char is begin y := 'tested'; return 'returned'; end;")
				c.execute("create or replace function raisesal (name char, raise number) return number is begin return raise + 100000; end;")
				self.db.commit()
			except:
				self.db.rollback()
				fail("procedure creation failed")

			self.proc_errors("PROC")
			self.proc_errors("FUNC")
			self.proc_errors("RAISESAL")

		finally:
			c.close()

	def tearDown(self):
		zxJDBCTest.tearDown(self)

	def proc_errors(self, name):
		c = self.cursor()
		try:
			c.execute("select * from user_errors where name like '%s%%'" % (name))
			errors = c.fetchall()
			try:
				assert errors is None, "found errors"
			except AssertionError, e:
				for a in errors:
					print a
				raise e
		finally:
			c.close()

	def testProcinout(self):
		c = self.cursor()
		try:
			p = Procedure(c, "procinout")
			stmt = p.prepareCall()
			params = ["testing"]
			params = p.normalizeParams(params)

			self.assertEquals(2, len(params))
			assert params[0] == p.PLACEHOLDER
			assert params[1] == "testing"
		finally:
			if stmt:
				stmt.close()
			c.close()

	def testFuncout(self):
		c = self.cursor()
		try:
			p = Procedure(c, "funcout")
			stmt = p.prepareCall()
			stmt.execute()
			self.assertEquals("returned", stmt.getString(1).strip())
			self.assertEquals("tested", stmt.getString(2).strip())
		finally:
			if stmt:
				stmt.close()
			c.close()

	def testProcinoutCall(self):
		c = self.cursor()
		try:
			c.callproc("procinout", ("testing",))
			assert c.fetchall() == None, "expected None"
		finally:
			c.close()

	def testFuncnoneCall(self):
		c = self.cursor()
		try:
			c.callproc("funcnone")
			assert c.fetchall() == None, "expected None"
		finally:
			c.close()

	def testFuncinCall(self):
		c = self.cursor()
		try:
			c.callproc("funcin", ("testing",))
			assert c.fetchall() == None, "expected None"
		finally:
			c.close()

	def testFuncoutCall(self):
		c = self.cursor()
		try:
			c.callproc("funcout")
			assert c.fetchall() == None, "expected None"
		finally:
			c.close()

	def testRaisesalaryCall(self):
		c = self.cursor()
		try:
			c.callproc("raisesal", ("jython developer", 18000))
			assert c.fetchall() == None, "expected None"
		finally:
			c.close()



Index: test.xml
===================================================================
RCS file: /cvsroot/jython/jython/Lib/test/zxjdbc/test.xml,v
retrieving revision 1.1
retrieving revision 1.2
diff -C2 -d -r1.1 -r1.2
*** test.xml	2001/12/14 04:20:03	1.1
--- test.xml	2001/12/17 03:45:38	1.2
***************
*** 146,150 ****
--- 146,152 ----
  			<testcase from="zxtest" import="BCPTest"/>
  			<testcase from="dbextstest" import="dbextsTestCase"/>
+ 			<testcase from="sptest" import="SPTest"/>
  		</test>
+ 		<!--
  		<test name="datasource" os="java">
  			<factory class="com.ziclix.python.sql.zxJDBC" method="connectx">
***************
*** 159,162 ****
--- 161,165 ----
  			<testcase from="zxtest" import="BCPTest"/>
  		</test>
+ 		-->
  		<table ref="texttable" name="c_texttable">
  			create table c_texttable (a int, b varchar2(4000) not null)

Index: zxtest.py
===================================================================
RCS file: /cvsroot/jython/jython/Lib/test/zxjdbc/zxtest.py,v
retrieving revision 1.1
retrieving revision 1.2
diff -C2 -d -r1.1 -r1.2
*** zxtest.py	2001/12/14 04:20:03	1.1
--- zxtest.py	2001/12/17 03:45:38	1.2
***************
*** 339,367 ****
  			c.close()
  
! 	def __updatecount(self, insert_only=0):
! 		from com.ziclix.python.sql.handler import UpdateCountDataHandler
  		c = self.cursor()
  		try:
- 			c.datahandler = UpdateCountDataHandler(c.datahandler)
- 			msg = "wrong instance, expected [UpdateCountDataHandler], got [%s]" % (str(c.datahandler))
- 			assert isinstance(c.datahandler, UpdateCountDataHandler), msg
  			c.execute("insert into zxtesting values (?, ?, ?)", [(500, 'bz', 'or')])
! 			assert c.datahandler.updateCount == 1, "expected [1], got [%d]" % (c.datahandler.updateCount)
  			# there's a *feature* in the mysql engine where it returns 0 for delete if there is no
  			#  where clause, regardless of the actual value.  using a where clause forces it to calculate
  			#  the appropriate value
  			c.execute("delete from zxtesting where 1>0")
! 			if not insert_only:
! 				assert c.datahandler.updateCount == 8, "expected [8], got [%d]" % (c.datahandler.updateCount)
  		finally:
  			c.close()
- 
- 	def testUpdateCountDataHandler(self):
- 		"""testing custom data handler for getting the update count"""
- 		self.__updatecount()
- 
- 	def testUpdateCountDataHandlerInsertOnly(self):
- 		"""testing custom data handler for getting the update count on inserts only"""
- 		self.__updatecount(1)
  
  	def _test_time(self, (tabname, sql), factory, values, _type, _cmp=cmp):
--- 339,354 ----
  			c.close()
  
! 	def testUpdateCount(self, insert_only=0):
  		c = self.cursor()
  		try:
  			c.execute("insert into zxtesting values (?, ?, ?)", [(500, 'bz', 'or')])
! 			assert c.updatecount == 1, "expected [1], got [%d]" % (c.updatecount)
  			# there's a *feature* in the mysql engine where it returns 0 for delete if there is no
  			#  where clause, regardless of the actual value.  using a where clause forces it to calculate
  			#  the appropriate value
  			c.execute("delete from zxtesting where 1>0")
! 			assert c.updatecount == 8, "expected [8], got [%d]" % (c.updatecount)
  		finally:
  			c.close()
  
  	def _test_time(self, (tabname, sql), factory, values, _type, _cmp=cmp):