#------------------------------------------------------------------------------
# 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.
#------------------------------------------------------------------------------

"""
2400 - Module for testing pools
"""

import threading
import unittest

import oracledb
import test_env

class TestCase(test_env.BaseTestCase):
    require_connection = False

    def __connect_and_drop(self):
        with self.pool.acquire() as connection:
            cursor = connection.cursor()
            cursor.execute("select count(*) from TestNumbers")
            count, = cursor.fetchone()
            self.assertEqual(count, 10)

    def __connect_and_generate_error(self):
        with self.pool.acquire() as connection:
            cursor = connection.cursor()
            self.assertRaisesRegex(oracledb.DatabaseError,"^ORA-01476:",
                                   cursor.execute, "select 1 / 0 from dual")

    def __callable_session_callback(self, conn, requested_tag):
        self.session_called = True

        supported_formats = {
            "SIMPLE" : "'YYYY-MM-DD HH24:MI'",
            "FULL" : "'YYYY-MM-DD HH24:MI:SS'"
        }

        supported_time_zones = {
            "UTC" : "'UTC'",
            "MST" : "'-07:00'"
        }

        supported_keys = {
            "NLS_DATE_FORMAT" : supported_formats,
            "TIME_ZONE" : supported_time_zones
        }
        if requested_tag is not None:
            state_parts = []
            for directive in requested_tag.split(";"):
                parts = directive.split("=")
                if len(parts) != 2:
                    raise ValueError("Tag must contain key=value pairs")
                key, value = parts
                value_dict = supported_keys.get(key)
                if value_dict is None:
                    raise ValueError("Tag only supports keys: %s" % \
                                     (", ".join(supported_keys)))
                actual_value = value_dict.get(value)
                if actual_value is None:
                    raise ValueError("Key %s only supports values: %s" % \
                                     (key, ", ".join(value_dict)))
                state_parts.append("%s = %s" % (key, actual_value))
            sql = "alter session set %s" % " ".join(state_parts)
            cursor = conn.cursor()
            cursor.execute(sql)
        conn.tag = requested_tag

    def __perform_reconfigure_test(self, parameter_name, parameter_value,
                                   min=3, max=30, increment=4, timeout=5,
                                   wait_timeout=5000, stmtcachesize=25,
                                   max_lifetime_session=1000,
                                   max_sessions_per_shard=3, ping_interval=30,
                                   getmode=oracledb.POOL_GETMODE_WAIT):
        creation_args = dict(min=min, max=max, increment=increment,
                             timeout=timeout, wait_timeout=wait_timeout,
                             stmtcachesize=stmtcachesize,
                             max_lifetime_session=max_lifetime_session,
                             max_sessions_per_shard=max_sessions_per_shard,
                             ping_interval=ping_interval, getmode=getmode)
        reconfigure_args = {}
        reconfigure_args[parameter_name] = parameter_value

        pool = test_env.get_pool(**creation_args)
        connection = pool.acquire()
        pool.reconfigure(**reconfigure_args)
        actual_args = dict(min=pool.min, max=pool.max,
                           increment=pool.increment, timeout=pool.timeout,
                           wait_timeout=pool.wait_timeout,
                           stmtcachesize=pool.stmtcachesize,
                           max_lifetime_session=pool.max_lifetime_session,
                           max_sessions_per_shard=pool.max_sessions_per_shard,
                           ping_interval=pool.ping_interval,
                           getmode=pool.getmode)
        expected_args = creation_args.copy()
        expected_args.update(reconfigure_args)
        self.assertEqual(actual_args, expected_args)

    def __verify_connection(self, connection, expected_user,
                            expected_proxy_user=None):
        cursor = connection.cursor()
        cursor.execute("""
                select
                    sys_context('userenv', 'session_user'),
                    sys_context('userenv', 'proxy_user')
                from dual""")
        actual_user, actual_proxy_user = cursor.fetchone()
        self.assertEqual(actual_user, expected_user.upper())
        self.assertEqual(actual_proxy_user,
                         expected_proxy_user and expected_proxy_user.upper())

    def test_2400_pool(self):
        "2400 - test that the pool is created and has the right attributes"
        pool = test_env.get_pool(min=2, max=8, increment=3,
                                 getmode=oracledb.POOL_GETMODE_WAIT)
        self.assertEqual(pool.username, test_env.get_main_user(),
                         "user name differs")
        self.assertEqual(pool.dsn, test_env.get_connect_string(),
                         "dsn differs")
        self.assertEqual(pool.max, 8, "max differs")
        self.assertEqual(pool.min, 2, "min differs")
        self.assertEqual(pool.increment, 3, "increment differs")
        self.assertEqual(pool.busy, 0, "busy not 0 at start")
        connection_1 = pool.acquire()
        self.assertEqual(pool.busy, 1, "busy not 1 after acquire")
        connection_2 = oracledb.connect(pool=pool)
        self.assertEqual(pool.busy, 2, "busy not 2 after acquire")
        self.assertEqual(pool.opened, 2, "opened differs")
        connection_3 = pool.acquire()
        self.assertEqual(pool.busy, 3, "busy not 3 after acquire")
        pool.release(connection_3)
        self.assertEqual(pool.busy, 2, "busy not 2 after release")
        pool.release(connection_1)
        connection_2.close()
        self.assertEqual(pool.busy, 0, "busy not 0 after release")
        pool.getmode = oracledb.POOL_GETMODE_NOWAIT
        self.assertEqual(pool.getmode, oracledb.POOL_GETMODE_NOWAIT)
        if test_env.get_client_version() >= (12, 2):
            pool.getmode = oracledb.POOL_GETMODE_TIMEDWAIT
            self.assertEqual(pool.getmode, oracledb.POOL_GETMODE_TIMEDWAIT)
        pool.stmtcachesize = 50
        self.assertEqual(pool.stmtcachesize, 50)
        pool.timeout = 10
        self.assertEqual(pool.timeout, 10)
        if test_env.get_client_version() >= (12, 1):
            pool.max_lifetime_session = 10
            self.assertEqual(pool.max_lifetime_session, 10)

    @unittest.skipIf(test_env.get_is_thin(),
                     "thin mode doesn't support proxy users yet")
    def test_2401_proxy_auth(self):
        "2401 - test that proxy authentication is possible"
        pool = test_env.get_pool(min=2, max=8, increment=3,
                                 getmode=oracledb.POOL_GETMODE_WAIT)
        self.assertEqual(pool.homogeneous, True,
                         "homogeneous should be True by default")
        self.assertRaisesRegex(oracledb.DatabaseError, "^DPI-1012:",
                               pool.acquire, user="missing_proxyuser")
        pool = test_env.get_pool(min=2, max=8, increment=3,
                                 getmode=oracledb.POOL_GETMODE_WAIT,
                                 homogeneous=False)
        msg = "homogeneous should be False after setting it in the constructor"
        self.assertEqual(pool.homogeneous, False, msg)
        connection = pool.acquire(user=test_env.get_proxy_user())
        cursor = connection.cursor()
        cursor.execute('select user from dual')
        result, = cursor.fetchone()
        self.assertEqual(result, test_env.get_proxy_user().upper())
        connection.close()

    def test_2403_rollback_on_release(self):
        "2403 - connection rolls back before released back to the pool"
        pool = test_env.get_pool(getmode=oracledb.POOL_GETMODE_WAIT)
        connection = pool.acquire()
        cursor = connection.cursor()
        cursor.execute("truncate table TestTempTable")
        cursor.execute("insert into TestTempTable (IntCol) values (1)")
        cursor.close()
        pool.release(connection)
        pool = test_env.get_pool(getmode=oracledb.POOL_GETMODE_WAIT)
        connection = pool.acquire()
        cursor = connection.cursor()
        cursor.execute("select count(*) from TestTempTable")
        count, = cursor.fetchone()
        self.assertEqual(count, 0)
        connection.close()

    def test_2404_threading(self):
        "2404 - test session pool with multiple threads"
        self.pool = test_env.get_pool(min=5, max=20, increment=2,
                                      getmode=oracledb.POOL_GETMODE_WAIT)
        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_2405_threading_with_errors(self):
        "2405 - test session pool with multiple threads (with errors)"
        self.pool = test_env.get_pool(min=5, max=20, increment=2,
                                      getmode=oracledb.POOL_GETMODE_WAIT)
        threads = []
        for i in range(20):
            thread = threading.Thread(None, self.__connect_and_generate_error)
            threads.append(thread)
            thread.start()
        for thread in threads:
            thread.join()

    def test_2406_purity(self):
        "2406 - test session pool with various types of purity"
        pool = test_env.get_pool(min=1, max=8, increment=1,
                                 getmode=oracledb.POOL_GETMODE_WAIT)

        # get connection and set the action
        action = "TEST_ACTION"
        connection = pool.acquire()
        connection.action = action
        cursor = connection.cursor()
        cursor.execute("select 1 from dual")
        cursor.close()
        pool.release(connection)
        self.assertEqual(pool.opened, 1, "opened (1)")

        # verify that the connection still has the action set on it
        connection = pool.acquire()
        cursor = connection.cursor()
        cursor.execute("select sys_context('userenv', 'action') from dual")
        result, = cursor.fetchone()
        self.assertEqual(result, action)
        cursor.close()
        pool.release(connection)
        self.assertEqual(pool.opened, 1, "opened (2)")

        # get a new connection with new purity (should not have state)
        connection = pool.acquire(purity=oracledb.ATTR_PURITY_NEW)
        cursor = connection.cursor()
        cursor.execute("select sys_context('userenv', 'action') from dual")
        result, = cursor.fetchone()
        self.assertEqual(result, None)
        cursor.close()
        pool.release(connection)

    @unittest.skipIf(test_env.get_is_thin(),
                     "thin mode doesn't support proxy users yet")
    def test_2407_heterogeneous(self):
        "2407 - test heterogeneous pool with user and password specified"
        pool = test_env.get_pool(min=2, max=8, increment=3, homogeneous=False,
                                 getmode=oracledb.POOL_GETMODE_WAIT)
        self.assertEqual(pool.homogeneous, 0)
        conn = pool.acquire()
        self.__verify_connection(pool.acquire(), test_env.get_main_user())
        conn.close()
        conn = pool.acquire(test_env.get_main_user(),
                            test_env.get_main_password())
        self.__verify_connection(conn, test_env.get_main_user())
        conn.close()
        conn = pool.acquire(test_env.get_proxy_user(),
                            test_env.get_proxy_password())
        self.__verify_connection(conn, test_env.get_proxy_user())
        conn.close()
        user_str = "%s[%s]" % \
                   (test_env.get_main_user(), test_env.get_proxy_user())
        conn = pool.acquire(user_str, test_env.get_main_password())
        self.__verify_connection(conn, test_env.get_proxy_user(),
                                 test_env.get_main_user())
        conn.close()

    @unittest.skipIf(test_env.get_is_thin(),
                     "thin mode doesn't support proxy users yet")
    def test_2408_heterogenous_without_user(self):
        "2408 - test heterogeneous pool without user and password specified"
        pool = test_env.get_pool(user="", password="", min=2, max=8,
                                 increment=3,
                                 getmode=oracledb.POOL_GETMODE_WAIT,
                                 homogeneous=False)
        conn = pool.acquire(test_env.get_main_user(),
                            test_env.get_main_password())
        self.__verify_connection(conn, test_env.get_main_user())
        conn.close()
        conn = pool.acquire(test_env.get_proxy_user(),
                            test_env.get_proxy_password())
        self.__verify_connection(conn, test_env.get_proxy_user())
        conn.close()
        user_str = "%s[%s]" % \
                   (test_env.get_main_user(), test_env.get_proxy_user())
        conn = pool.acquire(user_str, test_env.get_main_password())
        self.__verify_connection(conn, test_env.get_proxy_user(),
                                 test_env.get_main_user())

    @unittest.skipIf(test_env.get_is_thin(),
                     "thin mode doesn't support proxy users yet")
    def test_2409_heterogeneous_wrong_password(self):
        "2409 - test heterogeneous pool with wrong password specified"
        pool = test_env.get_pool(min=2, max=8, increment=3,
                                 getmode=oracledb.POOL_GETMODE_WAIT,
                                 homogeneous=False)
        self.assertRaisesRegex(oracledb.DatabaseError, "^ORA-01017:",
                               pool.acquire, test_env.get_proxy_user(),
                               "this is the wrong password")

    @unittest.skipIf(test_env.get_is_thin(),
                     "thin mode doesn't support tagging yet")
    def test_2410_tagging_session(self):
        "2410 - test tagging a session"
        pool = test_env.get_pool(min=2, max=8, increment=3,
                                 getmode=oracledb.POOL_GETMODE_NOWAIT)

        tag_mst = "TIME_ZONE=MST"
        tag_utc = "TIME_ZONE=UTC"

        conn = pool.acquire()
        self.assertEqual(conn.tag, None)
        pool.release(conn, tag=tag_mst)

        conn = pool.acquire()
        self.assertEqual(conn.tag, None)
        conn.tag = tag_utc
        conn.close()

        conn = pool.acquire(tag=tag_mst)
        self.assertEqual(conn.tag, tag_mst)
        conn.close()

        conn = pool.acquire(tag=tag_utc)
        self.assertEqual(conn.tag, tag_utc)
        conn.close()

    @unittest.skipIf(test_env.get_is_thin(),
                     "thin mode doesn't support session callbacks yet")
    def test_2411_plsql_session_callbacks(self):
        "2411 - test PL/SQL session callbacks"
        if test_env.get_client_version() < (12, 2):
            self.skipTest("PL/SQL session callbacks not supported before 12.2")
        callback = "pkg_SessionCallback.TheCallback"
        pool = test_env.get_pool(min=2, max=8, increment=3,
                                 getmode=oracledb.POOL_GETMODE_NOWAIT,
                                 session_callback=callback)
        tags = [
            "NLS_DATE_FORMAT=SIMPLE",
            "NLS_DATE_FORMAT=FULL;TIME_ZONE=UTC",
            "NLS_DATE_FORMAT=FULL;TIME_ZONE=MST"
        ]
        actual_tags = [None, None, "NLS_DATE_FORMAT=FULL;TIME_ZONE=UTC"]

        # truncate PL/SQL session callback log
        conn = pool.acquire()
        cursor = conn.cursor()
        cursor.execute("truncate table PLSQLSessionCallbacks")
        conn.close()

        # request sessions with each of the first two tags
        for tag in tags[:2]:
            conn = pool.acquire(tag=tag)
            conn.close()

        # for the last tag, use the matchanytag flag
        conn = pool.acquire(tag=tags[2], matchanytag=True)
        conn.close()

        # verify the PL/SQL session callback log is accurate
        conn = pool.acquire()
        cursor = conn.cursor()
        cursor.execute("""
                select RequestedTag, ActualTag
                from PLSQLSessionCallbacks
                order by FixupTimestamp""")
        results = cursor.fetchall()
        expected_results = list(zip(tags, actual_tags))
        self.assertEqual(results, expected_results)
        conn.close()

    @unittest.skipIf(test_env.get_is_thin(),
                     "thin mode doesn't support tagging yet")
    def test_2412_tagging_invalid_key(self):
        "2412 - testTagging with Invalid key"
        pool = test_env.get_pool(getmode=oracledb.POOL_GETMODE_NOWAIT)
        conn = pool.acquire()
        self.assertRaises(TypeError, pool.release, conn, tag=12345)
        if test_env.get_client_version() >= (12, 2):
            self.assertRaisesRegex(oracledb.DatabaseError, "^ORA-24488:",
                                   pool.release, conn, tag="INVALID_TAG")

    def test_2413_close_and_drop_connection_from_pool(self):
        "2413 - test dropping/closing a connection from the pool"
        pool = test_env.get_pool(min=1, max=8, increment=1,
                                 getmode=oracledb.POOL_GETMODE_WAIT)
        conn = pool.acquire()
        self.assertEqual(pool.busy, 1, "busy (1)")
        self.assertEqual(pool.opened, 1, "opened (1)")
        pool.drop(conn)
        self.assertEqual(pool.busy, 0, "busy (2)")
        self.assertEqual(pool.opened, 0, "opened (2)")
        conn = pool.acquire()
        self.assertEqual(pool.busy, 1, "busy (3)")
        self.assertEqual(pool.opened, 1, "opened (3)")
        conn.close()
        self.assertEqual(pool.busy, 0, "busy (4)")
        self.assertEqual(pool.opened, 1, "opened (4)")

    def test_2414_create_new_pure_connection(self):
        "2414 - test to ensure pure connections are being created correctly"
        pool = test_env.get_pool(min=1, max=2, increment=1,
                                 getmode=oracledb.POOL_GETMODE_WAIT)
        connection_1 = pool.acquire()
        connection_2 = pool.acquire()
        self.assertEqual(pool.opened, 2, "opened (1)")
        pool.release(connection_1)
        pool.release(connection_2)
        connection_3 = pool.acquire(purity=oracledb.ATTR_PURITY_NEW)
        self.assertEqual(pool.opened, 2, "opened (2)")
        pool.release(connection_3)

    @unittest.skipIf(test_env.get_is_thin(),
                     "thin mode doesn't support pool reconfigure yet")
    def test_2415_reconfigure_pool(self):
        "2415 - test to ensure reconfigure() updates pool properties"
        pool = test_env.get_pool(min=1, max=2, increment=1,
                                 getmode=oracledb.POOL_GETMODE_WAIT)
        self.assertEqual(pool.min, 1, "min (1)")
        self.assertEqual(pool.max, 2, "max (2)")
        self.assertEqual(pool.increment, 1, "increment (1)")
        self.assertEqual(pool.getmode, oracledb.POOL_GETMODE_WAIT,
                         "getmode differs")
        self.assertEqual(pool.timeout, 0, "timeout (0)")
        self.assertEqual(pool.wait_timeout, 5000, "wait_timeout (5000)")
        self.assertEqual(pool.max_lifetime_session, 0,
                         "max_lifetime_sessionmeout (0)")
        self.assertEqual(pool.max_sessions_per_shard, 0,
                         "max_sessions_per_shard (0)")
        self.assertEqual(pool.stmtcachesize, 20, "stmtcachesize (20)")
        self.assertEqual(pool.ping_interval, 60, "ping_interval (60)")

        pool.reconfigure(min=2, max=5, increment=2, timeout=30,
                         getmode=oracledb.POOL_GETMODE_TIMEDWAIT,
                         wait_timeout=3000, max_lifetime_session=20,
                         max_sessions_per_shard=2, stmtcachesize=30,
                         ping_interval=30)
        self.assertEqual(pool.min, 2, "min (2)")
        self.assertEqual(pool.max, 5, "max (5)")
        self.assertEqual(pool.increment, 2, "increment (2)")
        self.assertEqual(pool.getmode, oracledb.POOL_GETMODE_TIMEDWAIT,
                         "getmode differs")
        self.assertEqual(pool.timeout, 30, "timeout (30)")
        self.assertEqual(pool.wait_timeout, 3000, "wait_timeout (3000)")
        self.assertEqual(pool.max_lifetime_session, 20,
                         "max_lifetime_sessionmeout (20)")
        self.assertEqual(pool.max_sessions_per_shard, 2,
                         "max_sessions_per_shard (2)")
        self.assertEqual(pool.stmtcachesize, 30, "stmtcachesize (30)")
        self.assertEqual(pool.ping_interval, 30, "ping_interval (30)")

    @unittest.skipIf(test_env.get_is_thin(),
                     "thin mode doesn't support all the pool params yet")
    def test_2416_test_reconfigure_pool_with_missing_values(self):
        "2416 - test the reconfigure values are changed and rest unchanged"
        self.__perform_reconfigure_test("min", 5)
        self.__perform_reconfigure_test("max", 20)
        self.__perform_reconfigure_test("increment", 5)
        self.__perform_reconfigure_test("timeout", 10)
        self.__perform_reconfigure_test("wait_timeout", 8000)
        self.__perform_reconfigure_test("stmtcachesize", 40)
        self.__perform_reconfigure_test("max_lifetime_session", 2000)
        self.__perform_reconfigure_test("max_sessions_per_shard", 5)
        self.__perform_reconfigure_test("ping_interval", 50)
        self.__perform_reconfigure_test("getmode",
                                        oracledb.POOL_GETMODE_NOWAIT)

    @unittest.skipIf(test_env.get_is_thin(),
                     "thin mode doesn't support all the pool params yet")
    def test_2417_setting_each_pool_param(self):
        "2417 - test to see if specified parameters are set during creation"
        pool = test_env.get_pool(min=1, max=2, increment=1, timeout=10,
                                 wait_timeout=10, max_lifetime_session=20,
                                 max_sessions_per_shard=1, stmtcachesize=25,
                                 ping_interval=25,
                                 getmode=oracledb.POOL_GETMODE_WAIT)
        self.assertEqual(pool.min, 1, "min (1)")
        self.assertEqual(pool.max, 2, "max (2)")
        self.assertEqual(pool.increment, 1, "increment (1)")
        self.assertEqual(pool.getmode, oracledb.POOL_GETMODE_WAIT,
                         "getmode differs")
        self.assertEqual(pool.timeout, 10, "timeout (10)")
        self.assertEqual(pool.wait_timeout, 10, "wait_timeout (10)")
        self.assertEqual(pool.max_lifetime_session, 20,
                         "max_lifetime_sessionmeout (20)")
        self.assertEqual(pool.max_sessions_per_shard, 1,
                         "max_sessions_per_shard (1)")
        self.assertEqual(pool.stmtcachesize, 25, "stmtcachesize (25)")
        self.assertEqual(pool.ping_interval, 25, "ping_interval (25)")

    def test_2418_deprecations(self):
        "2418 - test to verify deprecations"
        callback = "pkg_SessionCallback.TheCallback"
        self.assertRaisesRegex(oracledb.ProgrammingError, "^DPY-2014:",
                               test_env.get_pool, min=1, max=2, increment=1,
                               wait_timeout=10, waitTimeout=10)
        self.assertRaisesRegex(oracledb.ProgrammingError, "^DPY-2014:",
                               test_env.get_pool, min=1, max=2, increment=1,
                               max_lifetime_session=20, maxLifetimeSession=20)
        self.assertRaisesRegex(oracledb.ProgrammingError, "^DPY-2014:",
                               test_env.get_pool, min=1, max=2, increment=1,
                               max_sessions_per_shard=1, maxSessionsPerShard=1)
        self.assertRaisesRegex(oracledb.ProgrammingError, "^DPY-2014:",
                               test_env.get_pool, min=2, max=8, increment=3,
                               getmode=oracledb.POOL_GETMODE_NOWAIT,
                               session_callback=callback,
                               sessionCallback=callback)

    def test_2419_statement_cache_size(self):
        "2419 - test to verify statement cache size is retained"
        pool = test_env.get_pool(min=1, max=2, increment=1,
                                 getmode=oracledb.POOL_GETMODE_WAIT,
                                 stmtcachesize=25)
        self.assertEqual(pool.stmtcachesize, 25, "stmtcachesize (25)")
        pool.stmtcachesize = 35
        self.assertEqual(pool.stmtcachesize, 35, "stmtcachesize (35)")

    @unittest.skipIf(test_env.get_is_thin(),
                     "thin mode doesn't support tagging yet")
    def test_2420_callable_session_callbacks(self):
        "2420 - test that session callbacks are being called correctly"
        pool = test_env.get_pool(min=2, max=5, increment=1,
                        session_callback=self.__callable_session_callback)

        # new connection with a tag should invoke the session callback
        with pool.acquire(tag="NLS_DATE_FORMAT=SIMPLE") as conn:
            cursor = conn.cursor()
            cursor.execute("select to_char(2021-05-20) from dual")
            result, = cursor.fetchone()
            self.assertEqual(self.session_called, True)

        # acquiring a connection with the same tag should not invoke the
        # session callback
        self.session_called = False
        with pool.acquire(tag="NLS_DATE_FORMAT=SIMPLE") as conn:
            cursor = conn.cursor()
            cursor.execute("select to_char(2021-05-20) from dual")
            result, = cursor.fetchone()
            self.assertEqual(self.session_called, False)

        # acquiring a connection with a new tag should invoke the session
        # callback
        self.session_called = False
        with pool.acquire(tag="NLS_DATE_FORMAT=FULL;TIME_ZONE=UTC") as conn:
            cursor = conn.cursor()
            cursor.execute("select to_char(current_date) from dual")
            result, = cursor.fetchone()
            self.assertEqual(self.session_called, True)

        # acquiring a connection with a new tag and specifying that a
        # connection with any tag can be acquired should invoke the session
        # callback
        self.session_called = False
        with pool.acquire(tag="NLS_DATE_FORMAT=FULL;TIME_ZONE=MST", \
                          matchanytag=True) as conn:
            cursor = conn.cursor()
            cursor.execute("select to_char(current_date) from dual")
            result, = cursor.fetchone()
            self.assertEqual(self.session_called, True)

        # new connection with no tag should invoke the session callback
        self.session_called = False
        with pool.acquire() as conn:
            cursor = conn.cursor()
            cursor.execute("select to_char(current_date) from dual")
            result, = cursor.fetchone()
            self.assertEqual(self.session_called, True)

    def test_2421_pool_close_normal_no_connections(self):
        "2421 - test closing a pool normally with no connections checked out"
        pool = test_env.get_pool(min=1, max=8, increment=1,
                                 getmode=oracledb.POOL_GETMODE_WAIT)
        pool.close()

    def test_2422_pool_close_normal_with_connections(self):
        "2422 - test closing a pool normally with connections checked out"
        pool = test_env.get_pool(min=1, max=8, increment=1,
                                 getmode=oracledb.POOL_GETMODE_WAIT)
        conn = pool.acquire()
        self.assertRaisesRegex(oracledb.InterfaceError, "^DPY-1005:",
                               pool.close)

    def test_2423_pool_close_force(self):
        "2423 - test closing a pool forcibly"
        pool = test_env.get_pool(min=1, max=8, increment=1,
                                 getmode=oracledb.POOL_GETMODE_WAIT)
        conn = pool.acquire()
        pool.close(force=True)

    def test_2424_exception_on_acquire_after_pool_closed(self):
        "2424 - using the pool after it is closed raises an exception"
        pool = test_env.get_pool(min=1, max=8, increment=1,
                                 getmode=oracledb.POOL_GETMODE_WAIT)
        pool.close()
        self.assertRaisesRegex(oracledb.InterfaceError, "^DPY-1002:",
                               pool.acquire)

    def test_2425_pool_with_no_connections(self):
        "2425 - using the pool beyond max limit raises an error"
        pool = test_env.get_pool(min=1, max=2, increment=1,
                                 getmode=oracledb.POOL_GETMODE_WAIT)
        connection_1 = pool.acquire()
        connection_2 = pool.acquire()
        pool.getmode = oracledb.POOL_GETMODE_NOWAIT
        self.assertRaisesRegex(oracledb.DatabaseError, "^DPY-4005:",
                               pool.acquire)

    def test_2426_session_callback_for_new_connections(self):
        "2426 - callable session callback is executed for new connections"
        class Counter:
            num_calls = 0
            @classmethod
            def session_callback(cls, conn, requested_tag):
                cls.num_calls += 1
        pool = test_env.get_pool(min=1, max=2, increment=1,
                                 session_callback=Counter.session_callback)
        with pool.acquire() as conn1:
            with pool.acquire() as conn2:
                pass
        with pool.acquire() as conn1:
            with pool.acquire() as conn2:
                pass
        self.assertEqual(Counter.num_calls, 2)

    def test_2427_drop_dead_connection_from_pool(self):
        "2427 - drop the pooled connection on receiving dead connection error"
        admin_conn = test_env.get_admin_connection()
        pool = test_env.get_pool(min=2, max=2, increment=2)

        # acquire connections from the pool and kill all the sessions
        with admin_conn.cursor() as admin_cursor:
            for conn in [pool.acquire() for i in range(2)]:
                with conn.cursor() as cursor:
                    cursor.execute("""
                        select
                            dbms_debug_jdwp.current_session_id,
                            dbms_debug_jdwp.current_session_serial
                        from dual""")
                    sid, serial = cursor.fetchone()
                    sql = f"alter system kill session '{sid},{serial}'"
                    admin_cursor.execute(sql)
                conn.close()
        self.assertEqual(pool.opened, 2)

        # when try to re-use the killed sessions error will be raised;
        # release all such connections
        for conn in [pool.acquire() for i in range(2)]:
            with conn.cursor() as cursor:
                self.assertRaisesRegex(oracledb.DatabaseError, "^DPY-4011:",
                                       cursor.execute, "select user from dual")
            conn.close()
        self.assertEqual(pool.opened, 0)

        # if a free connection is available, it can be used; otherwise a new
        # connection will be created
        for conn in [pool.acquire() for i in range(2)]:
            with conn.cursor() as cursor:
                cursor.execute("select user from dual")
                user, = cursor.fetchone()
                self.assertEqual(user, test_env.get_main_user().upper())
            conn.close()
        self.assertEqual(pool.opened, 2)

    def test_2428_acquire_conn_from_empty_pool(self):
        "2428 - acquire a connection from an empty pool (min=0)"
        pool = test_env.get_pool(min=0, max=2, increment=2)
        with pool.acquire() as conn:
            with conn.cursor() as cursor:
                cursor.execute("select user from dual")
                result, = cursor.fetchone()
                self.assertEqual(result, test_env.get_main_user().upper())

    @unittest.skipIf(test_env.get_is_thin(),
                     "thin mode doesn't support soda_metadata_cache" \
                     "parameter yet")
    def test_2429_soda_metadata_cache(self):
        "2429 - test soda_metadata_cache parameter"
        self.get_soda_database(minclient=(19, 11))
        pool = test_env.get_pool(min=0, max=2, increment=2)
        self.assertEqual(pool.soda_metadata_cache, False)
        pool = test_env.get_pool(min=0, max=2, increment=2,
                                 soda_metadata_cache=True)
        self.assertEqual(pool.soda_metadata_cache, True)
        pool.soda_metadata_cache = False
        self.assertEqual(pool.soda_metadata_cache, False)

    def test_2430_get_different_types_from_pooled_connections(self):
        "2430 - get different object types from different connections"
        pool = test_env.get_pool(min=1, max=2, increment=1)
        with pool.acquire() as conn:
            typ = conn.gettype("UDT_SUBOBJECT")
            self.assertEqual(typ.name, "UDT_SUBOBJECT")
        with pool.acquire() as conn:
            typ = conn.gettype("UDT_OBJECTARRAY")
            self.assertEqual(typ.name, "UDT_OBJECTARRAY")

    def test_2431_proxy_user_in_create(self):
        "2431 - test creating a pool using a proxy user"
        user_str = "%s[%s]" % \
                (test_env.get_main_user(), test_env.get_proxy_user())
        pool = test_env.get_pool(user=user_str)
        self.__verify_connection(pool.acquire(), test_env.get_proxy_user(),
                                 test_env.get_main_user())

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