315 lines
14 KiB
Python
315 lines
14 KiB
Python
#------------------------------------------------------------------------------
|
|
# 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()
|