134 lines
5.0 KiB
Python
134 lines
5.0 KiB
Python
"""Module for testing LOB (CLOB and BLOB) variables."""
|
|
|
|
class TestLobVar(BaseTestCase):
|
|
|
|
def __PerformTest(self, type, inputType):
|
|
if type == "CLOB":
|
|
longString = u""
|
|
else:
|
|
longString = ""
|
|
directType = getattr(cx_Oracle, type)
|
|
self.cursor.execute(u"truncate table Test%ss" % type)
|
|
for i in range(0, 11):
|
|
if i > 0:
|
|
if type == "CLOB":
|
|
char = unichr(ord('A') + i - 1)
|
|
else:
|
|
char = chr(ord('A') + i - 1)
|
|
longString += char * 25000
|
|
elif inputType != directType:
|
|
continue
|
|
self.cursor.setinputsizes(longString = inputType)
|
|
self.cursor.execute(u"""
|
|
insert into Test%ss (
|
|
IntCol,
|
|
%sCol
|
|
) values (
|
|
:integerValue,
|
|
:longString
|
|
)""" % (type, type),
|
|
integerValue = i,
|
|
longString = longString)
|
|
self.connection.commit()
|
|
self.cursor.execute(u"""
|
|
select *
|
|
from Test%ss
|
|
order by IntCol""" % type)
|
|
longString = ""
|
|
for row in self.cursor:
|
|
integerValue, lob = row
|
|
if integerValue == 0:
|
|
self.failUnlessEqual(lob.size(), 0)
|
|
self.failUnlessEqual(lob.read(), "")
|
|
else:
|
|
if type == "CLOB":
|
|
char = unichr(ord('A') + integerValue - 1)
|
|
prevChar = unichr(ord('A') + integerValue - 2)
|
|
actualValue = unicode(lob)
|
|
else:
|
|
char = chr(ord('A') + integerValue - 1)
|
|
prevChar = chr(ord('A') + integerValue - 2)
|
|
actualValue = str(lob)
|
|
longString += char * 25000
|
|
self.failUnlessEqual(lob.size(), len(longString))
|
|
self.failUnlessEqual(lob.read(), longString)
|
|
self.failUnlessEqual(actualValue, longString)
|
|
self.failUnlessEqual(lob.read(len(longString)), char)
|
|
if integerValue > 1:
|
|
offset = (integerValue - 1) * 25000 - 4
|
|
string = prevChar * 5 + char * 5
|
|
self.failUnlessEqual(lob.read(offset, 10), string)
|
|
|
|
def __TestTrim(self, type):
|
|
self.cursor.execute(u"truncate table Test%ss" % type)
|
|
self.cursor.setinputsizes(longString = getattr(cx_Oracle, type))
|
|
longString = "X" * 75000
|
|
if type == "CLOB":
|
|
longString = unicode(longString)
|
|
self.cursor.execute(u"""
|
|
insert into Test%ss (
|
|
IntCol,
|
|
%sCol
|
|
) values (
|
|
:integerValue,
|
|
:longString
|
|
)""" % (type, type),
|
|
integerValue = 1,
|
|
longString = longString)
|
|
self.cursor.execute(u"""
|
|
select %sCol
|
|
from Test%ss
|
|
where IntCol = 1""" % (type, type))
|
|
lob, = self.cursor.fetchone()
|
|
self.failUnlessEqual(lob.size(), 75000)
|
|
lob.trim(25000)
|
|
self.failUnlessEqual(lob.size(), 25000)
|
|
lob.trim()
|
|
self.failUnlessEqual(lob.size(), 0)
|
|
|
|
def testBLOBsIndirect(self):
|
|
"test binding and fetching BLOB data (indirectly)"
|
|
self.__PerformTest("BLOB", cx_Oracle.LONG_BINARY)
|
|
|
|
def testCLOBsIndirect(self):
|
|
"test binding and fetching CLOB data (indirectly)"
|
|
self.__PerformTest("CLOB", cx_Oracle.LONG_STRING)
|
|
|
|
def testBLOBsDirect(self):
|
|
"test binding and fetching BLOB data (directly)"
|
|
self.__PerformTest("BLOB", cx_Oracle.BLOB)
|
|
|
|
def testCLOBsDirect(self):
|
|
"test binding and fetching CLOB data (directly)"
|
|
self.__PerformTest("CLOB", cx_Oracle.CLOB)
|
|
|
|
def testCLOBCursorDescription(self):
|
|
"test cursor description is accurate for CLOBs"
|
|
self.cursor.execute(u"select * from TestCLOBs")
|
|
self.failUnlessEqual(self.cursor.description,
|
|
[ (u'INTCOL', cx_Oracle.NUMBER, 10, 22, 9, 0, 0),
|
|
(u'CLOBCOL', cx_Oracle.CLOB, -1, 4000, 0, 0, 0) ])
|
|
|
|
def testBLOBCursorDescription(self):
|
|
"test cursor description is accurate for BLOBs"
|
|
self.cursor.execute(u"select * from TestBLOBs")
|
|
self.failUnlessEqual(self.cursor.description,
|
|
[ (u'INTCOL', cx_Oracle.NUMBER, 10, 22, 9, 0, 0),
|
|
(u'BLOBCOL', cx_Oracle.BLOB, -1, 4000, 0, 0, 0) ])
|
|
|
|
def testBLOBTrim(self):
|
|
"test trimming a BLOB"
|
|
self.__TestTrim("BLOB")
|
|
|
|
def testCLOBTrim(self):
|
|
"test trimming a CLOB"
|
|
self.__TestTrim("CLOB")
|
|
|
|
def testMultipleFetch(self):
|
|
"test retrieving data from a CLOB after multiple fetches"
|
|
self.cursor.arraysize = 1
|
|
self.cursor.execute(u"select CLOBCol from TestCLOBS")
|
|
rows = self.cursor.fetchall()
|
|
self.failUnlessRaises(cx_Oracle.ProgrammingError, rows[1][0].read)
|
|
|