594 lines
26 KiB
Python
594 lines
26 KiB
Python
#------------------------------------------------------------------------------
|
|
# 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.
|
|
#------------------------------------------------------------------------------
|
|
|
|
"""
|
|
1100 - Module for testing connections
|
|
"""
|
|
|
|
import random
|
|
import string
|
|
import threading
|
|
import time
|
|
import unittest
|
|
|
|
import cx_Oracle as oracledb
|
|
import test_env
|
|
|
|
class TestCase(test_env.BaseTestCase):
|
|
requires_connection = False
|
|
|
|
def __connect_and_drop(self):
|
|
"""Connect to the database, perform a query and drop the connection."""
|
|
connection = test_env.get_connection(threaded=True)
|
|
cursor = connection.cursor()
|
|
cursor.execute("select count(*) from TestNumbers")
|
|
count, = cursor.fetchone()
|
|
self.assertEqual(count, 10)
|
|
|
|
def __verify_fetched_data(self, connection):
|
|
expected_data = [f"String {i + 1}" for i in range(10)]
|
|
sql = "select StringCol from TestStrings order by IntCol"
|
|
for i in range(5):
|
|
with connection.cursor() as cursor:
|
|
fetched_data = [s for s, in cursor.execute(sql)]
|
|
self.assertEqual(fetched_data, expected_data)
|
|
|
|
def __verify_args(self, connection):
|
|
self.assertEqual(connection.username, test_env.get_main_user(),
|
|
"user name differs")
|
|
self.assertEqual(connection.tnsentry, test_env.get_connect_string(),
|
|
"tnsentry differs")
|
|
self.assertEqual(connection.dsn, test_env.get_connect_string(),
|
|
"dsn differs")
|
|
|
|
def __verify_attributes(self, connection, attrName, value, sql):
|
|
setattr(connection, attrName, value)
|
|
cursor = connection.cursor()
|
|
cursor.execute(sql)
|
|
result, = cursor.fetchone()
|
|
self.assertEqual(result, value, "%s value mismatch" % attrName)
|
|
|
|
def test_1100_all_args(self):
|
|
"1100 - connection to database with user, password, TNS separate"
|
|
connection = test_env.get_connection()
|
|
self.__verify_args(connection)
|
|
|
|
def test_1101_app_context(self):
|
|
"1101 - test use of application context"
|
|
namespace = "CLIENTCONTEXT"
|
|
app_context_entries = [
|
|
( namespace, "ATTR1", "VALUE1" ),
|
|
( namespace, "ATTR2", "VALUE2" ),
|
|
( namespace, "ATTR3", "VALUE3" )
|
|
]
|
|
connection = test_env.get_connection(appcontext=app_context_entries)
|
|
cursor = connection.cursor()
|
|
for namespace, name, value in app_context_entries:
|
|
cursor.execute("select sys_context(:1, :2) from dual",
|
|
(namespace, name))
|
|
actual_value, = cursor.fetchone()
|
|
self.assertEqual(actual_value, value)
|
|
|
|
def test_1102_app_context_negative(self):
|
|
"1102 - test invalid use of application context"
|
|
self.assertRaises(TypeError, oracledb.connect,
|
|
user=test_env.get_main_user(),
|
|
password=test_env.get_main_password(),
|
|
dsn=test_env.get_connect_string(),
|
|
appcontext=[('userenv', 'action')])
|
|
|
|
def test_1103_attributes(self):
|
|
"1103 - test connection end-to-end tracing attributes"
|
|
connection = test_env.get_connection()
|
|
if test_env.get_client_version() >= (12, 1) \
|
|
and not self.is_on_oracle_cloud(connection):
|
|
sql = "select dbop_name from v$sql_monitor " \
|
|
"where sid = sys_context('userenv', 'sid')" \
|
|
"and status = 'EXECUTING'"
|
|
self.__verify_attributes(connection, "dbop", "oracledb_dbop", sql)
|
|
sql = "select sys_context('userenv', 'action') from dual"
|
|
self.__verify_attributes(connection, "action", "oracledb_Action", sql)
|
|
sql = "select sys_context('userenv', 'module') from dual"
|
|
self.__verify_attributes(connection, "module", "oracledb_Module", sql)
|
|
sql = "select sys_context('userenv', 'client_info') from dual"
|
|
self.__verify_attributes(connection, "clientinfo", "oracledb_cinfo",
|
|
sql)
|
|
sql = "select sys_context('userenv', 'client_identifier') from dual"
|
|
self.__verify_attributes(connection, "client_identifier",
|
|
"oracledb_cid", sql)
|
|
|
|
def test_1104_autocommit(self):
|
|
"1104 - test use of autocommit"
|
|
connection = test_env.get_connection()
|
|
cursor = connection.cursor()
|
|
other_connection = test_env.get_connection()
|
|
other_cursor = other_connection.cursor()
|
|
cursor.execute("truncate table TestTempTable")
|
|
cursor.execute("insert into TestTempTable (IntCol) values (1)")
|
|
other_cursor.execute("select IntCol from TestTempTable")
|
|
rows = other_cursor.fetchall()
|
|
self.assertEqual(rows, [])
|
|
connection.autocommit = True
|
|
cursor.execute("insert into TestTempTable (IntCol) values (2)")
|
|
other_cursor.execute("select IntCol from TestTempTable order by IntCol")
|
|
rows = other_cursor.fetchall()
|
|
self.assertEqual(rows, [(1,), (2,)])
|
|
|
|
def test_1105_bad_connect_string(self):
|
|
"1105 - connection to database with bad connect string"
|
|
self.assertRaises(oracledb.DatabaseError, oracledb.connect,
|
|
test_env.get_main_user())
|
|
self.assertRaises(oracledb.DatabaseError, oracledb.connect,
|
|
test_env.get_main_user() + "@" + \
|
|
test_env.get_connect_string())
|
|
self.assertRaises(oracledb.DatabaseError, oracledb.connect,
|
|
test_env.get_main_user() + "@" + \
|
|
test_env.get_connect_string() + "/" + \
|
|
test_env.get_main_password())
|
|
|
|
def test_1106_bad_password(self):
|
|
"1106 - connection to database with bad password"
|
|
self.assertRaises(oracledb.DatabaseError, oracledb.connect,
|
|
user=test_env.get_main_user(),
|
|
password=test_env.get_main_password() + "X",
|
|
dsn=test_env.get_connect_string())
|
|
|
|
def test_1107_change_password(self):
|
|
"1107 - test changing password"
|
|
connection = test_env.get_connection()
|
|
if self.is_on_oracle_cloud(connection):
|
|
self.skipTest("passwords on Oracle Cloud are strictly controlled")
|
|
sys_random = random.SystemRandom()
|
|
new_password = "".join(sys_random.choice(string.ascii_letters) \
|
|
for i in range(20))
|
|
connection.changepassword(test_env.get_main_password(), new_password)
|
|
connection = oracledb.connect(dsn=test_env.get_connect_string(),
|
|
user=test_env.get_main_user(),
|
|
password=new_password)
|
|
connection.changepassword(new_password, test_env.get_main_password())
|
|
|
|
def test_1108_change_password_negative(self):
|
|
"1108 - test changing password to an invalid value"
|
|
connection = test_env.get_connection()
|
|
if self.is_on_oracle_cloud(connection):
|
|
self.skipTest("passwords on Oracle Cloud are strictly controlled")
|
|
new_password = "1" * 150
|
|
self.assertRaises(oracledb.DatabaseError, connection.changepassword,
|
|
test_env.get_main_password(), new_password)
|
|
|
|
def test_1109_parse_password(self):
|
|
"1109 - test connecting with password containing / and @ symbols"
|
|
connection = test_env.get_connection()
|
|
if self.is_on_oracle_cloud(connection):
|
|
self.skipTest("passwords on Oracle Cloud are strictly controlled")
|
|
sys_random = random.SystemRandom()
|
|
chars = list(sys_random.choice(string.ascii_letters) for i in range(20))
|
|
chars[4] = "/"
|
|
chars[8] = "@"
|
|
new_password = "".join(chars)
|
|
connection.changepassword(test_env.get_main_password(), new_password)
|
|
try:
|
|
arg = "%s/%s@%s" % (test_env.get_main_user(), new_password,
|
|
test_env.get_connect_string())
|
|
oracledb.connect(arg)
|
|
finally:
|
|
connection.changepassword(new_password,
|
|
test_env.get_main_password())
|
|
|
|
def test_1110_encodings(self):
|
|
"1110 - connection with only encoding/nencoding specified should work"
|
|
connection = oracledb.connect(test_env.get_main_user(),
|
|
test_env.get_main_password(),
|
|
test_env.get_connect_string())
|
|
encoding = connection.encoding
|
|
nencoding = connection.nencoding
|
|
alt_encoding = "ISO-8859-1"
|
|
connection = oracledb.connect(test_env.get_main_user(),
|
|
test_env.get_main_password(),
|
|
test_env.get_connect_string(),
|
|
encoding=alt_encoding)
|
|
self.assertEqual(connection.encoding, alt_encoding)
|
|
self.assertEqual(connection.nencoding, nencoding)
|
|
connection = oracledb.connect(test_env.get_main_user(),
|
|
test_env.get_main_password(),
|
|
test_env.get_connect_string(),
|
|
nencoding=alt_encoding)
|
|
self.assertEqual(connection.encoding, encoding)
|
|
self.assertEqual(connection.nencoding, alt_encoding)
|
|
|
|
def test_1111_different_encodings(self):
|
|
"1111 - different encodings can be specified for encoding/nencoding"
|
|
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()
|
|
nchar_var = cursor.var(oracledb.DB_TYPE_NVARCHAR, 100)
|
|
nchar_var.setvalue(0, value)
|
|
cursor.execute("select :value from dual", value=nchar_var)
|
|
result, = cursor.fetchone()
|
|
self.assertEqual(result, value)
|
|
|
|
def test_1112_exception_on_close(self):
|
|
"1112 - confirm an exception is raised after closing a connection"
|
|
connection = test_env.get_connection()
|
|
connection.close()
|
|
self.assertRaises(oracledb.InterfaceError, connection.rollback)
|
|
|
|
def test_1113_connect_with_handle(self):
|
|
"1113 - test creating a connection using a handle"
|
|
connection = test_env.get_connection()
|
|
cursor = connection.cursor()
|
|
cursor.execute("truncate table TestTempTable")
|
|
int_value = random.randint(1, 32768)
|
|
cursor.execute("""
|
|
insert into TestTempTable (IntCol, StringCol)
|
|
values (:val, null)""", val=int_value)
|
|
connection2 = oracledb.connect(handle=connection.handle)
|
|
cursor = connection2.cursor()
|
|
cursor.execute("select IntCol from TestTempTable")
|
|
fetched_int_value, = cursor.fetchone()
|
|
self.assertEqual(fetched_int_value, int_value)
|
|
cursor.close()
|
|
self.assertRaises(oracledb.DatabaseError, connection2.close)
|
|
connection.close()
|
|
|
|
def test_1114_make_dsn(self):
|
|
"1114 - test making a data source name from host, port and sid"
|
|
format_string = "(DESCRIPTION=(ADDRESS=(PROTOCOL=TCP)" + \
|
|
"(HOST=%s)(PORT=%d))(CONNECT_DATA=(SID=%s)))"
|
|
args = ("hostname", 1521, "TEST")
|
|
result = oracledb.makedsn(*args)
|
|
self.assertEqual(result, format_string % args)
|
|
|
|
def test_1115_single_arg(self):
|
|
"1115 - connection to database with user, password, DSN together"
|
|
arg = "%s/%s@%s" % (test_env.get_main_user(),
|
|
test_env.get_main_password(),
|
|
test_env.get_connect_string())
|
|
connection = oracledb.connect(arg)
|
|
self.__verify_args(connection)
|
|
|
|
def test_1116_version(self):
|
|
"1116 - connection version is a string"
|
|
connection = test_env.get_connection()
|
|
self.assertTrue(isinstance(connection.version, str))
|
|
|
|
def test_1117_rollback_on_close(self):
|
|
"1117 - connection rolls back before close"
|
|
connection = test_env.get_connection()
|
|
cursor = connection.cursor()
|
|
cursor.execute("truncate table TestTempTable")
|
|
other_connection = test_env.get_connection()
|
|
other_cursor = other_connection.cursor()
|
|
other_cursor.execute("insert into TestTempTable (IntCol) values (1)")
|
|
other_cursor.close()
|
|
other_connection.close()
|
|
cursor.execute("select count(*) from TestTempTable")
|
|
count, = cursor.fetchone()
|
|
self.assertEqual(count, 0)
|
|
|
|
def test_1118_rollback_on_del(self):
|
|
"1118 - connection rolls back before destruction"
|
|
connection = test_env.get_connection()
|
|
cursor = connection.cursor()
|
|
cursor.execute("truncate table TestTempTable")
|
|
other_connection = test_env.get_connection()
|
|
other_cursor = other_connection.cursor()
|
|
other_cursor.execute("insert into TestTempTable (IntCol) values (1)")
|
|
del other_cursor
|
|
del other_connection
|
|
cursor.execute("select count(*) from TestTempTable")
|
|
count, = cursor.fetchone()
|
|
self.assertEqual(count, 0)
|
|
|
|
def test_1119_threading(self):
|
|
"1119 - multiple connections to database with multiple threads"
|
|
threads = []
|
|
for i in range(20):
|
|
thread = threading.Thread(None, self.__connect_and_drop)
|
|
threads.append(thread)
|
|
thread.start()
|
|
for thread in threads:
|
|
thread.join()
|
|
|
|
def test_1120_string_format(self):
|
|
"1120 - test string format of connection"
|
|
connection = test_env.get_connection()
|
|
expected_value = "<cx_Oracle.Connection to %s@%s>" % \
|
|
(test_env.get_main_user(), test_env.get_connect_string())
|
|
self.assertEqual(str(connection), expected_value)
|
|
|
|
def test_1121_ctx_mgr_close(self):
|
|
"1121 - test context manager - close"
|
|
connection = test_env.get_connection()
|
|
with connection:
|
|
cursor = connection.cursor()
|
|
cursor.execute("truncate table TestTempTable")
|
|
cursor.execute("insert into TestTempTable (IntCol) values (1)")
|
|
connection.commit()
|
|
cursor.execute("insert into TestTempTable (IntCol) values (2)")
|
|
self.assertRaises(oracledb.InterfaceError, connection.ping)
|
|
connection = test_env.get_connection()
|
|
cursor = connection.cursor()
|
|
cursor.execute("select count(*) from TestTempTable")
|
|
count, = cursor.fetchone()
|
|
self.assertEqual(count, 1)
|
|
|
|
def test_1122_connection_attributes(self):
|
|
"1122 - test connection attribute values"
|
|
connection = oracledb.connect(test_env.get_main_user(),
|
|
test_env.get_main_password(),
|
|
test_env.get_connect_string(),
|
|
encoding="ASCII")
|
|
self.assertEqual(connection.maxBytesPerCharacter, 1)
|
|
connection = oracledb.connect(test_env.get_main_user(),
|
|
test_env.get_main_password(),
|
|
test_env.get_connect_string(),
|
|
encoding="UTF-8")
|
|
self.assertEqual(connection.maxBytesPerCharacter, 4)
|
|
if test_env.get_client_version() >= (12, 1):
|
|
self.assertEqual(connection.ltxid, b'')
|
|
self.assertEqual(connection.current_schema, None)
|
|
connection.current_schema = "test_schema"
|
|
self.assertEqual(connection.current_schema, "test_schema")
|
|
self.assertEqual(connection.edition, None)
|
|
connection.external_name = "test_external"
|
|
self.assertEqual(connection.external_name, "test_external")
|
|
connection.internal_name = "test_internal"
|
|
self.assertEqual(connection.internal_name, "test_internal")
|
|
connection.stmtcachesize = 30
|
|
self.assertEqual(connection.stmtcachesize, 30)
|
|
self.assertRaises(TypeError, connection.stmtcachesize, 20.5)
|
|
self.assertRaises(TypeError, connection.stmtcachesize, "value")
|
|
|
|
def test_1123_closed_connection_attributes(self):
|
|
"1123 - test closed connection attribute values"
|
|
connection = test_env.get_connection()
|
|
connection.close()
|
|
attr_names = ["current_schema", "edition", "external_name",
|
|
"internal_name", "stmtcachesize"]
|
|
if test_env.get_client_version() >= (12, 1):
|
|
attr_names.append("ltxid")
|
|
for name in attr_names:
|
|
self.assertRaises(oracledb.InterfaceError, getattr, connection,
|
|
name)
|
|
|
|
def test_1124_ping(self):
|
|
"1124 - test connection ping"
|
|
connection = test_env.get_connection()
|
|
connection.ping()
|
|
|
|
def test_1125_transaction_begin(self):
|
|
"1125 - test begin, prepare, cancel transaction"
|
|
connection = test_env.get_connection()
|
|
cursor = connection.cursor()
|
|
cursor.execute("truncate table TestTempTable")
|
|
connection.begin(10, 'trxnId', 'branchId')
|
|
self.assertEqual(connection.prepare(), False)
|
|
connection.begin(10, 'trxnId', 'branchId')
|
|
cursor.execute("""
|
|
insert into TestTempTable (IntCol, StringCol)
|
|
values (1, 'tesName')""")
|
|
self.assertEqual(connection.prepare(), True)
|
|
connection.cancel()
|
|
connection.rollback()
|
|
cursor.execute("select count(*) from TestTempTable")
|
|
count, = cursor.fetchone()
|
|
self.assertEqual(count, 0)
|
|
|
|
def test_1126_multiple_transactions(self):
|
|
"1126 - test multiple transactions on the same connection"
|
|
connection = test_env.get_connection()
|
|
with connection.cursor() as cursor:
|
|
cursor.execute("truncate table TestTempTable")
|
|
|
|
id_ = random.randint(0, 2 ** 128)
|
|
xid = (0x1234, "%032x" % id_, "%032x" % 9)
|
|
connection.begin(*xid)
|
|
with connection.cursor() as cursor:
|
|
cursor.execute("""
|
|
insert into TestTempTable (IntCol, StringCol)
|
|
values (1, 'tesName')""")
|
|
self.assertEqual(connection.prepare(), True)
|
|
connection.commit()
|
|
|
|
for begin_trans in (True, False):
|
|
val = 3
|
|
if begin_trans:
|
|
connection.begin()
|
|
val = 2
|
|
with connection.cursor() as cursor:
|
|
cursor.execute("""
|
|
insert into TestTempTable (IntCol, StringCol)
|
|
values (:int_val, 'tesName')""", int_val=val)
|
|
connection.commit()
|
|
|
|
expected_rows = [(1, "tesName"), (2, "tesName"), (3, "tesName")]
|
|
with connection.cursor() as cursor:
|
|
cursor.execute("select IntCol, StringCol from TestTempTable")
|
|
self.assertEqual(cursor.fetchall(), expected_rows)
|
|
|
|
def test_1127_multiple_global_transactions(self):
|
|
"1127 - test multiple global transactions on the same connection"
|
|
connection = test_env.get_connection()
|
|
with connection.cursor() as cursor:
|
|
cursor.execute("truncate table TestTempTable")
|
|
|
|
id_ = random.randint(0, 2 ** 128)
|
|
xid = (0x1234, "%032x" % id_, "%032x" % 9)
|
|
connection.begin(*xid)
|
|
with connection.cursor() as cursor:
|
|
cursor.execute("""
|
|
insert into TestTempTable (IntCol, StringCol)
|
|
values (1, 'tesName')""")
|
|
self.assertEqual(connection.prepare(), True)
|
|
connection.commit()
|
|
|
|
for begin_trans in (True, False):
|
|
val = 3
|
|
if begin_trans:
|
|
connection.begin()
|
|
val = 2
|
|
with connection.cursor() as cursor:
|
|
cursor.execute("""
|
|
insert into TestTempTable (IntCol, StringCol)
|
|
values (:int_val, 'tesName')""", int_val=val)
|
|
connection.commit()
|
|
|
|
id2_ = random.randint(0, 2 ** 128)
|
|
xid2 = (0x1234, "%032x" % id2_, "%032x" % 9)
|
|
connection.begin(*xid2)
|
|
with connection.cursor() as cursor:
|
|
cursor.execute("""
|
|
insert into TestTempTable (IntCol, StringCol)
|
|
values (4, 'tesName')""")
|
|
self.assertEqual(connection.prepare(), True)
|
|
connection.commit()
|
|
|
|
expected_rows = [(1, "tesName"), (2, "tesName"), (3, "tesName"),
|
|
(4, "tesName")]
|
|
|
|
with connection.cursor() as cursor:
|
|
cursor.execute("select IntCol, StringCol from TestTempTable")
|
|
self.assertEqual(cursor.fetchall(), expected_rows)
|
|
|
|
def test_1128_exception_creating_global_txn_after_local_txn(self):
|
|
"1128 - test creating global txn after a local txn"
|
|
connection = test_env.get_connection()
|
|
with connection.cursor() as cursor:
|
|
cursor.execute("truncate table TestTempTable")
|
|
|
|
val = 2
|
|
with connection.cursor() as cursor:
|
|
cursor.execute("""
|
|
insert into TestTempTable (IntCol, StringCol)
|
|
values (:int_val, 'tesName')""", int_val=val)
|
|
|
|
id_ = random.randint(0, 2 ** 128)
|
|
xid = (0x1234, "%032x" % id_, "%032x" % 9)
|
|
self.assertRaises(oracledb.DatabaseError, connection.begin, *xid)
|
|
|
|
def test_1129_threading_single_connection(self):
|
|
"1129 - single connection to database with multiple threads"
|
|
with test_env.get_connection(threaded=True) as connection:
|
|
threads = [threading.Thread(target=self.__verify_fetched_data,
|
|
args=(connection,)) for i in range(3)]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join()
|
|
|
|
def test_1130_cancel(self):
|
|
"1130 - test connection cancel"
|
|
conn = test_env.get_connection()
|
|
def perform_cancel():
|
|
time.sleep(0.1)
|
|
conn.cancel()
|
|
thread = threading.Thread(target=perform_cancel)
|
|
thread.start()
|
|
try:
|
|
with conn.cursor() as cursor:
|
|
self.assertRaises(oracledb.OperationalError, cursor.callproc,
|
|
test_env.get_sleep_proc_name(), [2])
|
|
finally:
|
|
thread.join()
|
|
with conn.cursor() as cursor:
|
|
cursor.execute("select user from dual")
|
|
user, = cursor.fetchone()
|
|
self.assertEqual(user, test_env.get_main_user().upper())
|
|
|
|
def test_1131_change_password_during_connect(self):
|
|
"1131 - test changing password during connect"
|
|
connection = test_env.get_connection()
|
|
if self.is_on_oracle_cloud(connection):
|
|
self.skipTest("passwords on Oracle Cloud are strictly controlled")
|
|
sys_random = random.SystemRandom()
|
|
new_password = "".join(sys_random.choice(string.ascii_letters) \
|
|
for i in range(20))
|
|
connection = oracledb.connect(dsn=test_env.get_connect_string(),
|
|
user=test_env.get_main_user(),
|
|
password=test_env.get_main_password(),
|
|
newpassword=new_password)
|
|
connection = oracledb.connect(dsn=test_env.get_connect_string(),
|
|
user=test_env.get_main_user(),
|
|
password=new_password)
|
|
connection.changepassword(new_password, test_env.get_main_password())
|
|
|
|
def test_1132_autocommit_during_reexecute(self):
|
|
"1132 - test use of autocommit during reexecute"
|
|
sql = "insert into TestTempTable (IntCol, StringCol) values (:1, :2)"
|
|
data_to_insert = [
|
|
(1, "Test String #1"),
|
|
(2, "Test String #2")
|
|
]
|
|
connection = test_env.get_connection()
|
|
cursor = connection.cursor()
|
|
other_connection = test_env.get_connection()
|
|
other_cursor = other_connection.cursor()
|
|
cursor.execute("truncate table TestTempTable")
|
|
cursor.execute(sql, data_to_insert[0])
|
|
other_cursor.execute("select IntCol, StringCol from TestTempTable")
|
|
rows = other_cursor.fetchall()
|
|
self.assertEqual(rows, [])
|
|
connection.autocommit = True
|
|
cursor.execute(sql, data_to_insert[1])
|
|
other_cursor.execute("select IntCol, StringCol from TestTempTable")
|
|
rows = other_cursor.fetchall()
|
|
self.assertEqual(rows, data_to_insert)
|
|
|
|
def test_1133_current_schema(self):
|
|
"1133 - test current_schame is set properly"
|
|
conn = test_env.get_connection()
|
|
self.assertEqual(conn.current_schema, None)
|
|
|
|
user = test_env.get_main_user().upper()
|
|
proxy_user = test_env.get_proxy_user().upper()
|
|
cursor = conn.cursor()
|
|
cursor.execute(f'alter session set current_schema={proxy_user}')
|
|
self.assertEqual(conn.current_schema, proxy_user)
|
|
|
|
conn.current_schema = user
|
|
self.assertEqual(conn.current_schema, user)
|
|
|
|
cursor.execute("""
|
|
select sys_context('userenv', 'current_schema')
|
|
from dual""")
|
|
result, = cursor.fetchone()
|
|
self.assertEqual(result, user)
|
|
|
|
def test_1134_dbms_output(self):
|
|
"1134 - test dbms_output package"
|
|
conn = test_env.get_connection()
|
|
cursor = conn.cursor()
|
|
test_string = "Testing DBMS_OUTPUT package"
|
|
cursor.callproc("dbms_output.enable")
|
|
cursor.execute("""
|
|
begin
|
|
dbms_output.put_line(:val);
|
|
end; """, val=test_string)
|
|
string_var = cursor.var(str)
|
|
number_var = cursor.var(int)
|
|
cursor.callproc("dbms_output.get_line", (string_var, number_var))
|
|
self.assertEqual(string_var.getvalue(), test_string)
|
|
|
|
@unittest.skipIf(test_env.get_client_version() < (18, 1),
|
|
"unsupported client")
|
|
def test_1135_calltimeout(self):
|
|
"1135 - test connection call_timeout"
|
|
conn = test_env.get_connection()
|
|
conn.call_timeout = 500 # milliseconds
|
|
self.assertEqual(conn.call_timeout, 500)
|
|
self.assertRaises(oracledb.DatabaseError, conn.cursor().callproc,
|
|
test_env.get_sleep_proc_name(), [2])
|
|
|
|
if __name__ == "__main__":
|
|
test_env.run_test_cases()
|