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
|
#------------------------------------------------------------------------------
# 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.
#------------------------------------------------------------------------------
"""
4000 - Module for testing the cursor executemany() method
"""
import decimal
import oracledb
import test_env
class TestCase(test_env.BaseTestCase):
def test_4000_executemany_by_name(self):
"4000 - test executing a statement multiple times (named args)"
self.cursor.execute("truncate table TestTempTable")
rows = [{"value": n} for n in range(250)]
self.cursor.arraysize = 100
statement = "insert into TestTempTable (IntCol) values (:value)"
self.cursor.executemany(statement, rows)
self.connection.commit()
self.cursor.execute("select count(*) from TestTempTable")
count, = self.cursor.fetchone()
self.assertEqual(count, len(rows))
def test_4001_executemany_by_position(self):
"4001 - test executing a statement multiple times (positional args)"
self.cursor.execute("truncate table TestTempTable")
rows = [[n] for n in range(230)]
self.cursor.arraysize = 100
statement = "insert into TestTempTable (IntCol) values (:1)"
self.cursor.executemany(statement, rows)
self.connection.commit()
self.cursor.execute("select count(*) from TestTempTable")
count, = self.cursor.fetchone()
self.assertEqual(count, len(rows))
def test_4002_executemany_with_prepare(self):
"4002 - test executing a statement multiple times (with prepare)"
self.cursor.execute("truncate table TestTempTable")
rows = [[n] for n in range(225)]
self.cursor.arraysize = 100
statement = "insert into TestTempTable (IntCol) values (:1)"
self.cursor.prepare(statement)
self.cursor.executemany(None, rows)
self.connection.commit()
self.cursor.execute("select count(*) from TestTempTable")
count, = self.cursor.fetchone()
self.assertEqual(count, len(rows))
def test_4003_executemany_with_rebind(self):
"4003 - test executing a statement multiple times (with rebind)"
self.cursor.execute("truncate table TestTempTable")
rows = [[n] for n in range(235)]
self.cursor.arraysize = 100
statement = "insert into TestTempTable (IntCol) values (:1)"
self.cursor.executemany(statement, rows[:50])
self.cursor.executemany(statement, rows[50:])
self.connection.commit()
self.cursor.execute("select count(*) from TestTempTable")
count, = self.cursor.fetchone()
self.assertEqual(count, len(rows))
def test_4004_executemany_with_input_sizes_wrong(self):
"4004 - test executing multiple times (with input sizes wrong)"
cursor = self.connection.cursor()
cursor.setinputsizes(oracledb.NUMBER)
data = [[decimal.Decimal("25.8")], [decimal.Decimal("30.0")]]
cursor.executemany("declare t number; begin t := :1; end;", data)
def test_4005_executemany_with_multiple_batches(self):
"4005 - test executing multiple times (with multiple batches)"
self.cursor.execute("truncate table TestTempTable")
sql = "insert into TestTempTable (IntCol, StringCol1) values (:1, :2)"
self.cursor.executemany(sql, [(1, None), (2, None)])
self.cursor.executemany(sql, [(3, None), (4, "Testing")])
def test_4006_executemany_numeric(self):
"4006 - test executemany() with various numeric types"
self.cursor.execute("truncate table TestTempTable")
data = [
(1, 5),
(2, 7.0),
(3, 6.5),
(4, 2 ** 65),
(5, decimal.Decimal("24.5"))
]
sql = "insert into TestTempTable (IntCol, NumberCol) values (:1, :2)"
self.cursor.executemany(sql, data)
self.cursor.execute("""
select IntCol, NumberCol
from TestTempTable
order by IntCol""")
self.assertEqual(self.cursor.fetchall(), data)
def test_4007_executemany_with_resize(self):
"4007 - test executing a statement multiple times (with resize)"
self.cursor.execute("truncate table TestTempTable")
rows = [
(1, "First"),
(2, "Second"),
(3, "Third"),
(4, "Fourth"),
(5, "Fifth"),
(6, "Sixth"),
(7, "Seventh and the longest one")
]
sql = "insert into TestTempTable (IntCol, StringCol1) values (:1, :2)"
self.cursor.executemany(sql, rows)
self.cursor.execute("""
select IntCol, StringCol1
from TestTempTable
order by IntCol""")
fetched_rows = self.cursor.fetchall()
self.assertEqual(fetched_rows, rows)
def test_4008_executemany_with_exception(self):
"4008 - test executing a statement multiple times (with exception)"
self.cursor.execute("truncate table TestTempTable")
rows = [{"value": n} for n in (1, 2, 3, 2, 5)]
statement = "insert into TestTempTable (IntCol) values (:value)"
self.assertRaisesRegex(oracledb.DatabaseError, "^ORA-00001:",
self.cursor.executemany, statement, rows)
self.assertEqual(self.cursor.rowcount, 3)
def test_4009_executemany_with_invalid_parameters(self):
"4009 - test calling executemany() with invalid parameters"
sql = "insert into TestTempTable (IntCol, StringCol1) values (:1, :2)"
self.assertRaisesRegex(oracledb.ProgrammingError, "^DPY-2004:",
self.cursor.executemany, sql,
"These are not valid parameters")
def test_4010_executemany_no_parameters(self):
"4010 - test calling executemany() without any bind parameters"
num_rows = 5
self.cursor.execute("truncate table TestTempTable")
self.cursor.executemany("""
declare
t_Id number;
begin
select nvl(count(*), 0) + 1 into t_Id
from TestTempTable;
insert into TestTempTable (IntCol, StringCol1)
values (t_Id, 'Test String ' || t_Id);
end;""", num_rows)
self.assertEqual(self.cursor.rowcount, 0)
self.cursor.execute("select count(*) from TestTempTable")
count, = self.cursor.fetchone()
self.assertEqual(count, num_rows)
def test_4011_executemany_bound_earlier(self):
"4011 - test calling executemany() with binds performed earlier"
num_rows = 9
self.cursor.execute("truncate table TestTempTable")
var = self.cursor.var(int, arraysize=num_rows)
self.cursor.setinputsizes(var)
self.cursor.executemany("""
declare
t_Id number;
begin
select nvl(count(*), 0) + 1 into t_Id
from TestTempTable;
insert into TestTempTable (IntCol, StringCol1)
values (t_Id, 'Test String ' || t_Id);
select sum(IntCol) into :1
from TestTempTable;
end;""", num_rows)
self.assertEqual(self.cursor.rowcount, 0)
expected_data = [1, 3, 6, 10, 15, 21, 28, 36, 45]
self.assertEqual(var.values, expected_data)
def test_4012_executemany_with_plsql_binds(self):
"4012 - test executing plsql statements multiple times (with binds)"
var = self.cursor.var(int, arraysize=5)
self.cursor.setinputsizes(var)
data = [[25], [30], [None], [35], [None]]
exepected_data = [25, 30, None, 35, None]
self.cursor.executemany("declare t number; begin t := :1; end;", data)
self.assertEqual(var.values, exepected_data)
def test_4013_executemany_with_incorrect_params(self):
"4013 - test executemany with incorrect parameters"
self.assertRaisesRegex(oracledb.ProgrammingError, "^DPY-2004:",
self.cursor.executemany, "select :1 from dual",
[1])
def test_4014_executemany_with_mixed_binds_pos_first(self):
"4014 - test executemany with mixed binds (pos first)"
rows = [["test"], {"value": 1}]
self.assertRaisesRegex(oracledb.ProgrammingError, "^DPY-2006:",
self.cursor.executemany, "select :1 from dual",
rows)
def test_4015_executemany_with_mixed_binds_name_first(self):
"4015 - test executemany with mixed binds (name first)"
rows = [{"value": 1}, ["test"]]
self.assertRaisesRegex(oracledb.ProgrammingError, "^DPY-2006:",
self.cursor.executemany,
"select :value from dual", rows)
def test_4016_executemany_plsql_dml_returning(self):
"4016 - test executemany() with a pl/sql statement with dml returning"
num_rows = 5
self.cursor.execute("truncate table TestTempTable")
out_var = self.cursor.var(oracledb.NUMBER, arraysize=5)
self.cursor.setinputsizes(out_var)
self.cursor.executemany("""
declare
t_Id number;
begin
select nvl(count(*), 0) + 1 into t_Id
from TestTempTable;
insert into TestTempTable (IntCol, StringCol1)
values (t_Id, 'Test String ' || t_Id) returning
IntCol into :out_bind;
end;""", num_rows)
self.assertEqual(out_var.values, [1, 2, 3, 4, 5])
def test_4017_executemany_pl_sql_with_in_and_out_binds(self):
"4017 - test executemany() with pl/sql in binds and out binds"
self.cursor.execute("truncate table TestTempTable")
int_values = [5, 8, 17, 24, 6]
str_values = ["Test 5", "Test 8", "Test 17", "Test 24", "Test 6"]
out_bind = self.cursor.var(oracledb.NUMBER, arraysize=5)
self.cursor.setinputsizes(None, None, out_bind)
data = list(zip(int_values, str_values))
self.cursor.executemany("""
begin
insert into TestTempTable (IntCol, StringCol1)
values (:int_val, :str_val)
returning IntCol into :out_bind;
end;""", data)
self.assertEqual(out_bind.values, [5, 8, 17, 24, 6])
def test_4018_executemany_pl_sql_out_bind(self):
"4018 - test executemany() with pl/sql outbinds"
self.cursor.execute("truncate table TestTempTable")
out_bind = self.cursor.var(oracledb.NUMBER, arraysize=5)
self.cursor.setinputsizes(out_bind)
self.cursor.executemany("""
begin
:out_var := 5;
end;""", 5)
self.assertEqual(out_bind.values, [5, 5, 5, 5, 5])
def test_4019_re_executemany_pl_sql_with_in_and_out_binds(self):
"4019 - test re-executemany() with pl/sql in binds and out binds"
int_values = [5, 8, 17, 24, 6]
str_values = ["Test 5", "Test 8", "Test 17", "Test 24", "Test 6"]
data = list(zip(int_values, str_values))
for i in range(2):
self.cursor.execute("truncate table TestTempTable")
out_bind = self.cursor.var(oracledb.NUMBER, arraysize=5)
self.cursor.setinputsizes(None, None, out_bind)
self.cursor.executemany("""
begin
insert into TestTempTable (IntCol, StringCol1)
values (:int_val, :str_val)
returning IntCol into :out_bind;
end;""", data)
self.assertEqual(out_bind.values, [5, 8, 17, 24, 6])
def test_4020_executemany_with_plsql_single_row(self):
"4020 - test PL/SQL statement with single row bind"
value = 4020
var = self.cursor.var(int)
data = [[var, value]]
self.cursor.executemany("begin :1 := :2; end;", data)
self.assertEqual(var.values, [value])
def test_4021_defer_type_assignment(self):
"4021 - test deferral of type assignment"
self.cursor.execute("truncate table TestTempTable")
data = [(1, None), (2, 25)]
self.cursor.executemany("""
insert into TestTempTable
(IntCol, NumberCol)
values (:1, :2)""", data)
self.connection.commit()
self.cursor.execute("""
select IntCol, NumberCol
from TestTempTable
order by IntCol""")
fetched_data = self.cursor.fetchall()
self.assertEqual(data, fetched_data)
if __name__ == "__main__":
test_env.run_test_cases()
|
