290 lines
11 KiB
Python
290 lines
11 KiB
Python
#------------------------------------------------------------------------------
|
|
# Copyright (c) 2016, 2020, Oracle and/or its affiliates. All rights reserved.
|
|
#
|
|
# Portions Copyright 2007-2015, Anthony Tuininga. All rights reserved.
|
|
#
|
|
# Portions Copyright 2001-2007, Computronix (Canada) Ltd., Edmonton, Alberta,
|
|
# Canada. All rights reserved.
|
|
#------------------------------------------------------------------------------
|
|
|
|
"""
|
|
1900 - Module for testing LOB (CLOB and BLOB) variables
|
|
"""
|
|
|
|
import test_env
|
|
|
|
import cx_Oracle as oracledb
|
|
|
|
class TestCase(test_env.BaseTestCase):
|
|
|
|
def __get_temp_lobs(self, sid):
|
|
cursor = self.connection.cursor()
|
|
cursor.execute("""
|
|
select abstract_lobs
|
|
from v$temporary_lobs
|
|
where sid = :sid""", sid = sid)
|
|
row = cursor.fetchone()
|
|
if row is None:
|
|
return 0
|
|
return int(row[0])
|
|
|
|
def __perform_test(self, lob_type, input_type):
|
|
long_string = ""
|
|
db_type = getattr(oracledb, "DB_TYPE_" + lob_type)
|
|
self.cursor.execute("truncate table Test%ss" % lob_type)
|
|
for i in range(0, 11):
|
|
if i > 0:
|
|
char = chr(ord('A') + i - 1)
|
|
long_string += char * 25000
|
|
elif input_type is not db_type:
|
|
continue
|
|
self.cursor.setinputsizes(long_string = input_type)
|
|
if lob_type == "BLOB":
|
|
bind_value = long_string.encode()
|
|
else:
|
|
bind_value = long_string
|
|
self.cursor.execute("""
|
|
insert into Test%ss (
|
|
IntCol,
|
|
%sCol
|
|
) values (
|
|
:integer_value,
|
|
:long_string
|
|
)""" % (lob_type, lob_type),
|
|
integer_value=i,
|
|
long_string=bind_value)
|
|
self.connection.commit()
|
|
self.cursor.execute("""
|
|
select *
|
|
from Test%ss
|
|
order by IntCol""" % lob_type)
|
|
self.__validate_query(self.cursor, lob_type)
|
|
|
|
def __test_lob_operations(self, lob_type):
|
|
self.cursor.execute("truncate table Test%ss" % lob_type)
|
|
self.cursor.setinputsizes(long_string=getattr(oracledb, lob_type))
|
|
long_string = "X" * 75000
|
|
write_value = "TEST"
|
|
if lob_type == "BLOB":
|
|
long_string = long_string.encode("ascii")
|
|
write_value = write_value.encode("ascii")
|
|
self.cursor.execute("""
|
|
insert into Test%ss (
|
|
IntCol,
|
|
%sCol
|
|
) values (
|
|
:integer_value,
|
|
:long_string
|
|
)""" % (lob_type, lob_type),
|
|
integer_value=1,
|
|
long_string=long_string)
|
|
self.cursor.execute("""
|
|
select %sCol
|
|
from Test%ss
|
|
where IntCol = 1""" % (lob_type, lob_type))
|
|
lob, = self.cursor.fetchone()
|
|
self.assertEqual(lob.isopen(), False)
|
|
lob.open()
|
|
self.assertEqual(lob.isopen(), True)
|
|
lob.close()
|
|
self.assertEqual(lob.isopen(), False)
|
|
self.assertEqual(lob.size(), 75000)
|
|
lob.write(write_value, 75001)
|
|
self.assertEqual(lob.size(), 75000 + len(write_value))
|
|
self.assertEqual(lob.read(), long_string + write_value)
|
|
lob.write(write_value, 1)
|
|
self.assertEqual(lob.read(),
|
|
write_value + long_string[4:] + write_value)
|
|
lob.trim(25000)
|
|
self.assertEqual(lob.size(), 25000)
|
|
lob.trim()
|
|
self.assertEqual(lob.size(), 0)
|
|
|
|
def __test_temporary_lob(self, lob_type):
|
|
self.cursor.execute("truncate table Test%ss" % lob_type)
|
|
value = "A test string value"
|
|
if lob_type == "BLOB":
|
|
value = value.encode("ascii")
|
|
db_type = getattr(oracledb, "DB_TYPE_" + lob_type)
|
|
lob = self.connection.createlob(db_type)
|
|
lob.write(value)
|
|
self.cursor.execute("""
|
|
insert into Test%ss (IntCol, %sCol)
|
|
values (:int_val, :lob_val)""" % (lob_type, lob_type),
|
|
int_val=1,
|
|
lob_val=lob)
|
|
self.cursor.execute("select %sCol from Test%ss" % (lob_type, lob_type))
|
|
lob, = self.cursor.fetchone()
|
|
self.assertEqual(lob.read(), value)
|
|
|
|
def __validate_query(self, rows, lob_type):
|
|
long_string = ""
|
|
db_type = getattr(oracledb, "DB_TYPE_" + lob_type)
|
|
for row in rows:
|
|
integer_value, lob = row
|
|
self.assertEqual(lob.type, db_type)
|
|
if integer_value == 0:
|
|
self.assertEqual(lob.size(), 0)
|
|
expected_value = ""
|
|
if lob_type == "BLOB":
|
|
expected_value = expected_value.encode()
|
|
self.assertEqual(lob.read(), expected_value)
|
|
else:
|
|
char = chr(ord('A') + integer_value - 1)
|
|
prev_char = chr(ord('A') + integer_value - 2)
|
|
long_string += char * 25000
|
|
if lob_type == "BLOB":
|
|
actualValue = long_string.encode("ascii")
|
|
char = char.encode("ascii")
|
|
prev_char = prev_char.encode("ascii")
|
|
else:
|
|
actualValue = long_string
|
|
self.assertEqual(lob.size(), len(actualValue))
|
|
self.assertEqual(lob.read(), actualValue)
|
|
if lob_type == "CLOB":
|
|
self.assertEqual(str(lob), actualValue)
|
|
self.assertEqual(lob.read(len(actualValue)), char)
|
|
if integer_value > 1:
|
|
offset = (integer_value - 1) * 25000 - 4
|
|
string = prev_char * 5 + char * 5
|
|
self.assertEqual(lob.read(offset, 10), string)
|
|
|
|
def test_1900_bind_lob_value(self):
|
|
"1900 - test binding a LOB value directly"
|
|
self.cursor.execute("truncate table TestCLOBs")
|
|
self.cursor.execute("insert into TestCLOBs values (1, 'Short value')")
|
|
self.cursor.execute("select ClobCol from TestCLOBs")
|
|
lob, = self.cursor.fetchone()
|
|
self.cursor.execute("insert into TestCLOBs values (2, :value)",
|
|
value=lob)
|
|
|
|
def test_1901_blob_cursor_description(self):
|
|
"1901 - test cursor description is accurate for BLOBs"
|
|
self.cursor.execute("select * from TestBLOBs")
|
|
expected_value = [
|
|
('INTCOL', oracledb.DB_TYPE_NUMBER, 10, None, 9, 0, 0),
|
|
('BLOBCOL', oracledb.DB_TYPE_BLOB, None, None, None, None, 0)
|
|
]
|
|
self.assertEqual(self.cursor.description, expected_value)
|
|
|
|
def test_1902_blob_direct(self):
|
|
"1902 - test binding and fetching BLOB data (directly)"
|
|
self.__perform_test("BLOB", oracledb.DB_TYPE_BLOB)
|
|
|
|
def test_1903_blob_indirect(self):
|
|
"1903 - test binding and fetching BLOB data (indirectly)"
|
|
self.__perform_test("BLOB", oracledb.DB_TYPE_LONG_RAW)
|
|
|
|
def test_1904_blob_operations(self):
|
|
"1904 - test operations on BLOBs"
|
|
self.__test_lob_operations("BLOB")
|
|
|
|
def test_1905_clob_cursor_description(self):
|
|
"1905 - test cursor description is accurate for CLOBs"
|
|
self.cursor.execute("select * from TestCLOBs")
|
|
expected_value = [
|
|
('INTCOL', oracledb.DB_TYPE_NUMBER, 10, None, 9, 0, 0),
|
|
('CLOBCOL', oracledb.DB_TYPE_CLOB, None, None, None, None, 0)
|
|
]
|
|
self.assertEqual(self.cursor.description, expected_value)
|
|
|
|
def test_1906_clob_direct(self):
|
|
"1906 - test binding and fetching CLOB data (directly)"
|
|
self.__perform_test("CLOB", oracledb.DB_TYPE_CLOB)
|
|
|
|
def test_1907_clob_indirect(self):
|
|
"1907 - test binding and fetching CLOB data (indirectly)"
|
|
self.__perform_test("CLOB", oracledb.DB_TYPE_LONG)
|
|
|
|
def test_1908_clob_operations(self):
|
|
"1908 - test operations on CLOBs"
|
|
self.__test_lob_operations("CLOB")
|
|
|
|
def test_1909_create_temp_blob(self):
|
|
"1909 - test creating a temporary BLOB"
|
|
self.__test_temporary_lob("BLOB")
|
|
|
|
def test_1910_create_temp_clob(self):
|
|
"1910 - test creating a temporary CLOB"
|
|
self.__test_temporary_lob("CLOB")
|
|
|
|
def test_1911_create_temp_nclob(self):
|
|
"1911 - test creating a temporary NCLOB"
|
|
self.__test_temporary_lob("NCLOB")
|
|
|
|
def test_1912_multiple_fetch(self):
|
|
"1912 - test retrieving data from a CLOB after multiple fetches"
|
|
self.cursor.arraysize = 1
|
|
self.cursor.execute("select * from TestCLOBS")
|
|
rows = self.cursor.fetchall()
|
|
self.__validate_query(rows, "CLOB")
|
|
|
|
def test_1913_nclob_cursor_description(self):
|
|
"1913 - test cursor description is accurate for NCLOBs"
|
|
self.cursor.execute("select * from TestNCLOBs")
|
|
expected_value = [
|
|
('INTCOL', oracledb.DB_TYPE_NUMBER, 10, None, 9, 0, 0),
|
|
('NCLOBCOL', oracledb.DB_TYPE_NCLOB, None, None, None, None, 0)
|
|
]
|
|
self.assertEqual(self.cursor.description, expected_value)
|
|
|
|
def test_1914_nclob_direct(self):
|
|
"1914 - test binding and fetching NCLOB data (directly)"
|
|
self.__perform_test("NCLOB", oracledb.DB_TYPE_NCLOB)
|
|
|
|
def test_1915_nclob_different_encodings(self):
|
|
"1915 - test binding and fetching NCLOB data (different encodings)"
|
|
connection = oracledb.connect(test_env.get_main_user(),
|
|
test_env.get_main_password(),
|
|
test_env.get_connect_string(),
|
|
encoding="UTF-8", nencoding="UTF-16")
|
|
value = "\u03b4\u4e2a"
|
|
cursor = connection.cursor()
|
|
cursor.execute("truncate table TestNCLOBs")
|
|
cursor.setinputsizes(val=oracledb.DB_TYPE_NVARCHAR)
|
|
cursor.execute("insert into TestNCLOBs values (1, :val)", val=value)
|
|
cursor.execute("select NCLOBCol from TestNCLOBs")
|
|
nclob, = cursor.fetchone()
|
|
cursor.setinputsizes(val=oracledb.DB_TYPE_NVARCHAR)
|
|
cursor.execute("update TestNCLOBs set NCLOBCol = :val",
|
|
val=nclob.read() + value)
|
|
cursor.execute("select NCLOBCol from TestNCLOBs")
|
|
nclob, = cursor.fetchone()
|
|
self.assertEqual(nclob.read(), value + value)
|
|
|
|
def test_1916_nclob_indirect(self):
|
|
"1916 - test binding and fetching NCLOB data (indirectly)"
|
|
self.__perform_test("NCLOB", oracledb.DB_TYPE_LONG)
|
|
|
|
def test_1917_nclob_operations(self):
|
|
"1917 - test operations on NCLOBs"
|
|
self.__test_lob_operations("NCLOB")
|
|
|
|
def test_1918_temporary_lobs(self):
|
|
"1918 - test temporary LOBs"
|
|
cursor = self.connection.cursor()
|
|
cursor.arraysize = self.cursor.arraysize
|
|
cursor.execute("""
|
|
select sys_context('USERENV', 'SID')
|
|
from dual""")
|
|
sid, = cursor.fetchone()
|
|
temp_lobs = self.__get_temp_lobs(sid)
|
|
self.assertEqual(temp_lobs, 0)
|
|
cursor.execute("""
|
|
select extract(xmlcol, '/').getclobval()
|
|
from TestXML""")
|
|
for lob, in cursor:
|
|
value = lob.read()
|
|
del lob
|
|
cursor.close()
|
|
temp_lobs = self.__get_temp_lobs(sid)
|
|
self.assertEqual(temp_lobs, 0)
|
|
|
|
def test_1919_AssignStringBeyondArraySize(self):
|
|
"1919 - test assign string to NCLOB beyond array size"
|
|
nclobVar = self.cursor.var(oracledb.DB_TYPE_NCLOB)
|
|
self.assertRaises(IndexError, nclobVar.setvalue, 1, "test char")
|
|
|
|
if __name__ == "__main__":
|
|
test_env.run_test_cases()
|