test_env.py

# -----------------------------------------------------------------------------
# Copyright (c) 2020, 2025, 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.
# -----------------------------------------------------------------------------

# -----------------------------------------------------------------------------
# Sets the environment used by the python-oracledb test suite. Production
# applications should consider using External Authentication to avoid hard
# coded credentials.
#
# You can set values in environment variables to bypass having the test suite
# request the information it requires.
#
# PYO_TEST_MAIN_USER: user used for most test cases
# PYO_TEST_MAIN_PASSWORD: password of user used for most test cases
# PYO_TEST_PROXY_USER: user for testing proxying
# PYO_TEST_PROXY_PASSWORD: password of user for testing proxying
# PYO_TEST_CONNECT_STRING: connect string for test suite
# PYO_TEST_ADMIN_USER: administrative user for test suite
# PYO_TEST_ADMIN_PASSWORD: administrative password for test suite
# PYO_TEST_WALLET_LOCATION: location of wallet file (thin mode, mTLS)
# PYO_TEST_WALLET_PASSWORD: password for wallet file (thin mode, mTLS)
# PYO_TEST_DRIVER_MODE: python-oracledb mode (thick or thin) to use
# PYO_TEST_EXTERNAL_USER: user for testing external authentication
# PYO_TEST_EDITION_NAME: name of edition for editioning tests
# PYO_TEST_PLUGINS: list of plugins to import before running tests
#
# PYO_TEST_CONNECT_STRING can be set to an Easy Connect string, or a
# Net Service Name from a tnsnames.ora file or external naming service,
# or it can be the name of a local Oracle database instance.
#
# If oracledb is using Instant Client, then an Easy Connect string is generally
# appropriate. The syntax is:
#
# [//]host_name[:port][/service_name][:server_type][/instance_name]
#
# Commonly just the host_name and service_name are needed
# e.g. "localhost/orclpdb1" or "localhost/XEPDB1"
#
# If using a tnsnames.ora file, the file can be in a default
# location such as $ORACLE_HOME/network/admin/tnsnames.ora or
# /etc/tnsnames.ora. Alternatively set the TNS_ADMIN environment
# variable and put the file in $TNS_ADMIN/tnsnames.ora.
#
# The administrative user for cloud databases is ADMIN and the administrative
# user for on premises databases is SYSTEM.
# -----------------------------------------------------------------------------

importgetpass
importimportlib
importos
importsecrets
importsys
importstring
importunittest

importoracledb

# default values
DEFAULT_MAIN_USER="pythontest"
DEFAULT_PROXY_USER="pythontestproxy"
DEFAULT_CONNECT_STRING="localhost/orclpdb1"
DEFAULT_EDITION_NAME="pythonedition"

# dictionary containing all parameters; these are acquired as needed by the
# methods below (which should be used instead of consulting this dictionary
# directly) and then stored so that a value is not requested more than once
PARAMETERS= {}


def_initialize():
"""
 Performs initialization of the test environment. This ensures the desired
 mode is set and imports any required plugins.
 """
ifPARAMETERS.get("INITIALIZED"):
return
ifnotget_is_thin() andoracledb.is_thin_mode():
oracledb.init_oracle_client()
oracledb.defaults.thick_mode_dsn_passthrough=False
plugin_names=os.environ.get("PYO_TEST_PLUGINS")
ifplugin_namesisnotNone:
fornameinplugin_names.split(","):
module_name=f"oracledb.plugins.{name}"
print("importing module", module_name)
importlib.import_module(module_name)
PARAMETERS["INITIALIZED"] =True


defget_value(name, label, default_value=None, password=False):
try:
returnPARAMETERS[name]
exceptKeyError:
pass
env_name="PYO_TEST_"+name
value=os.environ.get(env_name)
ifvalueisNone:
ifdefault_valueisnotNone:
label+=" [%s]"%default_value
label+=": "
ifpassword:
value=getpass.getpass(label)
else:
value=input(label).strip()
ifnotvalue:
value=default_value
PARAMETERS[name] =value
returnvalue


defget_admin_connection(use_async=False):
_initialize()
admin_user=get_value("ADMIN_USER", "Administrative user", "admin")
admin_password=get_value(
"ADMIN_PASSWORD", f"Password for {admin_user}", password=True
 )
params=get_connect_params()
ifadmin_userandadmin_user.upper() =="SYS":
params=params.copy()
params.set(mode=oracledb.AUTH_MODE_SYSDBA)
method=oracledb.connect_asyncifuse_asyncelseoracledb.connect
returnmethod(
dsn=get_connect_string(),
params=params,
user=admin_user,
password=admin_password,
 )


asyncdefget_admin_connection_async(use_async=False):
returnawaitget_admin_connection(use_async=True)


defget_charset():
value=PARAMETERS.get("CHARSET")
ifvalueisNone:
withget_connection() asconn:
withconn.cursor() ascursor:
cursor.execute(
"""
 select value
 from nls_database_parameters
 where parameter = 'NLS_CHARACTERSET'
 """
 )
 (value,) =cursor.fetchone()
PARAMETERS["CHARSET"] =value
returnvalue


asyncdefget_charset_async():
value=PARAMETERS.get("CHARSET")
ifvalueisNone:
asyncwithget_connection_async() asconn:
withconn.cursor() ascursor:
awaitcursor.execute(
"""
 select value
 from nls_database_parameters
 where parameter = 'NLS_CHARACTERSET'
 """
 )
 (value,) =awaitcursor.fetchone()
PARAMETERS["CHARSET"] =value
returnvalue


defget_charset_ratios():
value=PARAMETERS.get("CS_RATIO")
ifvalueisNone:
connection=get_connection()
cursor=connection.cursor()
cursor.execute(
"""
 select
 cast('X' as varchar2(1)),
 cast('Y' as nvarchar2(1))
 from dual
 """
 )
varchar_column_info, nvarchar_column_info=cursor.description
value= (varchar_column_info[3], nvarchar_column_info[3])
PARAMETERS["CS_RATIO"] =value
returnvalue


asyncdefget_charset_ratios_async():
value=PARAMETERS.get("CS_RATIO")
ifvalueisNone:
connection=awaitget_connection_async()
cursor=connection.cursor()
awaitcursor.execute(
"""
 select
 cast('X' as varchar2(1)),
 cast('Y' as nvarchar2(1))
 from dual
 """
 )
varchar_column_info, nvarchar_column_info=cursor.description
value= (varchar_column_info[3], nvarchar_column_info[3])
PARAMETERS["CS_RATIO"] =value
returnvalue


defget_client_version():
name="CLIENT_VERSION"
value=PARAMETERS.get(name)
ifvalueisNone:
ifget_is_thin():
value= (23, 7)
else:
_initialize()
value=oracledb.clientversion()[:2]
PARAMETERS[name] =value
returnvalue


defget_connect_params():
wallet_location=get_wallet_location()
returnoracledb.ConnectParams(
user=get_main_user(),
password=get_main_password(),
config_dir=wallet_location,
wallet_location=wallet_location,
wallet_password=get_wallet_password(),
disable_oob=True,
 )


defget_connection(dsn=None, use_async=False, **kwargs):
_initialize()
ifdsnisNone:
dsn=get_connect_string()
method=oracledb.connect_asyncifuse_asyncelseoracledb.connect
returnmethod(dsn=dsn, params=get_connect_params(), **kwargs)


defget_connection_async(dsn=None, **kwargs):
returnget_connection(dsn, use_async=True, **kwargs)


defget_connect_string():
returnget_value(
"CONNECT_STRING", "Connect String", DEFAULT_CONNECT_STRING
 )


defget_edition_name():
returnget_value("EDITION_NAME", "Edition Name", DEFAULT_EDITION_NAME)


defget_is_drcp():
value=PARAMETERS.get("IS_DRCP")
ifvalueisNone:
params=oracledb.ConnectParams()
params.parse_connect_string(get_connect_string())
server_type=params.server_type
value= (
server_type=="pooled"
orisinstance(server_type, list)
and"pooled"inserver_type
 )
PARAMETERS["IS_DRCP"] =value
returnvalue


defget_is_implicit_pooling():
value=PARAMETERS.get("IS_IMPLICIT_POOLING")
ifvalueisNone:
ifnotget_is_drcp():
value=False
else:
params=oracledb.ConnectParams()
params.parse_connect_string(get_connect_string())
pool_boundary=params.pool_boundary
value= (
pool_boundaryisnotNone
orisinstance(pool_boundary, list)
and [sforsinpool_boundaryifs]
 )
PARAMETERS["IS_IMPLICIT_POOLING"] =value
returnvalue


defget_is_thin():
driver_mode=get_value("DRIVER_MODE", "Driver mode (thin|thick)", "thin")
returndriver_mode=="thin"


defget_main_password():
returnget_value(
"MAIN_PASSWORD", f"Password for {get_main_user()}", password=True
 )


defget_main_user():
returnget_value("MAIN_USER", "Main User Name", DEFAULT_MAIN_USER)


defget_pool(use_async=False, **kwargs):
_initialize()
method=oracledb.create_pool_asyncifuse_asyncelseoracledb.create_pool
returnmethod(dsn=get_connect_string(), params=get_pool_params(), **kwargs)


defget_pool_async(**kwargs):
returnget_pool(use_async=True, **kwargs)


defget_pool_params():
wallet_location=get_wallet_location()
returnoracledb.PoolParams(
user=get_main_user(),
password=get_main_password(),
config_dir=wallet_location,
wallet_location=wallet_location,
wallet_password=get_wallet_password(),
 )


defget_proxy_password():
returnget_value(
"PROXY_PASSWORD", f"Password for {get_proxy_user()}", password=True
 )


defget_proxy_user():
returnget_value("PROXY_USER", "Proxy User Name", DEFAULT_PROXY_USER)


defget_sleep_proc_name():
server_version=get_server_version()
return (
"dbms_session.sleep"ifserver_version[0] >=18else"dbms_lock.sleep"
 )


defget_server_version():
name="SERVER_VERSION"
value=PARAMETERS.get(name)
ifvalueisNone:
conn=get_connection()
value=tuple(int(s) forsinconn.version.split("."))[:2]
PARAMETERS[name] =value
returnvalue


asyncdefget_server_version_async():
name="SERVER_VERSION"
value=PARAMETERS.get(name)
ifvalueisNone:
asyncwithawaitget_connection_async() asconn:
value=tuple(int(s) forsinconn.version.split("."))[:2]
PARAMETERS[name] =value
returnvalue


defget_wallet_location():
ifget_is_thin():
returnget_value("WALLET_LOCATION", "Wallet Location")


defget_wallet_password():
ifget_is_thin():
returnget_value("WALLET_PASSWORD", "Wallet Password", password=True)


defget_external_user():
ifnotget_is_thin():
returnget_value("EXTERNAL_USER", "External User")


defget_random_string(length=10):
return"".join(secrets.choice(string.ascii_letters) foriinrange(length))


defis_on_oracle_cloud(connection):
server=get_server_version()
ifserver< (18, 0):
returnFalse
cursor=connection.cursor()
cursor.execute(
"""
 select sys_context('userenv', 'cloud_service')
 from dual
 """
 )
 (service_name,) =cursor.fetchone()
returnservice_nameisnotNone


asyncdefis_on_oracle_cloud_async(connection):
server=awaitget_server_version_async()
ifserver< (18, 0):
returnFalse
cursor=connection.cursor()
awaitcursor.execute(
"""
 select sys_context('userenv', 'cloud_service')
 from dual
 """
 )
 (service_name,) =awaitcursor.fetchone()
returnservice_nameisnotNone


defrun_sql_script(conn, script_name, **kwargs):
statement_parts= []
cursor=conn.cursor()
replace_values= [("&"+k+".", v) fork, vinkwargs.items()] + [
 ("&"+k, v) fork, vinkwargs.items()
 ]
script_dir=os.path.dirname(os.path.abspath(sys.argv[0]))
file_name=os.path.join(script_dir, "sql", script_name+".sql")
forlineinopen(file_name):
ifline.strip() =="/":
statement="".join(statement_parts).strip()
ifstatement:
forsearch_value, replace_valueinreplace_values:
statement=statement.replace(search_value, replace_value)
try:
cursor.execute(statement)
except:
print("Failed to execute SQL:", statement)
raise
statement_parts= []
else:
statement_parts.append(line)
cursor.execute(
"""
 select name, type, line, position, text
 from dba_errors
 where owner = upper(:owner)
 order by name, type, line, position
 """,
owner=get_main_user(),
 )
prev_name=prev_obj_type=None
forname, obj_type, line_num, position, textincursor:
ifname!=prev_nameorobj_type!=prev_obj_type:
print("%s (%s)"% (name, obj_type))
prev_name=name
prev_obj_type=obj_type
print(" %s/%s %s"% (line_num, position, text))


defrun_test_cases():
get_is_thin()
unittest.main(testRunner=unittest.TextTestRunner(verbosity=2))


defskip_soda_tests():
ifget_is_thin():
returnTrue
client=get_client_version()
ifclient< (18, 3):
returnTrue
server=get_server_version()
ifserver< (18, 0):
returnTrue
ifserver> (20, 1) andclient< (20, 1):
returnTrue
returnFalse


classDefaultsContextManager:
def__init__(self, attribute, desired_value):
self.attribute=attribute
self.desired_value=desired_value

def__enter__(self):
self.original_value=getattr(oracledb.defaults, self.attribute)
setattr(oracledb.defaults, self.attribute, self.desired_value)
returnself

def__exit__(self, exc_type, exc_val, exc_tb):
setattr(oracledb.defaults, self.attribute, self.original_value)


classSystemStatInfo:
get_sid_sql="select sys_context('userenv', 'sid') from dual"
get_stat_sql="""
 select ss.value
 from v$sesstat ss, v$statname sn
 where ss.sid = :sid
 and ss.statistic# = sn.statistic#
 and sn.name = :stat_name
 """
stat_name=None

def_initialize(self, connection):
self.prev_value=0
self.admin_conn=get_admin_connection()
withconnection.cursor() ascursor:
cursor.execute(self.get_sid_sql)
 (self.sid,) =cursor.fetchone()
self.get_value()

asyncdef_initialize_async(self, connection):
self.prev_value=0
self.admin_conn=awaitget_admin_connection_async()
withconnection.cursor() ascursor:
awaitcursor.execute(self.get_sid_sql)
 (self.sid,) =awaitcursor.fetchone()
awaitself.get_value_async()

defget_value(self):
withself.admin_conn.cursor() ascursor:
cursor.execute(
self.get_stat_sql, sid=self.sid, stat_name=self.stat_name
 )
 (current_value,) =cursor.fetchone()
diff_value=current_value-self.prev_value
self.prev_value=current_value
returndiff_value

asyncdefget_value_async(self):
withself.admin_conn.cursor() ascursor:
awaitcursor.execute(
self.get_stat_sql, sid=self.sid, stat_name=self.stat_name
 )
 (current_value,) =awaitcursor.fetchone()
diff_value=current_value-self.prev_value
self.prev_value=current_value
returndiff_value


classRoundTripInfo(SystemStatInfo):
stat_name="SQL*Net roundtrips to/from client"


classParseCountInfo(SystemStatInfo):
stat_name="parse count (total)"


classFullCodeErrorContextManager:

def__init__(self, full_codes):
self.full_codes=full_codes
iflen(full_codes) ==1:
self.message_fragment=f'Error "{full_codes[0]}"'
else:
message_fragment=", ".join(f'"{s}"'forsinfull_codes[:-1])
message_fragment+=f' or "{full_codes[-1]}"'
self.message_fragment=f"One of the errors {message_fragment}"

def__enter__(self):
returnself

def__exit__(self, exc_type, exc_value, tb):
ifexc_typeisNone:
raiseAssertionError(f"{self.message_fragment} was not raised.")
ifnotissubclass(exc_type, oracledb.Error):
returnFalse
ifissubclass(exc_type, oracledb.Error):
self.error_obj=exc_value.args[0]
ifself.error_obj.full_codenotinself.full_codes:
message= (
f"{self.message_fragment} should have been raised but "
f'"{self.error_obj.full_code}" was raised instead.'
 )
raiseAssertionError(message)
returnTrue


classBaseTestCase(unittest.TestCase):
requires_connection=True

defassertParseCount(self, n):
self.assertEqual(self.parse_count_info.get_value(), n)

defassertRaisesFullCode(self, *full_codes):
returnFullCodeErrorContextManager(full_codes)

defassertRoundTrips(self, n):
self.assertEqual(self.round_trip_info.get_value(), n)

defget_and_clear_queue(
self,
queue_name,
payload_type=None,
message="not supported with this client/server combination",
 ):
ifpayload_type=="JSON":
min_version= (21, 0)
if (
get_client_version() <min_version
orget_server_version() <min_version
 ):
self.skipTest(message)
elifisinstance(payload_type, str):
payload_type=self.conn.gettype(payload_type)
queue=self.conn.queue(queue_name, payload_type)
queue.deqoptions.wait=oracledb.DEQ_NO_WAIT
queue.deqoptions.deliverymode=oracledb.MSG_PERSISTENT_OR_BUFFERED
queue.deqoptions.visibility=oracledb.DEQ_IMMEDIATE
whilequeue.deqone():
pass
returnself.conn.queue(queue_name, payload_type)

defget_db_object_as_plain_object(self, obj):
ifobj.type.iscollection:
element_values= []
forvalueinobj.aslist():
ifisinstance(value, oracledb.DbObject):
value=self.get_db_object_as_plain_object(value)
elifisinstance(value, oracledb.LOB):
value=value.read()
element_values.append(value)
returnelement_values
attr_values= []
forattributeinobj.type.attributes:
value=getattr(obj, attribute.name)
ifisinstance(value, oracledb.DbObject):
value=self.get_db_object_as_plain_object(value)
elifisinstance(value, oracledb.LOB):
value=value.read()
attr_values.append(value)
returntuple(attr_values)

defget_sid_serial(self, conn=None):
"""
 Returns the sid and serial number of the connection as a 2-tuple.
 """
ifconnisNone:
conn=self.conn
ifget_is_thin():
return (conn.session_id, conn.serial_num)
else:
withconn.cursor() ascursor:
cursor.execute(
"""
 select
 dbms_debug_jdwp.current_session_id,
 dbms_debug_jdwp.current_session_serial
 from dual
 """
 )
returncursor.fetchone()

defget_soda_database(
self,
minclient=(18, 3),
minserver=(18, 0),
message="not supported with this client/server combination",
drop_collections=True,
 ):
client=get_client_version()
ifclient<minclient:
self.skipTest(message)
server=get_server_version()
ifserver<minserver:
self.skipTest(message)
ifserver> (20, 1) andclient< (20, 1):
self.skipTest(message)
soda_db=self.conn.getSodaDatabase()
ifdrop_collections:
fornameinsoda_db.getCollectionNames():
soda_db.openCollection(name).drop()
returnsoda_db

defis_on_oracle_cloud(self, connection=None):
ifconnectionisNone:
connection=self.conn
returnis_on_oracle_cloud(connection)

defsetUp(self):
ifself.requires_connection:
self.conn=get_connection()
self.cursor=self.conn.cursor()
self.cursor.execute("alter session set time_zone = '+00:00'")

defsetup_parse_count_checker(self, conn=None):
ifget_is_implicit_pooling():
self.skipTest("sessions can change with implicit pooling")
self.parse_count_info=ParseCountInfo()
self.parse_count_info._initialize(connorself.conn)

defsetup_round_trip_checker(self, conn=None):
ifget_is_implicit_pooling():
self.skipTest("sessions can change with implicit pooling")
self.round_trip_info=RoundTripInfo()
self.round_trip_info._initialize(connorself.conn)

deftearDown(self):
ifself.requires_connection:
self.conn.close()
delself.cursor
delself.conn


classBaseAsyncTestCase(unittest.IsolatedAsyncioTestCase):
requires_connection=True

asyncdefassertParseCount(self, n):
self.assertEqual(awaitself.parse_count_info.get_value_async(), n)

defassertRaisesFullCode(self, *full_codes):
returnFullCodeErrorContextManager(full_codes)

asyncdefassertRoundTrips(self, n):
self.assertEqual(awaitself.round_trip_info.get_value_async(), n)

asyncdefasyncSetUp(self):
ifself.requires_connection:
self.conn=awaitget_connection_async()
self.cursor=self.conn.cursor()
awaitself.cursor.execute("alter session set time_zone = '+00:00'")

asyncdefasyncTearDown(self):
ifself.requires_connection:
awaitself.conn.close()
delself.cursor
delself.conn

asyncdefget_and_clear_queue(
self,
queue_name,
payload_type=None,
message="not supported with this client/server combination",
 ):
ifpayload_type=="JSON":
ifget_server_version() < (21, 0):
self.skipTest(message)
elifisinstance(payload_type, str):
payload_type=awaitself.conn.gettype(payload_type)
queue=self.conn.queue(queue_name, payload_type)
queue.deqoptions.wait=oracledb.DEQ_NO_WAIT
queue.deqoptions.deliverymode=oracledb.MSG_PERSISTENT_OR_BUFFERED
queue.deqoptions.visibility=oracledb.DEQ_IMMEDIATE
whileawaitqueue.deqone():
pass
returnself.conn.queue(queue_name, payload_type)

asyncdefget_db_object_as_plain_object(self, obj):
ifobj.type.iscollection:
element_values= []
forvalueinobj.aslist():
ifisinstance(value, oracledb.DbObject):
value=awaitself.get_db_object_as_plain_object(value)
elifisinstance(value, oracledb.AsyncLOB):
value=awaitvalue.read()
element_values.append(value)
returnelement_values
attr_values= []
forattributeinobj.type.attributes:
value=getattr(obj, attribute.name)
ifisinstance(value, oracledb.DbObject):
value=awaitself.get_db_object_as_plain_object(value)
elifisinstance(value, oracledb.AsyncLOB):
value=awaitvalue.read()
attr_values.append(value)
returntuple(attr_values)

asyncdefget_sid_serial(self, conn=None):
"""
 Returns the sid and serial number of the connection as a 2-tuple.
 """
ifconnisNone:
conn=self.conn
ifget_is_thin():
return (conn.session_id, conn.serial_num)
else:
withconn.cursor() ascursor:
awaitcursor.execute(
"""
 select
 dbms_debug_jdwp.current_session_id,
 dbms_debug_jdwp.current_session_serial
 from dual
 """
 )
returnawaitcursor.fetchone()

asyncdefis_on_oracle_cloud(self, connection=None):
ifconnectionisNone:
connection=self.conn
returnawaitis_on_oracle_cloud_async(connection)

asyncdefsetup_parse_count_checker(self, conn=None):
ifget_is_implicit_pooling():
self.skipTest("sessions can change with implicit pooling")
self.parse_count_info=ParseCountInfo()
awaitself.parse_count_info._initialize_async(connorself.conn)

asyncdefsetup_round_trip_checker(self, conn=None):
ifget_is_implicit_pooling():
self.skipTest("sessions can change with implicit pooling")
self.round_trip_info=RoundTripInfo()
awaitself.round_trip_info._initialize_async(connorself.conn)