#------------------------------------------------------------------------------ # 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. #------------------------------------------------------------------------------ #------------------------------------------------------------------------------ # 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_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. #------------------------------------------------------------------------------ import getpass import os import sys import unittest import oracledb # default values DEFAULT_MAIN_USER = "pythontest" DEFAULT_PROXY_USER = "pythontestproxy" DEFAULT_CONNECT_STRING = "localhost/orclpdb1" # 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 get_value(name, label, default_value=None, password=False): try: return PARAMETERS[name] except KeyError: pass env_name = "PYO_TEST_" + name value = os.environ.get(env_name) if value is None: if default_value is not None: label += " [%s]" % default_value label += ": " if password: value = getpass.getpass(label) else: value = input(label).strip() if not value: value = default_value PARAMETERS[name] = value return value def get_admin_connection(): 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() if admin_user and admin_user.upper() == "SYS": params = params.copy() params.set(mode=oracledb.AUTH_MODE_SYSDBA) return oracledb.connect(dsn=get_connect_string(), params=params, user=admin_user, password=admin_password) def get_charset_ratios(): value = PARAMETERS.get("CS_RATIO") if value is None: 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 return value def get_client_version(): name = "CLIENT_VERSION" value = PARAMETERS.get(name) if value is None: if get_is_thin(): value = (21, 3) else: value = oracledb.clientversion()[:2] PARAMETERS[name] = value return value def get_connect_params(): name = "CONNECT_PARAMS" params = PARAMETERS.get(name) if params is None: wallet_location = get_wallet_location() params = oracledb.ConnectParams(user=get_main_user(), password=get_main_password(), config_dir=wallet_location, wallet_location=wallet_location, wallet_password=get_wallet_password()) PARAMETERS[name] = params return params def get_connection(dsn=None, **kwargs): if dsn is None: dsn = get_connect_string() return oracledb.connect(dsn=dsn, params=get_connect_params(), **kwargs) def get_connect_string(): return get_value("CONNECT_STRING", "Connect String", DEFAULT_CONNECT_STRING) def get_is_thin(): driver_mode = get_value("DRIVER_MODE", "Driver mode (thin|thick)", "thin") return driver_mode == "thin" def get_main_password(): return get_value("MAIN_PASSWORD", f"Password for {get_main_user()}", password=True) def get_main_user(): return get_value("MAIN_USER", "Main User Name", DEFAULT_MAIN_USER) def get_pool(**kwargs): return oracledb.create_pool(dsn=get_connect_string(), params=get_pool_params(), **kwargs) def get_pool_params(): name = "POOL_PARAMS" params = PARAMETERS.get(name) if params is None: wallet_location = get_wallet_location() params = oracledb.PoolParams(user=get_main_user(), password=get_main_password(), config_dir=wallet_location, wallet_location=wallet_location, wallet_password=get_wallet_password()) PARAMETERS[name] = params return params def get_proxy_password(): return get_value("PROXY_PASSWORD", f"Password for {get_proxy_user()}", password=True) def get_proxy_user(): return get_value("PROXY_USER", "Proxy User Name", DEFAULT_PROXY_USER) def get_sleep_proc_name(): server_version = get_server_version() return "dbms_session.sleep" if server_version[0] >= 18 \ else "dbms_lock.sleep" def get_server_version(): name = "SERVER_VERSION" value = PARAMETERS.get(name) if value is None: conn = get_connection() value = tuple(int(s) for s in conn.version.split("."))[:2] PARAMETERS[name] = value return value def get_wallet_location(): if get_is_thin(): return get_value("WALLET_LOCATION", "Wallet Location") def get_wallet_password(): if get_is_thin(): return get_value("WALLET_PASSWORD", "Wallet Password", password=True) def get_external_user(): if not get_is_thin(): return get_value("EXTERNAL_USER", "External User") def is_on_oracle_cloud(connection): server = get_server_version() if server < (18, 0): return False cursor = connection.cursor() cursor.execute(""" select sys_context('userenv', 'cloud_service') from dual""") service_name, = cursor.fetchone() return service_name is not None def run_sql_script(conn, script_name, **kwargs): statement_parts = [] cursor = conn.cursor() replace_values = [("&" + k + ".", v) for k, v in kwargs.items()] + \ [("&" + k, v) for k, v in kwargs.items()] script_dir = os.path.dirname(os.path.abspath(sys.argv[0])) file_name = os.path.join(script_dir, "sql", script_name + ".sql") for line in open(file_name): if line.strip() == "/": statement = "".join(statement_parts).strip() if statement: for search_value, replace_value in replace_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 for name, obj_type, line_num, position, text in cursor: if name != prev_name or obj_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)) def run_test_cases(): unittest.main(testRunner=unittest.TextTestRunner(verbosity=2)) def skip_soda_tests(): if get_is_thin(): return True client = get_client_version() if client < (18, 3): return True server = get_server_version() if server < (18, 0): return True if server > (20, 1) and client < (20, 1): return True return False class FetchLobsContextManager: def __init__(self, desired_value): self.desired_value = desired_value def __enter__(self): self.original_value = oracledb.defaults.fetch_lobs oracledb.defaults.fetch_lobs = self.desired_value return self def __exit__(self, exc_type, exc_val, exc_tb): oracledb.defaults.fetch_lobs = self.original_value class SystemStatInfo: stat_name = None def __init__(self, connection): self.prev_value = 0 self.admin_conn = get_admin_connection() with connection.cursor() as cursor: cursor.execute("select sys_context('userenv', 'sid') from dual") self.sid, = cursor.fetchone() self.get_value() def get_value(self): with self.admin_conn.cursor() as cursor: cursor.execute(""" select ss.value from v$sesstat ss, v$statname sn where ss.sid = :sid and ss.statistic# = sn.statistic# and sn.name = :stat_name""", sid=self.sid, stat_name=self.stat_name) current_value, = cursor.fetchone() diff_value = current_value - self.prev_value self.prev_value = current_value return diff_value class RoundTripInfo(SystemStatInfo): stat_name = "SQL*Net roundtrips to/from client" class ParseCountInfo(SystemStatInfo): stat_name = "parse count (total)" class BaseTestCase(unittest.TestCase): requires_connection = True def assertParseCount(self, n): self.assertEqual(self.parse_count_info.get_value(), n) def assertRoundTrips(self, n): self.assertEqual(self.round_trip_info.get_value(), n) def get_and_clear_queue(self, queue_name, payload_type=None, message="not supported with this client/server " \ "combination"): if payload_type == "JSON": if get_client_version() < (21, 0) \ or get_server_version() < (21, 0): self.skipTest(message) elif isinstance(payload_type, str): payload_type = self.connection.gettype(payload_type) queue = self.connection.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 while queue.deqone(): pass return self.connection.queue(queue_name, payload_type) def get_db_object_as_plain_object(self, obj): if obj.type.iscollection: element_values = [] for value in obj.aslist(): if isinstance(value, oracledb.DbObject): value = self.get_db_object_as_plain_object(value) elif isinstance(value, oracledb.LOB): value = value.read() element_values.append(value) return element_values attr_values = [] for attribute in obj.type.attributes: value = getattr(obj, attribute.name) if isinstance(value, oracledb.DbObject): value = self.get_db_object_as_plain_object(value) elif isinstance(value, oracledb.LOB): value = value.read() attr_values.append(value) return tuple(attr_values) def get_soda_database(self, minclient=(18, 3), minserver=(18, 0), message="not supported with this client/server " \ "combination"): client = get_client_version() if client < minclient: self.skipTest(message) server = get_server_version() if server < minserver: self.skipTest(message) if server > (20, 1) and client < (20, 1): self.skipTest(message) return self.connection.getSodaDatabase() def is_on_oracle_cloud(self, connection=None): if connection is None: connection = self.connection return is_on_oracle_cloud(connection) def setUp(self): if self.requires_connection: self.connection = get_connection() self.cursor = self.connection.cursor() def setup_parse_count_checker(self): self.parse_count_info = ParseCountInfo(self.connection) def setup_round_trip_checker(self): self.round_trip_info = RoundTripInfo(self.connection) def tearDown(self): if self.requires_connection: self.connection.close() del self.cursor del self.connection # ensure that thick mode is enabled, if desired if not get_is_thin(): oracledb.init_oracle_client()