From: brian z. <bz...@us...> - 2002-01-10 03:42:35
|
Update of /cvsroot/jython/jython/Lib/test/zxjdbc In directory usw-pr-cvs1:/tmp/cvs-serv3844/Lib/test/zxjdbc Modified Files: dbextstest.py zxtest.py Log Message: more robust testing Index: dbextstest.py =================================================================== RCS file: /cvsroot/jython/jython/Lib/test/zxjdbc/dbextstest.py,v retrieving revision 1.2 retrieving revision 1.3 diff -C2 -d -r1.2 -r1.3 *** dbextstest.py 2001/12/29 07:16:55 1.2 --- dbextstest.py 2002/01/10 03:42:32 1.3 *************** *** 39,42 **** --- 39,45 ---- self.db = dbexts.dbexts(cfg=fp.name) self.db.verbose = 0 + for table in ("one", "two"): + try: self.db.raw("drop table %s" % (table)) + except: pass self.db.raw("create table one (a int, b int, c varchar(32))") self.db.raw("create table two (a int, b int, c varchar(32))") *************** *** 80,84 **** finally: self.db.commit(c) - c.close() if not failed: --- 83,86 ---- *************** *** 176,181 **** """testing dbexts.table(tabname)""" self.db.table("one") ! assert not self.db.results == None, "results were None" self.assertEquals(3, len(self.db.results)) def testOut(self): --- 178,188 ---- """testing dbexts.table(tabname)""" self.db.table("one") ! assert self.db.results is not None, "results were None" self.assertEquals(3, len(self.db.results)) + self.db.table() + found = 0 + for a in self.db.results: + if a[2].lower() in ("one", "two"): found += 1 + self.assertEquals(2, found) def testOut(self): Index: zxtest.py =================================================================== RCS file: /cvsroot/jython/jython/Lib/test/zxjdbc/zxtest.py,v retrieving revision 1.8 retrieving revision 1.9 diff -C2 -d -r1.8 -r1.9 *** zxtest.py 2002/01/08 05:39:32 1.8 --- zxtest.py 2002/01/10 03:42:32 1.9 *************** *** 104,117 **** c = self.cursor() try: ! c.execute("select * from zxtesting") cnt = 0 for a in c: - assert a is not None, "row is None" self.assertEquals(3, len(a)) cnt += 1 self.assertEquals(7, cnt) c.execute("select * from zxtesting") finally: c.close() def testColumns(self): --- 104,163 ---- c = self.cursor() try: ! # first with a for loop cnt = 0 + c.execute("select * from zxtesting") for a in c: self.assertEquals(3, len(a)) cnt += 1 self.assertEquals(7, cnt) + # then with a while loop + cnt = 0 + c.execute("select * from zxtesting") + while 1: + try: + self.assertEquals(3, len(c.next())) + except StopIteration: + break + cnt += 1 + self.assertEquals(7, cnt) + finally: + c.close() + + def testClosingCursor(self): + """testing that a closed cursor throws an exception""" + c = self.cursor() + try: c.execute("select * from zxtesting") finally: c.close() + self.assertRaises(zxJDBC.InternalError, c.execute, ("select * from zxtesting",)) + + def testNativeSQL(self): + """testing the connection's ability to convert sql""" + sql = self.db.nativesql("select * from zxtesting where id = ?") + assert sql is not None + assert len(sql) > 0 + + def testTables(self): + """testing cursor.tables()""" + c = self.cursor() + try: + c.tables(None, None, None, None) + # let's look for zxtesting + found = 0 + while not found: + try: + found = "zxtesting" == c.next()[2].lower() + except StopIteration: + break + assert found, "expected to find 'zxtesting'" + c.tables(None, None, "zxtesting", None) + self.assertEquals(1, len(c.fetchall())) + c.tables(None, None, "zxtesting", ("TABLE",)) + self.assertEquals(1, len(c.fetchall())) + c.tables(None, None, "zxtesting", ("table",)) + self.assertEquals(1, len(c.fetchall())) + finally: + c.close() def testColumns(self): *************** *** 119,127 **** c = self.cursor() try: c.columns(None, None, "zxtesting", None) f = c.fetchall() ! assert c.rowcount == 3, "columns() failed to report correct number of columns, expected [3], got [%d]" % (c.rowcount) f.sort(lambda x, y: cmp(x[3], y[3])) ! assert "name" == f[1][3].lower(), "expected [name], got [%s]" % (f[1][3].lower()) finally: c.close() --- 165,196 ---- c = self.cursor() try: + # deliberately copied so as to produce useful line numbers + c.columns(None, None, "zxtesting", None) f = c.fetchall() ! self.assertEquals(3, c.rowcount) f.sort(lambda x, y: cmp(x[3], y[3])) ! self.assertEquals("name", f[1][3].lower()) ! ! # if the db engine handles mixed case, then don't ask about a different ! # case because it will fail ! if not self.db.__connection__.getMetaData().storesMixedCaseIdentifiers(): ! c.columns(None, None, "ZXTESTING", None) ! f = c.fetchall() ! self.assertEquals(3, c.rowcount) ! f.sort(lambda x, y: cmp(x[3], y[3])) ! self.assertEquals("name", f[1][3].lower()) ! finally: ! c.close() ! ! def testBestRow(self): ! """testing bestrow which finds the optimal set of columns that uniquely identify a row""" ! c = self.cursor() ! try: ! # we're really just testing that this doesn't blow up ! c.bestrow(None, None, "zxtesting") ! f = c.fetchall() ! if f is not None: # we might as well see that it worked ! self.assertEquals(1, len(f)) finally: c.close() *************** *** 228,232 **** os.remove(fp.name) ! def __calendar(self): c = Calendar.getInstance() c.setTime(JDate()) --- 297,301 ---- os.remove(fp.name) ! def calendar(self): c = Calendar.getInstance() c.setTime(JDate()) *************** *** 238,242 **** # Java uses milliseconds and Python uses seconds, so adjust the time accordingly # seeded with Java ! c = self.__calendar() o = zxJDBC.DateFromTicks(c.getTime().getTime() / 1000L) v = zxJDBC.Date(c.get(Calendar.YEAR), c.get(Calendar.MONTH) + 1, c.get(Calendar.DATE)) --- 307,311 ---- # Java uses milliseconds and Python uses seconds, so adjust the time accordingly # seeded with Java ! c = self.calendar() o = zxJDBC.DateFromTicks(c.getTime().getTime() / 1000L) v = zxJDBC.Date(c.get(Calendar.YEAR), c.get(Calendar.MONTH) + 1, c.get(Calendar.DATE)) *************** *** 256,260 **** # seeded with Java ! c = self.__calendar() o = zxJDBC.TimeFromTicks(c.getTime().getTime() / 1000L) v = zxJDBC.Time(c.get(Calendar.HOUR), c.get(Calendar.MINUTE), c.get(Calendar.SECOND)) --- 325,329 ---- # seeded with Java ! c = self.calendar() o = zxJDBC.TimeFromTicks(c.getTime().getTime() / 1000L) v = zxJDBC.Time(c.get(Calendar.HOUR), c.get(Calendar.MINUTE), c.get(Calendar.SECOND)) *************** *** 274,278 **** # seeded with Java ! c = self.__calendar() o = zxJDBC.TimestampFromTicks(c.getTime().getTime() / 1000L) v = zxJDBC.Timestamp(c.get(Calendar.YEAR), c.get(Calendar.MONTH) + 1, c.get(Calendar.DATE), --- 343,347 ---- # seeded with Java ! c = self.calendar() o = zxJDBC.TimestampFromTicks(c.getTime().getTime() / 1000L) v = zxJDBC.Timestamp(c.get(Calendar.YEAR), c.get(Calendar.MONTH) + 1, c.get(Calendar.DATE), *************** *** 617,621 **** class LOBTest(zxJDBCTestCase): ! def __blob(self, obj=0): assert self.has_table("blobtable"), "no blob table" tabname, sql = self.table("blobtable") --- 686,690 ---- class LOBTest(zxJDBCTestCase): ! def _test_blob(self, obj=0): assert self.has_table("blobtable"), "no blob table" tabname, sql = self.table("blobtable") *************** *** 670,680 **** def testBLOBAsString(self): """testing BLOB as string""" ! self.__blob() def testBLOBAsPyFile(self): """testing BLOB as PyFile""" ! self.__blob(1) ! def __clob(self, asfile=0): assert self.has_table("clobtable"), "no clob table" tabname, sql = self.table("clobtable") --- 739,749 ---- def testBLOBAsString(self): """testing BLOB as string""" ! self._test_blob() def testBLOBAsPyFile(self): """testing BLOB as PyFile""" ! self._test_blob(1) ! def _test_clob(self, asfile=0): assert self.has_table("clobtable"), "no clob table" tabname, sql = self.table("clobtable") *************** *** 712,720 **** def testCLOBAsString(self): """testing CLOB as string""" ! self.__clob(0) def testCLOBAsPyFile(self): """testing CLOB as PyFile""" ! self.__clob(1) class BCPTestCase(zxJDBCTestCase): --- 781,789 ---- def testCLOBAsString(self): """testing CLOB as string""" ! self._test_clob(0) def testCLOBAsPyFile(self): """testing CLOB as PyFile""" ! self._test_clob(1) class BCPTestCase(zxJDBCTestCase): *************** *** 747,772 **** cnt = Pipe().pipe(dbSource, csvSink) - 1 # ignore the header row - - finally: - writer.close() - src.close() - os.remove(fn) - - def _testXMLPipe(self): - """testing the XML pipe""" - from java.io import PrintWriter, FileWriter - from com.ziclix.python.sql.pipe import Pipe - from com.ziclix.python.sql.pipe.db import DBSource - from com.ziclix.python.sql.pipe.xml import XMLSink - - try: - src = self.connect() - fn = tempfile.mktemp(suffix="csv") - writer = PrintWriter(FileWriter(fn)) - xmlSink = XMLSink(writer) - - dbSource = DBSource(src, c.datahandler.__class__, "zxtesting", None, None, None) - - cnt = Pipe().pipe(dbSource, xmlSink) - 1 # ignore the header row finally: --- 816,819 ---- |