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
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
|
#------------------------------------------------------------------------------
# Copyright (c) 2020, 2022, Oracle and/or its affiliates.
#
# This software is dual-licensed to you under the Universal Permissive License
# (UPL) 1.0 as shown at https://oss.oracle.com/licenses/upl and Apache License
# 2.0 as shown at http://www.apache.org/licenses/LICENSE-2.0. You may choose
# either license.
#
# If you elect to accept the software under the Apache License, Version 2.0,
# the following applies:
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#------------------------------------------------------------------------------
"""
3900 - Module for testing the cursor execute() method
"""
import oracledb
import test_env
class TestCase(test_env.BaseTestCase):
def test_3900_execute_no_args(self):
"3900 - test executing a statement without any arguments"
result = self.cursor.execute("begin null; end;")
self.assertEqual(result, None)
def test_3901_execute_no_statement_with_args(self):
"3901 - test executing a None statement with bind variables"
self.assertRaisesRegex(oracledb.ProgrammingError, "^DPY-2001:",
self.cursor.execute, None, x=5)
def test_3902_execute_empty_keyword_args(self):
"3902 - test executing a statement with args and empty keyword args"
simple_var = self.cursor.var(oracledb.NUMBER)
args = [simple_var]
kwargs = {}
result = self.cursor.execute("begin :1 := 25; end;", args, **kwargs)
self.assertEqual(result, None)
self.assertEqual(simple_var.getvalue(), 25)
def test_3903_execute_keyword_args(self):
"3903 - test executing a statement with keyword arguments"
simple_var = self.cursor.var(oracledb.NUMBER)
result = self.cursor.execute("begin :value := 5; end;",
value=simple_var)
self.assertEqual(result, None)
self.assertEqual(simple_var.getvalue(), 5)
def test_3904_execute_dictionary_arg(self):
"3904 - test executing a statement with a dictionary argument"
simple_var = self.cursor.var(oracledb.NUMBER)
dict_arg = dict(value=simple_var)
result = self.cursor.execute("begin :value := 10; end;", dict_arg)
self.assertEqual(result, None)
self.assertEqual(simple_var.getvalue(), 10)
def test_3905_execute_multiple_arg_types(self):
"3905 - test executing a statement with both a dict and keyword args"
simple_var = self.cursor.var(oracledb.NUMBER)
dict_arg = dict(value=simple_var)
self.assertRaisesRegex(oracledb.ProgrammingError, "^DPY-2005:",
self.cursor.execute,
"begin :value := 15; end;", dict_arg,
value=simple_var)
def test_3906_execute_and_modify_array_size(self):
"3906 - test executing a statement and then changing the array size"
self.cursor.execute("select IntCol from TestNumbers")
self.cursor.arraysize = 20
self.assertEqual(len(self.cursor.fetchall()), 10)
def test_3907_bad_execute(self):
"3907 - test that subsequent executes succeed after bad execute"
sql = "begin raise_application_error(-20000, 'this); end;"
self.assertRaisesRegex(oracledb.DatabaseError,
"^ORA-06550:|^ORA-01756:", self.cursor.execute,
sql)
self.cursor.execute("begin null; end;")
def test_3908_fetch_after_bad_execute(self):
"3908 - test that subsequent fetches fail after bad execute"
self.assertRaisesRegex(oracledb.DatabaseError, "^ORA-00904:",
self.cursor.execute, "select y from dual")
self.assertRaisesRegex(oracledb.InterfaceError, "^DPY-1003:",
self.cursor.fetchall)
def test_3909_execute_bind_names_with_incorrect_bind(self):
"3909 - test executing a statement with an incorrect named bind"
statement = "select * from TestStrings where IntCol = :value"
self.assertRaisesRegex(oracledb.DatabaseError, "^DPY-4008:|^ORA-01036:",
self.cursor.execute, statement, value2=3)
def test_3910_execute_with_named_binds(self):
"3910 - test executing a statement with named binds"
statement = "select * from TestNumbers where IntCol = :value1 " + \
"and LongIntCol = :value2"
result = self.cursor.execute(statement, value1=1, value2=38)
self.assertEqual(len(result.fetchall()), 1)
def test_3911_execute_bind_position_with_incorrect_bind(self):
"3911 - test executing a statement with an incorrect positional bind"
statement = "select * from TestNumbers where IntCol = :value " + \
"and LongIntCol = :value2"
self.assertRaisesRegex(oracledb.DatabaseError,
"^DPY-4009:|^ORA-01008:",
self.cursor.execute, statement, [3])
def test_3912_execute_with_positional_binds(self):
"3912 - test executing a statement with positional binds"
statement = "select * from TestNumbers where IntCol = :value " + \
"and LongIntCol = :value2"
result = self.cursor.execute(statement, [1,38])
self.assertEqual(len(result.fetchall()), 1)
def test_3913_execute_with_rebinding_bind_name(self):
"3913 - test executing a statement after rebinding a named bind"
statement = "begin :value := :value2 + 5; end;"
simple_var = self.cursor.var(oracledb.NUMBER)
simple_var2 = self.cursor.var(oracledb.NUMBER)
simple_var2.setvalue(0, 5)
result = self.cursor.execute(statement, value=simple_var,
value2=simple_var2)
self.assertEqual(result, None)
self.assertEqual(simple_var.getvalue(), 10)
simple_var = self.cursor.var(oracledb.NATIVE_FLOAT)
simple_var2 = self.cursor.var(oracledb.NATIVE_FLOAT)
simple_var2.setvalue(0, 10)
result = self.cursor.execute(statement, value=simple_var,
value2=simple_var2)
self.assertEqual(result, None)
self.assertEqual(simple_var.getvalue(), 15)
def test_3914_bind_by_name_with_duplicates(self):
"3914 - test executing a PL/SQL statement with duplicate binds"
statement = "begin :value := :value + 5; end;"
simple_var = self.cursor.var(oracledb.NUMBER)
simple_var.setvalue(0, 5)
result = self.cursor.execute(statement, value=simple_var)
self.assertEqual(result, None)
self.assertEqual(simple_var.getvalue(), 10)
def test_3915_positional_bind_with_duplicates(self):
"3915 - test executing a PL/SQL statement with duplicate binds"
statement = "begin :value := :value + 5; end;"
simple_var = self.cursor.var(oracledb.NUMBER)
simple_var.setvalue(0, 5)
self.cursor.execute(statement, [simple_var])
self.assertEqual(simple_var.getvalue(), 10)
def test_3916_execute_with_incorrect_bind_values(self):
"3916 - test executing a statement with an incorrect number of binds"
statement = "begin :value := :value2 + 5; end;"
var = self.cursor.var(oracledb.NUMBER)
var.setvalue(0, 5)
self.assertRaisesRegex(oracledb.DatabaseError, "^DPY-4010:|^ORA-01008:",
self.cursor.execute, statement)
self.assertRaisesRegex(oracledb.DatabaseError, "^DPY-4010:|^ORA-01008:",
self.cursor.execute, statement, value=var)
self.assertRaisesRegex(oracledb.DatabaseError, "^DPY-4008:|^ORA-01036:",
self.cursor.execute, statement, value=var,
value2=var, value3=var)
def test_3917_change_in_size_on_successive_bind(self):
"3917 - change in size on subsequent binds does not use optimised path"
self.cursor.execute("truncate table TestTempTable")
data = [
(1, "Test String #1"),
(2, "ABC" * 100)
]
sql = "insert into TestTempTable (IntCol, StringCol1) values (:1, :2)"
for row in data:
self.cursor.execute(sql, row)
self.connection.commit()
self.cursor.execute("select IntCol, StringCol1 from TestTempTable")
self.assertEqual(self.cursor.fetchall(), data)
def test_3918_dml_can_use_optimised_path(self):
"3918 - test that dml can use optimised path"
data_to_insert = [
(1, "Test String #1"),
(2, "Test String #2"),
(3, "Test String #3")
]
self.cursor.execute("truncate table TestTempTable")
sql = "insert into TestTempTable (IntCol, StringCol1) values (:1, :2)"
for row in data_to_insert:
with self.connection.cursor() as cursor:
cursor.execute(sql, row)
self.connection.commit()
self.cursor.execute("""
select IntCol, StringCol1
from TestTempTable
order by IntCol""")
self.assertEqual(self.cursor.fetchall(), data_to_insert)
def test_3919_execute_with_invalid_parameters(self):
"3919 - test calling execute() with invalid parameters"
sql = "insert into TestTempTable (IntCol, StringCol1) values (:1, :2)"
self.assertRaisesRegex(oracledb.ProgrammingError, "^DPY-2003:",
self.cursor.execute, sql,
"These are not valid parameters")
def test_3920_execute_with_mixed_binds(self):
"3920 - test calling execute() with mixed binds"
self.cursor.execute("truncate table TestTempTable")
self.cursor.setinputsizes(None, None, str)
data = dict(val1=1, val2="Test String 1")
self.assertRaisesRegex(oracledb.ProgrammingError, "^DPY-2006:",
self.cursor.execute, """
insert into TestTempTable (IntCol, StringCol1)
values (:1, :2)
returning StringCol1 into :out_var""", data)
def test_3921_bind_by_name_with_double_quotes(self):
"3921 - test binding by name with double quotes"
sql = 'select :"_value1" + :"VaLue_2" + :"3VALUE" from dual'
data = {'"_value1"': 1, '"VaLue_2"': 2, '"3VALUE"': 3}
self.cursor.execute(sql, data)
result, = self.cursor.fetchone()
self.assertEqual(result, 6)
def test_3922_resize_buffer(self):
"3922 - test executing a statement with different input buffer sizes"
sql = """
insert into TestTempTable (IntCol, StringCol1, StringCol2)
values (:int_col, :str_val1, :str_val2) returning IntCol
into :ret_data"""
values1 = {"int_col" : 1, "str_val1": '{"a", "b"}', "str_val2" : None}
values2 = {"int_col" : 2, "str_val1": None, "str_val2" : '{"a", "b"}'}
values3 = {"int_col" : 3, "str_val1": '{"a"}', "str_val2" : None}
self.cursor.execute("truncate table TestTempTable")
ret_bind = self.cursor.var(oracledb.DB_TYPE_VARCHAR, arraysize=1)
self.cursor.setinputsizes(ret_data=ret_bind)
self.cursor.execute(sql, values1)
self.assertEqual(ret_bind.values, [['1']])
ret_bind = self.cursor.var(oracledb.DB_TYPE_VARCHAR, arraysize=1)
self.cursor.setinputsizes(ret_data=ret_bind)
self.cursor.execute(sql, values2)
self.assertEqual(ret_bind.values, [['2']])
ret_bind = self.cursor.var(oracledb.DB_TYPE_VARCHAR, arraysize=1)
self.cursor.setinputsizes(ret_data=ret_bind)
self.cursor.execute(sql, values3)
self.assertEqual(ret_bind.values, [['3']])
def test_3923_rowfactory_callable(self):
"3923 - test using rowfactory"
self.cursor.execute("truncate table TestTempTable")
self.cursor.execute("""
insert into TestTempTable (IntCol, StringCol1)
values (1, 'Test 1')""")
self.connection.commit()
self.cursor.execute("select IntCol, StringCol1 from TestTempTable")
column_names = [col[0] for col in self.cursor.description]
self.cursor.rowfactory = lambda *row: dict(zip(column_names, row))
self.assertEqual(self.cursor.fetchall(),
[{'INTCOL': 1, 'STRINGCOL1': 'Test 1'}])
def test_3924_rowfactory_execute_same_sql(self):
"3924 - test executing same query after setting rowfactory"
self.cursor.execute("truncate table TestTempTable")
data = [
(1, 'Test 1'),
(2, 'Test 2')
]
self.cursor.executemany("""
insert into TestTempTable (IntCol, StringCol1)
values (:1, :2)""", data)
self.connection.commit()
self.cursor.execute("select IntCol, StringCol1 from TestTempTable")
column_names = [col[0] for col in self.cursor.description]
self.cursor.rowfactory = lambda *row: dict(zip(column_names, row))
results_1 = self.cursor.fetchall()
self.cursor.execute("select IntCol, StringCol1 from TestTempTable")
results_2 = self.cursor.fetchall()
self.assertEqual(results_1, results_2)
def test_3925_rowfactory_execute_different_sql(self):
"3925 - test executing different query after setting rowfactory"
self.cursor.execute("truncate table TestTempTable")
data = [
(1, 'Test 1'),
(2, 'Test 2')
]
self.cursor.executemany("""
insert into TestTempTable (IntCol, StringCol1)
values (:1, :2)""", data)
self.connection.commit()
self.cursor.execute("select IntCol, StringCol1 from TestTempTable")
column_names = [col[0] for col in self.cursor.description]
self.cursor.rowfactory = lambda *row: dict(zip(column_names, row))
self.cursor.execute("""
select IntCol, StringCol from TestSTrings
where IntCol between 1 and 3 order by IntCol""")
expected_data = [
(1, 'String 1'),
(2, 'String 2'),
(3, 'String 3')
]
self.assertEqual(self.cursor.fetchall(), expected_data)
def test_3926_rowfactory_on_refcursor(self):
"3926 - test setting rowfactory on a REF cursor"
with self.connection.cursor() as cursor:
ref_cursor = cursor.callfunc("pkg_TestRefCursors.TestReturnCursor",
oracledb.DB_TYPE_CURSOR, [2])
column_names = [col[0] for col in ref_cursor.description]
ref_cursor.rowfactory = lambda *row: dict(zip(column_names, row))
self.assertEqual(ref_cursor.fetchall(),
[{'INTCOL': 1, 'STRINGCOL': 'String 1'},
{'INTCOL': 2, 'STRINGCOL': 'String 2'}])
def test_3927_subclassed_string(self):
"3927 - test using a subclassed string as bind parameter keys"
class my_str(str):
pass
self.cursor.execute("truncate table TestTempTable")
keys = {my_str("str_val"): oracledb.DB_TYPE_VARCHAR}
self.cursor.setinputsizes(**keys)
values = {
my_str("int_val"): 3927,
my_str("str_val"): "3927 - String Value"
}
self.cursor.execute("""
insert into TestTempTable (IntCol, StringCol1)
values (:int_val, :str_val)""", values)
self.cursor.execute("select IntCol, StringCol1 from TestTempTable")
self.assertEqual(self.cursor.fetchall(),
[(3927, "3927 - String Value")])
if __name__ == "__main__":
test_env.run_test_cases()
|
