From: brian z. <bz...@us...> - 2001-12-29 07:16:58
|
Update of /cvsroot/jython/jython/Lib/test/zxjdbc In directory usw-pr-cvs1:/tmp/cvs-serv2592 Modified Files: dbextstest.py sptest.py test.xml zxtest.py Log Message: stored procedure changes Index: dbextstest.py =================================================================== RCS file: /cvsroot/jython/jython/Lib/test/zxjdbc/dbextstest.py,v retrieving revision 1.1 retrieving revision 1.2 diff -C2 -d -r1.1 -r1.2 *** dbextstest.py 2001/12/14 04:20:03 1.1 --- dbextstest.py 2001/12/29 07:16:55 1.2 *************** *** 101,104 **** --- 101,111 ---- assert len(r) == 2, "expected [2], got [%d]" % (len(r)) + def testUpdateCount(self): + """testing update count""" + + self._insertInto("one", 45) + self.db.raw("delete from one where a > ?", [(12,)]) + self.assertEquals(32, self.db.updatecount) + def testQueryWithMaxRows(self): """testing query with max rows""" *************** *** 225,226 **** --- 232,248 ---- done = bcp.transfer(self.db) assert done == 32, "expecting thirty two rows to be inserted, found [%d]" % (done) + + def testAutocommit(self): + """testing the autocommit functionality""" + for u in (0, 1): + self.db.autocommit = u + try: + self.db.isql("select * from one") + except Exception, e: + fail("failed autocommit query with u=[%d], v=[%d]" % (u, v)) + for v in (0, 1): + self.db.db.autocommit = v + try: + self.db.isql("select * from one") + except Exception, e: + fail("failed autocommit query with u=[%d], v=[%d]" % (u, v)) Index: sptest.py =================================================================== RCS file: /cvsroot/jython/jython/Lib/test/zxjdbc/sptest.py,v retrieving revision 1.2 retrieving revision 1.3 diff -C2 -d -r1.2 -r1.3 *** sptest.py 2001/12/21 03:45:27 1.2 --- sptest.py 2001/12/29 07:16:55 1.3 *************** *** 6,21 **** # 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() --- 6,15 ---- # Copyright (c) 2001 brian zimmer <bz...@zi...> ! from zxtest import zxCoreTestCase ! class OracleSPTest(zxCoreTestCase): def setUp(self): ! zxCoreTestCase.setUp(self) c = self.cursor() *************** *** 23,47 **** 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 in 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: --- 17,39 ---- try: try: ! c.execute("drop table sptest") except: self.db.rollback() try: ! c.execute("create table sptest (x varchar2(20))") ! c.execute("create or replace procedure procnone is begin insert into sptest values ('testing'); end;") ! c.execute("create or replace procedure procin (y in varchar2) is begin insert into sptest values (y); end;") ! c.execute("create or replace procedure procout (y out varchar2) is begin y := 'tested'; end;") ! c.execute("create or replace procedure procinout (y out varchar2, z in varchar2) is begin insert into sptest values (z); y := 'tested'; end;") ! c.execute("create or replace function funcnone return varchar2 is begin return 'tested'; end;") ! c.execute("create or replace function funcin (y varchar2) return varchar2 is begin return y || y; end;") ! c.execute("create or replace function funcout (y out varchar2) return varchar2 is begin y := 'tested'; return 'returned'; end;") self.db.commit() except: self.db.rollback() ! self.fail("procedure creation failed") self.proc_errors("PROC") self.proc_errors("FUNC") finally: *************** *** 49,58 **** 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: --- 41,50 ---- def tearDown(self): ! zxCoreTestCase.tearDown(self) def proc_errors(self, name): c = self.cursor() try: ! c.execute("select * from user_errors where name like '%s%%'" % (name.upper())) errors = c.fetchall() try: *************** *** 65,74 **** c.close() def testProcin(self): c = self.cursor() try: ! c.callproc("procin", ("testProcin",)) self.assertEquals(None, c.fetchall()) ! c.execute("select * from plsqltest") self.assertEquals(1, len(c.fetchall())) finally: --- 57,108 ---- c.close() + def testCursor(self): + c = self.cursor() + try: + + c.execute("insert into sptest values ('a')") + c.execute("insert into sptest values ('b')") + c.execute("insert into sptest values ('c')") + c.execute("insert into sptest values ('d')") + c.execute("insert into sptest values ('e')") + + c.execute(""" + CREATE OR REPLACE PACKAGE types + AS + TYPE ref_cursor IS REF CURSOR; + END; + """) + + c.execute(""" + CREATE OR REPLACE FUNCTION funccur(v_x IN VARCHAR) + RETURN types.ref_cursor + AS + funccur_cursor types.ref_cursor; + BEGIN + OPEN funccur_cursor FOR + SELECT x FROM sptest WHERE x < v_x; + RETURN funccur_cursor; + END; + """) + + self.proc_errors("funccur") + + c.callproc("funccur", ("z",)) + data = c.fetchall() + self.assertEquals(5, len(data)) + c.callproc("funccur", ("c",)) + data = c.fetchall() + self.assertEquals(2, len(data)) + + finally: + c.close() + def testProcin(self): c = self.cursor() try: ! params = ["testProcin"] ! c.callproc("procin", params) self.assertEquals(None, c.fetchall()) ! c.execute("select * from sptest") self.assertEquals(1, len(c.fetchall())) finally: *************** *** 78,85 **** c = self.cursor() try: ! c.callproc("procinout", ("testing",)) data = c.fetchone() ! assert data is not None, "data was None" ! self.assertEquals("tested", data[0]) finally: c.close() --- 112,123 ---- c = self.cursor() try: ! params = [None, "testing"] ! c.callproc("procinout", params) data = c.fetchone() ! assert data is None, "data was not None" ! c.execute("select * from sptest") ! data = c.fetchone() ! self.assertEquals("testing", data[0]) ! self.assertEquals("tested", params[0]) finally: c.close() *************** *** 99,103 **** c = self.cursor() try: ! c.callproc("funcin", ("testing",)) self.assertEquals(1, c.rowcount) data = c.fetchone() --- 137,155 ---- c = self.cursor() try: ! params = ["testing"] ! c.callproc("funcin", params) ! self.assertEquals(1, c.rowcount) ! data = c.fetchone() ! assert data is not None, "data was None" ! self.assertEquals(1, len(data)) ! self.assertEquals("testingtesting", data[0]) ! finally: ! c.close() ! ! def testCallingWithKws(self): ! c = self.cursor() ! try: ! params = ["testing"] ! c.callproc("funcin", params=params) self.assertEquals(1, c.rowcount) data = c.fetchone() *************** *** 111,133 **** c = self.cursor() try: ! c.callproc("funcout") data = c.fetchone() assert data is not None, "data was None" ! self.assertEquals(2, len(data)) self.assertEquals("returned", data[0]) ! self.assertEquals("tested", data[1].strip()) finally: c.close() ! def testRaisesalary(self): c = self.cursor() try: ! c.callproc("raisesal", ("jython developer", 18000)) data = c.fetchone() assert data is not None, "data was None" self.assertEquals(1, len(data)) ! self.assertEquals(18000 + 100000, data[0]) finally: c.close() --- 163,238 ---- c = self.cursor() try: ! params = [None] ! c.callproc("funcout", params) data = c.fetchone() assert data is not None, "data was None" ! self.assertEquals(1, len(data)) self.assertEquals("returned", data[0]) ! self.assertEquals("tested", params[0].strip()) finally: c.close() ! def testMultipleFetch(self): ! """testing the second fetch call to a callproc() is None""" c = self.cursor() try: ! c.callproc("funcnone") data = c.fetchone() assert data is not None, "data was None" + data = c.fetchone() + assert data is None, "data was not None" + finally: + c.close() + + class SQLServerSPTest(zxCoreTestCase): + + def testProcWithResultSet(self): + c = self.cursor() + try: + c.execute("use ziclix") + + self.assertEquals("ziclix", c.connection.__connection__.getCatalog()) + + try: + c.execute("drop table sptest") + except: + pass + + c.execute("create table sptest (a int, b varchar(32))") + c.execute("insert into sptest values (1, 'hello')") + c.execute("insert into sptest values (2, 'there')") + c.execute("insert into sptest values (3, 'goodbye')") + + try: + c.execute("drop procedure sp_proctest") + except: + pass + + c.execute(""" + create procedure sp_proctest (@A int) + as + select a, b from sptest where a <= @A + """) + + c.callproc(("ziclix", "jython", "sp_proctest"), (2,)) + data = c.fetchall() + self.assertEquals(2, len(data)) + self.assertEquals(2, len(c.description)) + assert c.nextset() is not None, "expected an additional result set" + data = c.fetchall() self.assertEquals(1, len(data)) ! self.assertEquals(1, len(c.description)) finally: c.close() + def testSalesByCategory(self): + c = self.cursor() + try: + c.execute("use northwind") + c.callproc(("northwind", "dbo", "SalesByCategory"), ["Seafood", "1998"]) + data = c.fetchall() + assert data is not None, "no results from SalesByCategory" + assert len(data) > 0, "expected numerous results" + finally: + c.close() Index: test.xml =================================================================== RCS file: /cvsroot/jython/jython/Lib/test/zxjdbc/test.xml,v retrieving revision 1.4 retrieving revision 1.5 diff -C2 -d -r1.4 -r1.5 *** test.xml 2001/12/21 21:06:34 1.4 --- test.xml 2001/12/29 07:16:55 1.5 *************** *** 19,27 **** <argument name="driver" value="org.postgresql.Driver"/> </factory> ! <testcase from="zxtest" import="zxCoreTest"> <!-- Returns any empty ResultSet every time. --> <ignore name="testIndexInfo"/> </testcase> ! <testcase from="zxtest" import="BCPTest"/> <testcase from="dbextstest" import="dbextsTestCase"/> </test> --- 19,27 ---- <argument name="driver" value="org.postgresql.Driver"/> </factory> ! <testcase from="zxtest" import="zxAPITestCase"> <!-- Returns any empty ResultSet every time. --> <ignore name="testIndexInfo"/> </testcase> ! <testcase from="zxtest" import="BCPTestCase"/> <testcase from="dbextstest" import="dbextsTestCase"/> </test> *************** *** 36,43 **** <keyword name="portNumber" value="5432" type="int"/> </factory> ! <testcase from="zxtest" import="zxCoreTest"> <ignore name="testIndexInfo"/> </testcase> ! <testcase from="zxtest" import="BCPTest"/> </test> --> --- 36,43 ---- <keyword name="portNumber" value="5432" type="int"/> </factory> ! <testcase from="zxtest" import="zxAPITestCase"> <ignore name="testIndexInfo"/> </testcase> ! <testcase from="zxtest" import="BCPTestCase"/> </test> --> *************** *** 79,84 **** <argument name="driver" value="org.gjt.mm.mysql.Driver"/> </factory> ! <testcase from="zxtest" import="zxCoreTest"/> ! <testcase from="zxtest" import="BCPTest"/> <testcase from="dbextstest" import="dbextsTestCase"/> </test> --- 79,84 ---- <argument name="driver" value="org.gjt.mm.mysql.Driver"/> </factory> ! <testcase from="zxtest" import="zxAPITestCase"/> ! <testcase from="zxtest" import="BCPTestCase"/> <testcase from="dbextstest" import="dbextsTestCase"/> </test> *************** *** 92,97 **** <keyword name="port" value="3306" type="int"/> </factory> ! <testcase from="zxtest" import="zxCoreTest"/> ! <testcase from="zxtest" import="BCPTest"/> </test> <!-- --- 92,97 ---- <keyword name="port" value="3306" type="int"/> </factory> ! <testcase from="zxtest" import="zxAPITestCase"/> ! <testcase from="zxtest" import="BCPTestCase"/> </test> <!-- *************** *** 101,106 **** <keyword name="INITIAL_CONTEXT_FACTORY" value="com.sun.jndi.fscontext.RefFSContextFactory"/> </factory> ! <testcase from="zxtest" import="zxCoreTest"/> ! <testcase from="zxtest" import="BCPTest"/> </test> --> --- 101,106 ---- <keyword name="INITIAL_CONTEXT_FACTORY" value="com.sun.jndi.fscontext.RefFSContextFactory"/> </factory> ! <testcase from="zxtest" import="zxAPITestCase"/> ! <testcase from="zxtest" import="BCPTestCase"/> </test> --> *************** *** 139,148 **** <argument name="driver" value="oracle.jdbc.driver.OracleDriver"/> </factory> ! <testcase from="zxtest" import="zxCoreTest"> <ignore name="testRowid"/> </testcase> ! <testcase from="zxtest" import="BCPTest"/> <testcase from="dbextstest" import="dbextsTestCase"/> ! <testcase from="sptest" import="SPTest"/> </test> <!-- --- 139,148 ---- <argument name="driver" value="oracle.jdbc.driver.OracleDriver"/> </factory> ! <testcase from="zxtest" import="zxAPITestCase"> <ignore name="testRowid"/> </testcase> ! <testcase from="zxtest" import="BCPTestCase"/> <testcase from="dbextstest" import="dbextsTestCase"/> ! <testcase from="sptest" import="OracleSPTest"/> </test> <!-- *************** *** 154,161 **** <keyword name="password" value="jython"/> </factory> ! <testcase from="zxtest" import="zxCoreTest"> <ignore name="testRowid"/> </testcase> ! <testcase from="zxtest" import="BCPTest"/> </test> --> --- 154,162 ---- <keyword name="password" value="jython"/> </factory> ! <testcase from="zxtest" import="zxAPITestCase"> <ignore name="testRowid"/> </testcase> ! <testcase from="zxtest" import="BCPTestCase"/> ! <testcase from="sptest" import="OracleSPTest"/> </test> --> *************** *** 185,188 **** --- 186,248 ---- </table> </vendor> + <vendor name="microsoft" + datahandler="com.ziclix.python.sql.handler.SQLServerDataHandler"> + <test name="opta driver" os="java"> + <factory class="com.ziclix.python.sql.zxJDBC" method="connect"> + <argument name="url" value="jdbc:inetdae7:localhost:1433?database=ziclix"/> + <argument name="usr" value="jython"/> + <argument name="pwd" value="jython"/> + <argument name="driver" value="com.inet.tds.TdsDriver"/> + </factory> + <!-- + <testcase from="zxtest" import="zxAPITestCase"> + <ignore name="testRowid"/> + </testcase> + <testcase from="dbextstest" import="dbextsTestCase"> + <ignore name="testBulkcopy"/> + <ignore name="testBulkcopyWithDynamicColumns"/> + </testcase> + --> + <testcase from="sptest" import="SQLServerSPTest"/> + </test> + <test name="microsoft driver" os="java"> + <factory class="com.ziclix.python.sql.zxJDBC" method="connect"> + <argument name="url" value="jdbc:microsoft:sqlserver://localhost:1433"/> + <argument name="usr" value="jython"/> + <argument name="pwd" value="jython"/> + <argument name="driver" value="com.microsoft.jdbc.sqlserver.SQLServerDriver"/> + </factory> + <!-- + <testcase from="zxtest" import="zxAPITestCase"> + <ignore name="testRowid"/> + </testcase> + <testcase from="zxtest" import="BCPTestCase"/> + <testcase from="dbextstest" import="dbextsTestCase"/> + <testcase from="sptest" import="SQLServerSPTest"/> + --> + </test> + <table ref="texttable" name="c_texttable"> + create table c_texttable (a int not null, primary key(a), b text not null) + </table> + <table ref="floattable" name="c_floattable"> + create table c_floattable (a int, b numeric(5,2)) + </table> + <table ref="datetable" name="c_datetable"> + create table c_datetable (a int, b datetime) + </table> + <table ref="timetable" name="c_timetable"> + create table c_timetable (a int, b datetime) + </table> + <table ref="timestamptable" name="c_timestamptable"> + create table c_timestamptable (a int, b datetime) + </table> + <table ref="blobtable" name="b_blobtable"> + create table b_blobtable (a int, b blob) + </table> + <table ref="pktable" name="b_pktable"> + create table b_pktable (a int not null, b int, primary key(a)) + </table> + </vendor> + <!-- <vendor name="cloudscape" datahandler="com.ziclix.python.sql.JDBC20DataHandler"> *************** *** 194,201 **** <argument name="driver" value="COM.cloudscape.core.JDBCDriver"/> </factory> ! <testcase from="zxtest" import="zxCoreTest"> <ignore name="testRowid"/> </testcase> ! <testcase from="zxtest" import="BCPTest"/> <testcase from="dbextstest" import="dbextsTestCase"/> </test> --- 254,261 ---- <argument name="driver" value="COM.cloudscape.core.JDBCDriver"/> </factory> ! <testcase from="zxtest" import="zxAPITestCase"> <ignore name="testRowid"/> </testcase> ! <testcase from="zxtest" import="BCPTestCase"/> <testcase from="dbextstest" import="dbextsTestCase"/> </test> *************** *** 221,227 **** create table b_pktable (a int not null, b int, primary key(a)) </table> - <table ref="autoincrementtable" name="aitable"> - create table aitable (a int with default autoincrement not null, b int, primary key(a)) - </table> </vendor> </tests> --- 281,285 ---- create table b_pktable (a int not null, b int, primary key(a)) </table> </vendor> + --> </tests> Index: zxtest.py =================================================================== RCS file: /cvsroot/jython/jython/Lib/test/zxjdbc/zxtest.py,v retrieving revision 1.3 retrieving revision 1.4 diff -C2 -d -r1.3 -r1.4 *** zxtest.py 2001/12/21 03:45:27 1.3 --- zxtest.py 2001/12/29 07:16:55 1.4 *************** *** 11,15 **** import tempfile, os, time, runner ! class zxJDBCTest(runner.SQLTestCase): def connect(self): --- 11,24 ---- import tempfile, os, time, runner ! class zxCoreTestCase(runner.SQLTestCase): ! ! def setUp(self): ! runner.SQLTestCase.setUp(self) ! self.db = self.connect() ! self.db.autocommit = 0 ! ! def tearDown(self): ! self.db.close() ! runner.SQLTestCase.tearDown(self) def connect(self): *************** *** 25,30 **** return c def setUp(self): ! runner.SQLTestCase.setUp(self) self.db = self.connect() self.db.autocommit = 0 --- 34,41 ---- return c + class zxJDBCTestCase(zxCoreTestCase): + def setUp(self): ! zxCoreTestCase.setUp(self) self.db = self.connect() self.db.autocommit = 0 *************** *** 61,70 **** c.close() ! try: ! self.db.close() ! finally: ! self.db = None ! class zxCoreTest(zxJDBCTest): def testConnection(self): --- 72,78 ---- c.close() ! zxCoreTestCase.tearDown(self) ! class zxAPITestCase(zxJDBCTestCase): def testConnection(self): *************** *** 72,75 **** --- 80,91 ---- assert self.db, "invalid connection" + def testAutocommit(self): + """testing autocommit functionality""" + if self.db.__connection__.getMetaData().supportsTransactions(): + self.db.autocommit = 1 + self.assertEquals(1, self.db.__connection__.getAutoCommit()) + self.db.autocommit = 0 + self.assertEquals(0, self.db.__connection__.getAutoCommit()) + def testSimpleQuery(self): """testing simple queries with cursor.execute(), no parameters""" *************** *** 172,191 **** assert self.has_table("texttable"), "missing attribute texttable" fp = open(tempfile.mktemp(), "w") try: ! c = self.cursor() ! c.execute(self.table("texttable")[1]) ! data = fp.name * 300 ! data = data[:3500] ! fp.write(data) ! fp.flush() ! fp.close() ! fp = open(fp.name, "r") ! c.execute("insert into %s (a, b) values (?, ?)" % (self.table("texttable")[0]), [(0, fp)], {1:zxJDBC.LONGVARCHAR}) ! self.db.commit() ! c.execute("select b from %s" % (self.table("texttable")[0])) ! f = c.fetchall() ! assert len(f) == 1, "expected [1] row, got [%d]" % (len(f)) ! assert len(f[0][0]) == len(data), "expected [%d], got [%d]" % (len(data), len(f[0][0])) ! assert data == f[0][0], "failed to retrieve the same text as inserted" finally: c.execute("drop table %s" % (self.table("texttable")[0])) --- 188,211 ---- assert self.has_table("texttable"), "missing attribute texttable" fp = open(tempfile.mktemp(), "w") + c = self.cursor() try: ! try: ! c.execute(self.table("texttable")[1]) ! data = fp.name * 300 ! data = data[:3500] ! fp.write(data) ! fp.flush() ! fp.close() ! fp = open(fp.name, "r") ! c.execute("insert into %s (a, b) values (?, ?)" % (self.table("texttable")[0]), [(0, fp)], {1:zxJDBC.LONGVARCHAR}) ! self.db.commit() ! c.execute("select b from %s" % (self.table("texttable")[0])) ! f = c.fetchall() ! assert len(f) == 1, "expected [1] row, got [%d]" % (len(f)) ! assert len(f[0][0]) == len(data), "expected [%d], got [%d]" % (len(data), len(f[0][0])) ! assert data == f[0][0], "failed to retrieve the same text as inserted" ! except Exception, e: ! print e ! raise e finally: c.execute("drop table %s" % (self.table("texttable")[0])) *************** *** 577,589 **** c.close() ! class LOBTest(zxJDBCTest): def __blob(self, obj=0): assert self.has_table("blobtable"), "no blob table" - c = self.cursor() tabname, sql = self.table("blobtable") fn = tempfile.mktemp() fp = None try: --- 597,609 ---- c.close() ! class LOBTest(zxJDBCTestCase): def __blob(self, obj=0): assert self.has_table("blobtable"), "no blob table" tabname, sql = self.table("blobtable") fn = tempfile.mktemp() fp = None + c = self.cursor() try: *************** *** 640,646 **** def __clob(self, asfile=0): assert self.has_table("clobtable"), "no clob table" - c = self.cursor() tabname, sql = self.table("clobtable") try: hello = "hello" * 1024 * 10 --- 660,666 ---- def __clob(self, asfile=0): assert self.has_table("clobtable"), "no clob table" tabname, sql = self.table("clobtable") + c = self.cursor() try: hello = "hello" * 1024 * 10 *************** *** 680,684 **** self.__clob(1) ! class BCPTest(zxJDBCTest): def testCSVPipe(self): --- 700,704 ---- self.__clob(1) ! class BCPTestCase(zxJDBCTestCase): def testCSVPipe(self): |