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

"""
4500 - Module for testing connection parameters.
"""

import os
import tempfile

import oracledb
import test_env

class TestCase(test_env.BaseTestCase):
    requires_connection = False

    def __test_writable_parameter(self, name, value):
        """
        Tests that a writable parameter can be written to and the modified
        value read back successfully.
        """
        params = oracledb.ConnectParams()
        orig_value = getattr(params, name)
        copied_params = params.copy()
        args = {}
        args[name] = value
        params.set(**args)
        self.assertEqual(getattr(params, name), value)
        self.assertEqual(getattr(copied_params, name), orig_value)

    def test_4500_simple_easy_connect_with_port(self):
        "4500 - test simple EasyConnect string parsing with port specified"
        params = oracledb.ConnectParams()
        params.parse_connect_string("my_host:1578/my_service_name")
        self.assertEqual(params.host, "my_host")
        self.assertEqual(params.port, 1578)
        self.assertEqual(params.service_name, "my_service_name")

    def test_4501_simple_easy_connect_without_port(self):
        "4501 - test simple Easy Connect string parsing with no port specified"
        params = oracledb.ConnectParams()
        params.parse_connect_string("my_host2/my_service_name2")
        self.assertEqual(params.host, "my_host2")
        self.assertEqual(params.port, 1521)
        self.assertEqual(params.service_name, "my_service_name2")

    def test_4502_simple_easy_connect_with_server_type(self):
        "4502 - test simple EasyConnect string parsing with DRCP enabled"
        params = oracledb.ConnectParams()
        params.parse_connect_string("my_host3.org/my_service_name3:pooled")
        self.assertEqual(params.host, "my_host3.org")
        self.assertEqual(params.service_name, "my_service_name3")
        self.assertEqual(params.server_type, "pooled")
        params.parse_connect_string("my_host3/my_service_name3:ShArEd")
        self.assertEqual(params.server_type, "shared")

    def test_4503_simple_connect_descriptor(self):
        "4503 - test simple name-value pair format connect string"
        connect_string = \
            "(DESCRIPTION=(ADDRESS=(PROTOCOL=TCP)(HOST=my_host4)(PORT=1589))" \
            "(CONNECT_DATA=(SERVICE_NAME=my_service_name4)))"
        params = oracledb.ConnectParams()
        params.parse_connect_string(connect_string)
        self.assertEqual(params.host, "my_host4")
        self.assertEqual(params.port, 1589)
        self.assertEqual(params.service_name, "my_service_name4")

    def test_4504_search_tnsnames(self):
        "4504 - test simple tnsnames entry"
        connect_string = \
            "(DESCRIPTION=(ADDRESS=(PROTOCOL=TCP)(HOST=my_host5)(PORT=1624))" \
            "(CONNECT_DATA=(SERVICE_NAME=my_service_name5)))"
        alias = "tns_alias = " + connect_string
        with tempfile.TemporaryDirectory() as temp_dir:
            file_name = os.path.join(temp_dir, "tnsnames.ora")
            with open(file_name, "w") as f:
                f.write(alias)
            params = oracledb.ConnectParams(config_dir=temp_dir)
            params.parse_connect_string("tns_alias")
        self.assertEqual(params.host, "my_host5")
        self.assertEqual(params.port, 1624)
        self.assertEqual(params.service_name, "my_service_name5")

    def test_4505_easy_connect_with_protocol(self):
        "4505 - test EasyConnect with protocol"
        params = oracledb.ConnectParams()
        params.parse_connect_string("tcps://my_host6/my_service_name6")
        self.assertEqual(params.host, "my_host6")
        self.assertEqual(params.service_name, "my_service_name6")
        self.assertEqual(params.protocol, "tcps")

    def test_4506_easy_connect_with_invalid_protocol(self):
        "4506 - test EasyConnect with invalid protocol"
        params = oracledb.ConnectParams()
        self.assertRaisesRegex(oracledb.DatabaseError, "^DPY-4021:",
                               params.parse_connect_string,
                               "invalid_proto://my_host7/my_service_name7")

    def test_4507_exception_on_ipc_protocol(self):
        "4507 - confirm an exception is raised if using ipc protocol"
        connect_string = \
            "(DESCRIPTION=(ADDRESS=(PROTOCOL=ipc)(KEY=my_view8))" \
            "(CONNECT_DATA=(SERVICE_NAME=my_service_name8)))"
        params = oracledb.ConnectParams()
        self.assertRaisesRegex(oracledb.DatabaseError, "^DPY-4021:",
                               params.parse_connect_string, connect_string)

    def test_4508_retry_count_and_delay(self):
        "4508 - connect descriptor with retry count and retry delay"
        connect_string = \
            "(DESCRIPTION=(RETRY_COUNT=6)(RETRY_DELAY=5)" \
            "(ADDRESS=(PROTOCOL=TCP)(HOST=my_host9)(PORT=1593))" \
            "(CONNECT_DATA=(SERVICE_NAME=my_service_name9)))"
        params = oracledb.ConnectParams()
        params.parse_connect_string(connect_string)
        self.assertEqual(params.retry_count, 6)
        self.assertEqual(params.retry_delay, 5)

    def test_4509_connect_descriptor_expire_time(self):
        "4509 - connect descriptor with expire_time setting"
        connect_string = \
            "(DESCRIPTION=(EXPIRE_TIME=12)" \
            "(ADDRESS=(PROTOCOL=TCP)(HOST=my_host11)(PORT=1594))" \
            "(CONNECT_DATA=(SERVICE_NAME=my_service_name11)))"
        params = oracledb.ConnectParams()
        params.parse_connect_string(connect_string)
        self.assertEqual(params.expire_time, 12)

    def test_4510_pool_parameters(self):
        "4510 - connect descriptor with pool parameters"
        connect_string = \
            "(DESCRIPTION=(ADDRESS=(PROTOCOL=TCP)(HOST=my_host12)(PORT=694))" \
            "(CONNECT_DATA=(SERVICE_NAME=my_service_name12)" \
            "(POOL_CONNECTION_CLASS=cclass_12)(POOL_PURITY=SELF)))"
        params = oracledb.ConnectParams()
        params.parse_connect_string(connect_string)
        self.assertEqual(params.cclass, "cclass_12")
        self.assertEqual(params.purity, oracledb.PURITY_SELF)
        connect_string = connect_string.replace("SELF", "NEW")
        params.parse_connect_string(connect_string)
        self.assertEqual(params.purity, oracledb.PURITY_NEW)

    def test_4511_invalid_pool_purity(self):
        "4511 - connect descriptor with invalid pool purity"
        connect_string = \
            "(DESCRIPTION=(ADDRESS=(PROTOCOL=TCP)(HOST=my_host13)(PORT=695))" \
            "(CONNECT_DATA=(SERVICE_NAME=my_service_name13)" \
            "(POOL_CONNECTION_CLASS=cclass_13)(POOL_PURITY=INVALID)))"
        params = oracledb.ConnectParams()
        self.assertRaisesRegex(oracledb.DatabaseError, "^DPY-4022:",
                               params.parse_connect_string, connect_string)

    def test_4512_tcp_connect_timeout(self):
        "4512 - connect descriptor with transport connect timeout values"
        connect_string = \
            "(DESCRIPTION=(TRANSPORT_CONNECT_TIMEOUT=500 ms)" \
            "(ADDRESS=(PROTOCOL=TCP)(HOST=my_host14)(PORT=695))" \
            "(CONNECT_DATA=(SERVICE_NAME=my_service_name14)))"
        params = oracledb.ConnectParams()
        params.parse_connect_string(connect_string)
        self.assertEqual(params.tcp_connect_timeout, 0.5)
        connect_string = connect_string.replace("500 ms", "15 SEC")
        params.parse_connect_string(connect_string)
        self.assertEqual(params.tcp_connect_timeout, 15)
        connect_string = connect_string.replace("15 SEC", "5 min")
        params.parse_connect_string(connect_string)
        self.assertEqual(params.tcp_connect_timeout, 300)
        connect_string = connect_string.replace("5 min", "34")
        params.parse_connect_string(connect_string)
        self.assertEqual(params.tcp_connect_timeout, 34)

    def test_4513_easy_connect_without_service_name(self):
        "4513 - test EasyConnect string parsing with no service name specified"
        params = oracledb.ConnectParams()
        params.parse_connect_string("my_host15:1578/")
        self.assertEqual(params.host, "my_host15")
        self.assertEqual(params.port, 1578)
        self.assertEqual(params.service_name, "")

    def test_4514_missing_entry_in_tnsnames(self):
        "4514 - test missing entry in tnsnames"
        with tempfile.TemporaryDirectory() as temp_dir:
            params = oracledb.ConnectParams(config_dir=temp_dir)
            file_name = os.path.join(temp_dir, "tnsnames.ora")
            with open(file_name, "w") as f:
                f.write("# no entries")
            self.assertRaisesRegex(oracledb.DatabaseError, "DPY-4000",
                                   params.parse_connect_string, "tns_alias")

    def test_4515_easy_connect_with_port_missing(self):
        "4515 - test EasyConnect string parsing with port value missing"
        params = oracledb.ConnectParams()
        params.parse_connect_string("my_host17:/my_service_name17")
        self.assertEqual(params.host, "my_host17")
        self.assertEqual(params.port, 1521)
        self.assertEqual(params.service_name, "my_service_name17")

    def test_4516_invalid_number_in_connect_string(self):
        "4516 - test connect descriptor with invalid number"
        params = oracledb.ConnectParams()
        connect_string = \
            "(DESCRIPTION=(RETRY_COUNT=wrong)(RETRY_DELAY=5)" \
            "(ADDRESS=(PROTOCOL=TCP)(HOST=my_host18)(PORT=1598))" \
            "(CONNECT_DATA=(SERVICE_NAME=my_service_name18)))"
        self.assertRaisesRegex(oracledb.DatabaseError, "DPY-4018",
                               params.parse_connect_string, connect_string)

    def test_4517_security_options(self):
        "4517 - test connect descriptor with security options"
        options = [
            ("CN=unknown19a", "/tmp/wallet_loc19a", "On", True),
            ("CN=unknown19b", "/tmp/wallet_loc19b", "False", False),
            ("CN=unknown19c", "/tmp/wallet_loc19c", "Off", False),
            ("CN=unknown19d", "/tmp/wallet_loc19d", "True", True),
            ("CN=unknown19e", "/tmp/wallet_loc19e", "yes", True),
            ("CN=unknown19f", "/tmp/wallet_loc19f", "no", False)
        ]
        for dn, wallet_loc, match_option, match_value in options:
            params = oracledb.ConnectParams()
            connect_string = f"""
                (DESCRIPTION=(ADDRESS=(PROTOCOL=TCP)(HOST=my_host19)(PORT=872))
                (CONNECT_DATA=(SERVICE_NAME=my_service_name19))
                (SECURITY=(SSL_SERVER_CERT_DN="{dn}")
                (SSL_SERVER_DN_MATCH={match_option})
                (MY_WALLET_DIRECTORY="{wallet_loc}")))"""
            params.parse_connect_string(connect_string)
            self.assertEqual(params.ssl_server_cert_dn, dn)
            self.assertEqual(params.wallet_location, wallet_loc)
            self.assertEqual(params.ssl_server_dn_match, match_value)

    def test_4518_easy_connect_security_options(self):
        "4518 - test easy connect string with security options"
        options = [
            ("CN=unknown20a", "/tmp/wallet_loc20a", "On", True),
            ("CN=unknown20b", "/tmp/wallet_loc20b", "False", False),
            ("CN=unknown20c", "/tmp/wallet_loc20c", "Off", False),
            ("CN=unknown20d", "/tmp/wallet_loc20d", "True", True),
            ("CN=unknown20e", "/tmp/wallet_loc20e", "yes", True),
            ("CN=unknown20f", "/tmp/wallet_loc20f", "no", False)
        ]
        for dn, wallet_loc, match_option, match_value in options:
            params = oracledb.ConnectParams()
            connect_string = f'''
                my_host20/my_server_name20?
                ssl_server_cert_dn="{dn}"&
                ssl_server_dn_match= {match_option} &
                wallet_location = "{wallet_loc}"'''
            params = oracledb.ConnectParams()
            params.parse_connect_string(connect_string)
            self.assertEqual(params.ssl_server_cert_dn, dn)
            self.assertEqual(params.ssl_server_dn_match, match_value)
            self.assertEqual(params.wallet_location, wallet_loc)

    def test_4519_easy_connect_description_options(self):
        "4519 - test easy connect string with description options"
        params = oracledb.ConnectParams()
        connect_string = 'my_host21/my_server_name21?' \
            'expire_time=5&' \
            'retry_delay=10&' \
            'retry_count=12&' \
            'transport_connect_timeout=2.5'
        params.parse_connect_string(connect_string)
        self.assertEqual(params.expire_time, 5)
        self.assertEqual(params.retry_delay, 10)
        self.assertEqual(params.retry_count, 12)
        self.assertEqual(params.tcp_connect_timeout, 2.5)

    def test_4520_easy_connect_invalid_parameters(self):
        "4520 - test easy connect string with invalid parameters"
        params = oracledb.ConnectParams()
        connect_string_prefix = 'my_host22/my_server_name22?'
        suffixes = [
            'expire_time=invalid',
            'expire_time'
        ]
        for suffix in suffixes:
            self.assertRaisesRegex(oracledb.DatabaseError, "DPY-4018",
                                   params.parse_connect_string,
                                   connect_string_prefix + suffix)

    def test_4521_connect_string_with_newlines_and_spaces(self):
        "4521 - test connect string containing spaces and newlines"
        params = oracledb.ConnectParams()
        connect_string = \
            '(DESCRIPTION=(ADDRESS=(PROTOCOL=TCP) \n(HOST=my_host23)\n' \
            '(PORT=1560))(CONNECT_DATA=  (SERVICE_NAME=my_service_name23))' \
            '(SECURITY=(MY_WALLET_DIRECTORY="my wallet dir 23")))'
        params.parse_connect_string(connect_string)
        self.assertEqual(params.host, "my_host23")
        self.assertEqual(params.port, 1560)
        self.assertEqual(params.service_name, "my_service_name23")
        self.assertEqual(params.wallet_location, "my wallet dir 23")

    def test_4522_missing_tnsnames(self):
        "4522 - test missing tnsnames.ora in configuration directory"
        with tempfile.TemporaryDirectory() as temp_dir:
            params = oracledb.ConnectParams(config_dir=temp_dir)
            self.assertRaisesRegex(oracledb.DatabaseError, "DPY-4026:",
                                   params.parse_connect_string, "tns_alias")

    def test_4523_missing_config_dir(self):
        "4523 - test missing configuration directory"
        params = oracledb.ConnectParams(config_dir="/missing")
        self.assertRaisesRegex(oracledb.DatabaseError, "DPY-4026:",
                               params.parse_connect_string, "tns_alias")

    def test_4524_invalid_entries_in_tnsnames(self):
        "4524 - test tnsnames.ora with invalid entries"
        connect_string = \
            "(DESCRIPTION=(ADDRESS=(PROTOCOL=TCP)(HOST=my_host24)(PORT=1148))" \
            "(CONNECT_DATA=(SERVICE_NAME=my_service_name24)))"
        alias = f"tns_alias24 = {connect_string}"
        with tempfile.TemporaryDirectory() as temp_dir:
            file_name = os.path.join(temp_dir, "tnsnames.ora")
            with open(file_name, "w") as f:
                print("invalid_alias = something to ignore", file=f)
                print("some garbage data which should be ignored", file=f)
                print(alias, file=f)
            params = oracledb.ConnectParams(config_dir=temp_dir)
            params.parse_connect_string("tns_alias24")
        self.assertEqual(params.host, "my_host24")
        self.assertEqual(params.port, 1148)
        self.assertEqual(params.service_name, "my_service_name24")

    def test_4525_single_address_list(self):
        "4525 - test connect string with an address list"
        params = oracledb.ConnectParams()
        connect_string = \
            '(DESCRIPTION=(LOAD_BALANCE=ON)(RETRY_COUNT=5)(RETRY_DELAY=2)'  \
            '(ADDRESS_LIST=(LOAD_BALANCE=ON)' \
            '(ADDRESS=(PROTOCOL=tcp)(PORT=1521)(HOST=my_host25))' \
            '(ADDRESS=(PROTOCOL=tcps)(PORT=222)(HOST=my_host26)))' \
            '(CONNECT_DATA=(SERVICE_NAME=my_service_name25)))'
        params.parse_connect_string(connect_string)
        self.assertEqual(params.host, ["my_host25", "my_host26"])
        self.assertEqual(params.port, [1521, 222])
        self.assertEqual(params.protocol, ["tcp", "tcps"])
        self.assertEqual(params.service_name, "my_service_name25")
        self.assertEqual(params.retry_count, 5)
        self.assertEqual(params.retry_delay, 2)

    def test_4526_multiple_address_lists(self):
        "4526 - test connect string with multiple address lists"
        params = oracledb.ConnectParams()
        connect_string = \
            '(DESCRIPTION=(LOAD_BALANCE=ON)(RETRY_COUNT=5)(RETRY_DELAY=2)'  \
            '(ADDRESS_LIST=(LOAD_BALANCE=ON)' \
            '(ADDRESS=(PROTOCOL=tcp)(PORT=1521)(HOST=my_host26))' \
            '(ADDRESS=(PROTOCOL=tcp)(PORT=222)(HOST=my_host27)))' \
            '(ADDRESS_LIST=(LOAD_BALANCE=ON)' \
            '(ADDRESS=(PROTOCOL=tcps)(PORT=5555)(HOST=my_host28))' \
            '(ADDRESS=(PROTOCOL=tcps)(PORT=444)(HOST=my_host29)))' \
            '(CONNECT_DATA=(SERVICE_NAME=my_service_name26)))'
        params.parse_connect_string(connect_string)
        hosts = ["my_host26", "my_host27", "my_host28", "my_host29"]
        self.assertEqual(params.host, hosts)
        self.assertEqual(params.port, [1521, 222, 5555, 444])
        self.assertEqual(params.protocol, ["tcp", "tcp", "tcps", "tcps"])
        self.assertEqual(params.service_name, "my_service_name26")
        self.assertEqual(params.retry_count, 5)
        self.assertEqual(params.retry_delay, 2)

    def test_4527_multiple_descriptions(self):
        "4527 - test connect string with multiple descriptions"
        params = oracledb.ConnectParams()
        connect_string = \
            '(DESCRIPTION_LIST=(FAIL_OVER=ON)(LOAD_BALANCE=OFF)' \
            '(DESCRIPTION=(LOAD_BALANCE=OFF)(RETRY_COUNT=1)(RETRY_DELAY=1)' \
            '(ADDRESS_LIST=(ADDRESS=(PROTOCOL=tcp)(PORT=5001)' \
            '(HOST=my_host30))' \
            '(ADDRESS=(PROTOCOL=tcp)(PORT=1521)(HOST=my_host31)))' \
            '(ADDRESS_LIST=(ADDRESS=(PROTOCOL=tcp)(PORT=5002) ' \
            '(HOST=my_host32))' \
            '(ADDRESS=(PROTOCOL=tcp)(PORT=5003)(HOST=my_host33)))' \
            '(CONNECT_DATA=(SERVICE_NAME=my_service_name27)))' \
            '(DESCRIPTION=(LOAD_BALANCE=OFF)(RETRY_COUNT=2)(RETRY_DELAY=3)' \
            '(ADDRESS_LIST = (ADDRESS=(PROTOCOL=tcp)(PORT=5001)' \
            '(HOST=my_host34))' \
            '(ADDRESS=(PROTOCOL=tcp)(PORT=5001)(HOST=my_host35)))' \
            '(ADDRESS_LIST=(ADDRESS=(PROTOCOL=tcp)(PORT=5001)' \
            '(HOST=my_host36))' \
            '(ADDRESS=(PROTOCOL=tcps)(HOST=my_host37)(PORT=1521)))' \
            '(CONNECT_DATA=(SERVICE_NAME=my_service_name28))))'
        params.parse_connect_string(connect_string)
        hosts = ["my_host30", "my_host31", "my_host32", "my_host33", \
                 "my_host34", "my_host35", "my_host36", "my_host37"]
        ports = [5001, 1521, 5002, 5003, 5001, 5001, 5001, 1521]
        protocols = ["tcp", "tcp", "tcp", "tcp", "tcp", "tcp", "tcp", "tcps"]
        service_names = ["my_service_name27", "my_service_name28"]
        self.assertEqual(params.host, hosts)
        self.assertEqual(params.port, ports)
        self.assertEqual(params.protocol, protocols)
        self.assertEqual(params.service_name, service_names)
        self.assertEqual(params.retry_count, [1, 2])
        self.assertEqual(params.retry_delay, [1, 3])

    def test_4528_https_proxy(self):
        "4528 - test connect strings with https_proxy defined"
        params = oracledb.ConnectParams()
        connect_string = \
            "(DESCRIPTION=" \
            "(ADDRESS=(HTTPS_PROXY=proxy_4528a)(HTTPS_PROXY_PORT=4528)" \
            "(PROTOCOL=TCP)(HOST=my_host4528a)(PORT=8528))" \
            "(CONNECT_DATA=(SERVICE_NAME=my_service_name4528a)))"
        params.parse_connect_string(connect_string)
        self.assertEqual(params.https_proxy, "proxy_4528a")
        self.assertEqual(params.https_proxy_port, 4528)
        connect_string = "tcps://my_host_4528b/my_service_name_4528b?" \
                "https_proxy=proxy_4528b&https_proxy_port=9528"
        params.parse_connect_string(connect_string)
        self.assertEqual(params.https_proxy, "proxy_4528b")
        self.assertEqual(params.https_proxy_port, 9528)

    def test_4529_server_type(self):
        "4529 - test connect strings with server_type defined"
        params = oracledb.ConnectParams()
        connect_string = \
            "(DESCRIPTION=" \
            "(ADDRESS=(PROTOCOL=TCP)(HOST=my_host4529)(PORT=4529))" \
            "(CONNECT_DATA=(SERVER=DEDICATED)" \
            "(SERVICE_NAME=my_service_name4529)))"
        params.parse_connect_string(connect_string)
        self.assertEqual(params.server_type, "dedicated")
        connect_string = connect_string.replace("DEDICATED", "INVALID")
        self.assertRaisesRegex(oracledb.DatabaseError, "^DPY-4028:",
                               params.parse_connect_string, connect_string)

    def test_4530_writable_params(self):
        "4530 - test writable parameters"
        self.__test_writable_parameter("appcontext", [("a", "b", "c")])
        self.__test_writable_parameter("config_dir", "config_dir_4530")
        self.__test_writable_parameter("disable_oob", True)
        self.__test_writable_parameter("edition", "edition_4530")
        self.__test_writable_parameter("events", True)
        self.__test_writable_parameter("matchanytag", True)
        self.__test_writable_parameter("mode", oracledb.AUTH_MODE_SYSDBA)
        self.__test_writable_parameter("shardingkey", [1, 2, 3])
        self.__test_writable_parameter("stmtcachesize", 25)
        self.__test_writable_parameter("supershardingkey", [1, 2, 3])
        self.__test_writable_parameter("tag", "tag_4530")
        self.__test_writable_parameter("debug_jdwp", "host=host;port=4530")
        self.__test_writable_parameter("externalauth", True)

    def test_4531_build_connect_string_with_tcp_connect_timeout(self):
        "4531 - test building connect string with TCP connect timeout"
        host = "my_host4531"
        service_name = "my_service4531"
        options = [
            (25, "25"),
            (120, "2min"),
            (2.5, "2500ms"),
            (3.4328, "3432ms")
        ]
        for in_val, out_val in options:
            params = oracledb.ConnectParams(host=host,
                                            service_name=service_name,
                                            tcp_connect_timeout=in_val)
            tcp_timeout_val = f"(TRANSPORT_CONNECT_TIMEOUT={out_val})"
            connect_string = f"(DESCRIPTION={tcp_timeout_val}" + \
                             f"(ADDRESS_LIST=(ADDRESS=(PROTOCOL=tcp)" + \
                             f"(HOST={host})(PORT=1521)))(CONNECT_DATA=" + \
                             f"(SERVICE_NAME={service_name}))" + \
                             f"(SECURITY=(SSL_SERVER_DN_MATCH=ON)))"
            self.assertEqual(params.get_connect_string(), connect_string)

    def test_4532_multiple_alias_entry_tnsnames(self):
        "4532 - test tnsnames.ora with multiple aliases on one line"
        connect_string = \
            "(DESCRIPTION=(ADDRESS=(PROTOCOL=TCP)(HOST=my_host32)(PORT=1132))" \
            "(CONNECT_DATA=(SERVICE_NAME=my_service_name32)))"
        aliases = f"tns_alias32a,tns_alias32b = {connect_string}"
        with tempfile.TemporaryDirectory() as temp_dir:
            file_name = os.path.join(temp_dir, "tnsnames.ora")
            with open(file_name, "w") as f:
                print(aliases, file=f)
            params = oracledb.ConnectParams(config_dir=temp_dir)
            for name in ("tns_alias32a", "tns_alias32b"):
                params.parse_connect_string(name)
                self.assertEqual(params.host, "my_host32")
                self.assertEqual(params.port, 1132)
                self.assertEqual(params.service_name, "my_service_name32")

    def test_4533_easy_connect_with_pool_parameters(self):
        "4533 - test EasyConnect with pool parameters"
        options = [
            ("cclass_33a", "self", oracledb.PURITY_SELF),
            ("cclass_33b", "new", oracledb.PURITY_NEW)
        ]
        for cclass, purity_str, purity_int in options:
            connect_string = f"my_host_33/my_service_name_33:pooled?" \
                             f"pool_connection_class={cclass}&" \
                             f"pool_purity={purity_str}"
            params = oracledb.ConnectParams()
            params.parse_connect_string(connect_string)
            self.assertEqual(params.host, "my_host_33")
            self.assertEqual(params.service_name, "my_service_name_33")
            self.assertEqual(params.port, 1521)
            self.assertEqual(params.server_type, "pooled")
            self.assertEqual(params.cclass, cclass)
            self.assertEqual(params.purity, purity_int)

    def test_4534_connect_descriptor_small_container_first(self):
        "4534 - test connect descriptor with different containers (small 1st)"
        connect_string = """
            (DESCRIPTION=
                (ADDRESS=(PROTOCOL=tcp)(HOST=host1)(PORT=1521))
                (ADDRESS_LIST=
                    (ADDRESS=(PROTOCOL=tcp)(HOST=host2a)(PORT=1522))
                    (ADDRESS=(PROTOCOL=tcp)(HOST=host2b)(PORT=1523)))
                (ADDRESS=(PROTOCOL=tcp)(HOST=host3)(PORT=1524))
                (CONNECT_DATA=(SERVICE_NAME=my_service_34))
            )"""
        params = oracledb.ConnectParams()
        params.parse_connect_string(connect_string)
        self.assertEqual(params.host, ["host1", "host2a", "host2b", "host3"])

    def test_4535_connect_descriptor_small_container_second(self):
        "4535 - test connect descriptor with different containers (small 2nd)"
        connect_string = """
            (DESCRIPTION=
                (ADDRESS_LIST=
                    (ADDRESS=(PROTOCOL=tcp)(HOST=host1a)(PORT=1532))
                    (ADDRESS=(PROTOCOL=tcp)(HOST=host1b)(PORT=1533)))
                (ADDRESS=(PROTOCOL=tcp)(HOST=host2)(PORT=1534))
                (ADDRESS_LIST=
                    (ADDRESS=(PROTOCOL=tcp)(HOST=host3a)(PORT=1535))
                    (ADDRESS=(PROTOCOL=tcp)(HOST=host3b)(PORT=1536)))
                (CONNECT_DATA=(SERVICE_NAME=my_service_34))
            )"""
        params = oracledb.ConnectParams()
        params.parse_connect_string(connect_string)
        self.assertEqual(params.host,
                         ["host1a", "host1b", "host2", "host3a", "host3b"])

    def test_4536_build_connect_string_with_source_route(self):
        "4536 - test building connect string with source route designation"
        host = "my_host4536"
        service_name = "my_service4536"
        options = [
            ("on", True),
            ("off", False),
            ("true", True),
            ("false", False),
            ("yes", True),
            ("no", False)
        ]

        for in_val, has_section in options:
            connect_string = f"""
                (DESCRIPTION=
                    (SOURCE_ROUTE={in_val})
                    (ADDRESS=(PROTOCOL=tcp)(HOST=host1)(PORT=1521))
                    (ADDRESS=(PROTOCOL=tcp)(HOST=host2)(PORT=1522))
                    (CONNECT_DATA=(SERVICE_NAME=my_service_35))
                )"""
            params = oracledb.ConnectParams()
            params.parse_connect_string(connect_string)
            source_route_clause = "(SOURCE_ROUTE=ON)" if has_section else ""
            connect_string = \
                    f"(DESCRIPTION={source_route_clause}" + \
                    f"(ADDRESS_LIST=" + \
                    f"(ADDRESS=(PROTOCOL=tcp)(HOST=host1)(PORT=1521))" + \
                    f"(ADDRESS=(PROTOCOL=tcp)(HOST=host2)(PORT=1522)))" + \
                    f"(CONNECT_DATA=(SERVICE_NAME=my_service_35))" + \
                    f"(SECURITY=(SSL_SERVER_DN_MATCH=ON)))"
            self.assertEqual(params.get_connect_string(), connect_string)

if __name__ == "__main__":
    test_env.run_test_cases()