{{ message }}
forked from oracle/python-cx_Oracle
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_1900_lob_var.py
More file actions
284 lines (255 loc) · 11.2 KB
/
Copy pathtest_1900_lob_var.py
File metadata and controls
284 lines (255 loc) · 11.2 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
#------------------------------------------------------------------------------
# Copyright (c) 2016, 2021, 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 cx_Oracle as oracledb
import test_env
class TestCase(test_env.BaseTestCase):
def __get_temp_lobs(self, sid):
cursor = self.connection.cursor()
cursor.execute("""
select cache_lobs + nocache_lobs + 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":
expected_value = long_string.encode("ascii")
char = char.encode("ascii")
prev_char = prev_char.encode("ascii")
else:
expected_value = long_string
self.assertEqual(lob.size(), len(expected_value))
self.assertEqual(lob.read(), expected_value)
if lob_type == "CLOB":
self.assertEqual(str(lob), expected_value)
self.assertEqual(lob.read(len(expected_value)), 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, False),
('CLOBCOL', oracledb.DB_TYPE_CLOB, None, None, None, None, False)
]
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_non_ascii_chars(self):
"1915 - test binding and fetching NCLOB data (with non-ASCII chars)"
value = "\u03b4\u4e2a"
self.cursor.execute("truncate table TestNCLOBs")
self.cursor.setinputsizes(val=oracledb.DB_TYPE_NVARCHAR)
self.cursor.execute("insert into TestNCLOBs values (1, :val)",
val=value)
self.cursor.execute("select NCLOBCol from TestNCLOBs")
nclob, = self.cursor.fetchone()
self.cursor.setinputsizes(val=oracledb.DB_TYPE_NVARCHAR)
self.cursor.execute("update TestNCLOBs set NCLOBCol = :val",
val=nclob.read() + value)
self.cursor.execute("select NCLOBCol from TestNCLOBs")
nclob, = self.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_assign_string_beyond_array_size(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()
You can’t perform that action at this time.