zxtest.py

来自「mallet是自然语言处理、机器学习领域的一个开源项目。」· Python 代码 · 共 1,243 行 · 第 1/3 页

			c.arraysize = -1
			f = c.fetchmany()
			assert len(f) == 7, "expecting [7] rows, got [%d]" % (len(f))
		finally:
			c.close()

	def testStaticFetching(self):
		"""testing various static fetch methods"""
		self._test_fetching(0)

	def testDynamicFetching(self):
		"""testing various dynamic fetch methods"""
		self._test_fetching(1)

	def testFetchingBeforeExecute(self):
		"""testing fetch methods before execution"""
		c = self.cursor()
		try:
			try:
				c.fetchall()
			except zxJDBC.Error, e:
				pass
			else:
				self.fail("excepted exception calling fetchall() prior to execute()")
		finally:
			c.close()

	def testBindingsWithNoParams(self):
		"""testing bindings with no params"""
		c = self.cursor()
		try:
			self.assertRaises(zxJDBC.ProgrammingError, c.execute, "select * from zxtesting", {0:zxJDBC.INTEGER})

			# test an inappropriate value for a binding
			self.assertRaises(zxJDBC.ProgrammingError, c.execute, "select * from zxtesting", {0:{}})
		finally:
			c.close()

	def testDynamicCursor(self):
		"""testing dynamic cursor queries"""
		c = self.cursor(1)
		try:
			c.execute("select * from zxtesting")
			f = c.fetchmany(4)
			assert len(f) == 4, "expected [4] rows, got [%d]" % (len(f))
		finally:
			c.close()

	def testRowid(self):
		"""testing the autoincrement facilities of the different handlers"""
		assert self.has_table("autoincrementtable"), "no autoincrement table"

		c = self.cursor()
		assert c.lastrowid == None, "expected initial lastrowid to be None"

		try:

			tabname, sql = self.table("autoincrementtable")
			c.execute(sql)

			c.execute("insert into %s (b) values (?)" % (tabname), [(0,)])
			assert c.lastrowid is not None, "lastrowid is None"

			try:
				for idx in range(c.lastrowid + 1, c.lastrowid + 25):
					c.execute("insert into %s (b) values (?)" % (tabname), [(idx,)])
					assert c.lastrowid is not None, "lastrowid is None"
					self.assertEquals(idx, c.lastrowid)
			except:
				self.db.rollback()

		finally:
			if self.has_table("post_autoincrementtable"):
				try:
					sequence, sql = self.table("post_autoincrementtable")
					c.execute(sql)
					self.db.commit()
				except:
					self.db.rollback()

			try:
				c.execute("drop table %s" % (tabname))
				self.db.commit()
			except:
				self.db.rollback()

			self.db.commit()
			c.close()

	def _test_fetchapi(self, dynamic=0):
		"""Test the public Java API for Fetch"""
		from com.ziclix.python.sql import Fetch, WarningListener
		cur = self.cursor()
		try:
			c = self.db.__connection__
			stmt = c.prepareStatement("select * from zxtesting where id < ?")
			stmt.setInt(1, 5)
			rs = stmt.executeQuery()
			fetch = Fetch.newFetch(cur.datahandler, dynamic)
			class WL(WarningListener):
				def warning(self, event):
					raise event.getWarning()
			wl = WL()
			fetch.addWarningListener(wl)
			# the RS is closed by Fetch
			fetch.add(rs)

			if not dynamic:
				self.assertEquals(4, fetch.getRowCount())
			assert fetch.fetchone()
			assert fetch.fetchmany(2)
			assert fetch.fetchall()
			assert not fetch.fetchall()
			self.assertEquals(4, fetch.getRowCount())
			assert fetch.removeWarningListener(wl)
			fetch.close()
			stmt.close()
		finally:
			cur.close()

	def testStaticFetchAPI(self):
		"""Test static Java Fetch API"""
		self._test_fetchapi(0)

	def testDynamicFetchAPI(self):
		"""Test dynamic Java Fetch API"""
		self._test_fetchapi(1)

class LOBTestCase(zxJDBCTestCase):

	def _test_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:

			hello = ("hello",) * 1024

			c.execute(sql)
			self.db.commit()

			from java.io import FileOutputStream, FileInputStream, ObjectOutputStream, ObjectInputStream, ByteArrayInputStream

			fp = FileOutputStream(fn)
			oos = ObjectOutputStream(fp)
			oos.writeObject(hello)
			fp.close()

			fp = FileInputStream(fn)
			blob = ObjectInputStream(fp)
			value = blob.readObject()
			fp.close()
			assert hello == value, "unable to serialize properly"

			if obj == 1:
				fp = open(fn, "rb")
			else:
				fp = FileInputStream(fn)

			c.execute("insert into %s (a, b) values (?, ?)" % (tabname), [(0, fp)], {1:zxJDBC.BLOB})
			self.db.commit()

			c.execute("select * from %s" % (tabname))
			f = c.fetchall()
			bytes = f[0][1]
			blob = ObjectInputStream(ByteArrayInputStream(bytes)).readObject()

			assert hello == blob, "blobs are not equal"

		finally:
			c.execute("drop table %s" % (tabname))
			c.close()
			self.db.commit()
			if os.path.exists(fn):
				if fp:
					fp.close()
				os.remove(fn)

	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")

		c = self.cursor()
		try:
			hello = "hello" * 1024 * 10

			c.execute(sql)
			self.db.commit()

			if asfile:
				fp = open(tempfile.mktemp(), "w")
				fp.write(hello)
				fp.flush()
				fp.close()
				obj = open(fp.name, "r")
			else:
				obj = hello

			c.execute("insert into %s (a, b) values (?, ?)" % (tabname), [(0, obj)], {1:zxJDBC.CLOB})
			c.execute("select * from %s" % (tabname), maxrows=1)
			f = c.fetchall()
			assert len(f) == 1, "expected [%d], got [%d]" % (1, len(f))
			assert hello == f[0][1], "clobs are not equal"

		finally:
			c.execute("drop table %s" % (tabname))
			c.close()
			self.db.commit()
			if asfile:
				obj.close()
				os.remove(obj.name)

	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):

	def testCSVPipe(self):
		"""testing the CSV 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.csv import CSVSink

		try:
			src = self.connect()
			fn = tempfile.mktemp(suffix="csv")
			writer = PrintWriter(FileWriter(fn))
			csvSink = CSVSink(writer)

			c = self.cursor()
			try:
				c.execute("insert into zxtesting (id, name, state) values (?, ?, ?)", [(1000, 'this,has,a,comma', 'and a " quote')])
				c.execute("insert into zxtesting (id, name, state) values (?, ?, ?)", [(1001, 'this,has,a,comma and a "', 'and a " quote')])
				# ORACLE has a problem calling stmt.setObject(index, null)
				c.execute("insert into zxtesting (id, name, state) values (?, ?, ?)", [(1010, '"this,has,a,comma"', None)], {2:zxJDBC.VARCHAR})
				self.db.commit()
			finally:
				self.db.rollback()
				c.close()

			dbSource = DBSource(src, c.datahandler.__class__, "zxtesting", None, None, None)

			cnt = Pipe().pipe(dbSource, csvSink) - 1 # ignore the header row

		finally:
			writer.close()
			src.close()
			os.remove(fn)

	def testDBPipe(self):
		"""testing the DB pipe"""
		from com.ziclix.python.sql.pipe import Pipe
		from com.ziclix.python.sql.pipe.db import DBSource, DBSink

		try:
			src = self.connect()
			dst = self.connect()

			c = self.cursor()
			c.execute("create table zxtestingbcp (id int not null, name varchar(20), state varchar(2), primary key (id))")
			self.db.commit()

			c.execute("select count(*) from zxtesting")
			one = c.fetchone()[0]
			c.close()

			dbSource = DBSource(src, c.datahandler.__class__, "zxtesting", None, None, None)
			dbSink = DBSink(dst, c.datahandler.__class__, "zxtestingbcp", None, None, 1)

			cnt = Pipe().pipe(dbSource, dbSink) - 1 # ignore the header row

			c = self.cursor()
			c.execute("select count(*) from zxtestingbcp")
			two = c.fetchone()[0]
			c.execute("delete from zxtestingbcp")
			self.db.commit()
			c.close()

			assert one == two, "expected [%d] rows in destination, got [%d] (sql)" % (one, two)
			assert one == cnt, "expected [%d] rows in destination, got [%d] (bcp)" % (one, cnt)

			# this tests the internal assert in BCP.  we need to handle the case where we exclude
			# all the rows queried (based on the fact no columns exist) but rows were fetched
			# also make sure (eg, Oracle) that the column name case is ignored
			dbSource = DBSource(src, c.datahandler.__class__, "zxtesting", None, ["id"], None)
			dbSink = DBSink(dst, c.datahandler.__class__, "zxtestingbcp", ["id"], None, 1)

			self.assertRaises(zxJDBC.Error, Pipe().pipe, dbSource, dbSink)

			params = [(4,)]

			dbSource = DBSource(src, c.datahandler.__class__, "zxtesting", "id > ?", None, params)
			dbSink = DBSink(dst, c.datahandler.__class__, "zxtestingbcp", None, None, 1)

			cnt = Pipe().pipe(dbSource, dbSink) - 1 # ignore the header row

			c = self.cursor()
			c.execute("select count(*) from zxtesting where id > ?", params)
			one = c.fetchone()[0]
			c.execute("select count(*) from zxtestingbcp")
			two = c.fetchone()[0]
			c.close()
			assert one == two, "expected [%d] rows in destination, got [%d] (sql)" % (one, two)
			assert one == cnt, "expected [%d] rows in destination, got [%d] (bcp)" % (one, cnt)

		finally:
			try:
				c = self.cursor()
				try:
					c.execute("drop table zxtestingbcp")
					self.db.commit()
				except:
					self.db.rollback()
			finally:
				c.close()

			try:
				src.close()
			except:
				src = None
			try:
				dst.close()
			except:
				dst = None

	def testBCP(self):
		"""testing bcp parameters and functionality"""
		from com.ziclix.python.sql.util import BCP

		import dbexts

		try:
			src = self.connect()
			dst = self.connect()

			c = self.cursor()
			c.execute("create table zxtestingbcp (id int not null, name varchar(20), state varchar(2), primary key (id))")
			self.db.commit()

			c.execute("select count(*) from zxtesting")
			one = c.fetchone()[0]
			c.close()

			b = BCP(src, dst)
			if hasattr(self, "datahandler"):
				b.sourceDataHandler = self.datahandler
				b.destinationDataHandler = self.datahandler
			cnt = b.bcp("zxtesting", toTable="zxtestingbcp")

			c = self.cursor()
			c.execute("select count(*) from zxtestingbcp")
			two = c.fetchone()[0]
			c.execute("delete from zxtestingbcp")
			self.db.commit()
			c.close()

			assert one == two, "expected [%d] rows in destination, got [%d] (sql)" % (one, two)
			assert one == cnt, "expected [%d] rows in destination, got [%d] (bcp)" % (one, cnt)

			# this tests the internal assert in BCP.  we need to handle the case where we exclude
			# all the rows queried (based on the fact no columns exist) but rows were fetched
			# also make sure (eg, Oracle) that the column name case is ignored
			self.assertRaises(zxJDBC.Error, b.bcp, "zxtesting", toTable="zxtestingbcp", include=["id"], exclude=["id"])

			params = [(4,)]
			cnt = b.bcp("zxtesting", "id > ?", params, toTable="zxtestingbcp")

			c = self.cursor()
			c.execute("select count(*) from zxtesting where id > ?", params)
			one = c.fetchone()[0]
			c.execute("select count(*) from zxtestingbcp")
			two = c.fetchone()[0]
			c.close()
			assert one == two, "expected [%d] rows in destination, got [%d] (sql)" % (one, two)
			assert one == cnt, "expected [%d] rows in destination, got [%d] (bcp)" % (one, cnt)

		finally:
			try:
				c = self.cursor()
				try:
					c.execute("drop table zxtestingbcp")
					self.db.commit()
				except:
					self.db.rollback()
			finally:
				c.close()

			try:
				src.close()
			except:
				src = None
			try:
				dst.close()
			except:
				dst = None