Added more tests.

Author: anthony-tuininga

tests/test_1100_connection.py

@@ -890,6 +890,31 @@ def test_1152(self):
         (fetched_value,) = cursor.fetchone()
         self.assertEqual(conn.serial_num, fetched_value)
 
+    @unittest.skipUnless(
+        test_env.get_is_thin(),
+        "thick mode doesn't support registered protocols",
+    )
+    def test_1153(self):
+        "1153 - test passed params in hook with standalone connection"
+        sdu = 4096
+        params = test_env.get_connect_params()
+        protocol = "proto-test"
+        orig_connect_string = test_env.get_connect_string()
+        connect_string = f"{protocol}://{orig_connect_string}"
+
+        def hook(passed_protocol, passed_protocol_arg, passed_params):
+            self.assertEqual(passed_protocol, protocol)
+            self.assertEqual(passed_protocol_arg, orig_connect_string)
+            passed_params.parse_connect_string(passed_protocol_arg)
+            passed_params.set(sdu=sdu)
+
+        try:
+            oracledb.register_protocol(protocol, hook)
+            oracledb.connect(dsn=connect_string, params=params)
+            self.assertEqual(params.sdu, sdu)
+        finally:
+            oracledb.register_protocol(protocol, None)
+
 
 if __name__ == "__main__":
     test_env.run_test_cases()

tests/test_2400_pool.py

@@ -951,6 +951,34 @@ def test_2446(self):
         )
         self.__verify_create_arg("driver_name", "newdriver", sql)
 
+    @unittest.skipUnless(
+        test_env.get_is_thin(),
+        "thick mode doesn't support registered protocols",
+    )
+    def test_2447(self):
+        "2447 - test register_parameter with pooled connection"
+        sdu = 4096
+        params = test_env.get_pool_params()
+        protocol = "proto-test"
+        orig_connect_string = test_env.get_connect_string()
+        connect_string = f"{protocol}://{orig_connect_string}"
+
+        def hook(passed_protocol, passed_protocol_arg, passed_params):
+            self.assertEqual(passed_protocol, protocol)
+            self.assertEqual(passed_protocol_arg, orig_connect_string)
+            passed_params.parse_connect_string(passed_protocol_arg)
+            passed_params.set(sdu=sdu)
+
+        try:
+            oracledb.register_protocol(protocol, hook)
+            pool = oracledb.create_pool(dsn=connect_string, params=params)
+            self.assertEqual(params.sdu, sdu)
+            with pool.acquire():
+                pass
+            pool.close()
+        finally:
+            oracledb.register_protocol(protocol, None)
+
 
 if __name__ == "__main__":
     test_env.run_test_cases()

tests/test_5300_connection_async.py

@@ -669,6 +669,27 @@ async def test_5352(self):
             (fetched_value,) = await cursor.fetchone()
             self.assertEqual(conn.serial_num, fetched_value)
 
+    async def test_5353(self):
+        "5353 - test passed params in hook with standalone connection"
+        sdu = 4096
+        params = test_env.get_connect_params()
+        protocol = "proto-test"
+        orig_connect_string = test_env.get_connect_string()
+        connect_string = f"{protocol}://{orig_connect_string}"
+
+        def hook(passed_protocol, passed_protocol_arg, passed_params):
+            self.assertEqual(passed_protocol, protocol)
+            self.assertEqual(passed_protocol_arg, orig_connect_string)
+            passed_params.parse_connect_string(passed_protocol_arg)
+            passed_params.set(sdu=sdu)
+
+        try:
+            oracledb.register_protocol(protocol, hook)
+            await oracledb.connect_async(dsn=connect_string, params=params)
+            self.assertEqual(params.sdu, sdu)
+        finally:
+            oracledb.register_protocol(protocol, None)
+
 
 if __name__ == "__main__":
     test_env.run_test_cases()

tests/test_5500_pool_async.py

@@ -555,6 +555,32 @@ async def test_5534(self):
         )
         await self.__verify_create_arg("driver_name", "newdriver", sql)
 
+    async def test_5535(self):
+        "5535 - test register_parameter with pooled connection"
+        sdu = 4096
+        params = test_env.get_pool_params()
+        protocol = "proto-test"
+        orig_connect_string = test_env.get_connect_string()
+        connect_string = f"{protocol}://{orig_connect_string}"
+
+        def hook(passed_protocol, passed_protocol_arg, passed_params):
+            self.assertEqual(passed_protocol, protocol)
+            self.assertEqual(passed_protocol_arg, orig_connect_string)
+            passed_params.parse_connect_string(passed_protocol_arg)
+            passed_params.set(sdu=sdu)
+
+        try:
+            oracledb.register_protocol(protocol, hook)
+            pool = oracledb.create_pool_async(
+                dsn=connect_string, params=params
+            )
+            self.assertEqual(params.sdu, sdu)
+            async with pool.acquire():
+                pass
+            await pool.close()
+        finally:
+            oracledb.register_protocol(protocol, None)
+
 
 if __name__ == "__main__":
     test_env.run_test_cases()