Improved test suite and checks.

Author: anthony-tuininga

src/oracledb/pool.py

@@ -1458,6 +1458,8 @@ def add_pool(self, alias, pool):
         Adds a pool to the cache. An exception is raised if a pool is already
         cached with the given alias.
         """
+        if not isinstance(alias, str):
+            raise TypeError("pool_alias must be a string")
         with self.lock:
             if alias in self.pools:
                 errors._raise_err(errors.ERR_NAMED_POOL_EXISTS, alias=alias)

tests/test_2400_pool.py

@@ -987,35 +987,61 @@ def test_2448(self):
         self.assertEqual(conn.edition, edition)
 
     def test_2449(self):
-        "2449 - test create_pool() and get_pool() with name"
-        name = "pool_alias_2449"
-        pool = test_env.get_pool(pool_alias=name)
-        self.assertIs(pool, oracledb.get_pool(name))
+        "2449 - test create_pool() and get_pool() with alias"
+        alias = "pool_alias_2449"
+        pool = test_env.get_pool(pool_alias=alias)
+        self.assertIs(pool, oracledb.get_pool(alias))
         pool.close()
 
     def test_2450(self):
-        "2450 - test create_pool() twice with the same name"
-        name = "pool_alias_2450"
-        pool = test_env.get_pool(pool_alias=name)
+        "2450 - test create_pool() twice with the same alias"
+        alias = "pool_alias_2450"
+        pool = test_env.get_pool(pool_alias=alias)
         with self.assertRaisesFullCode("DPY-2055"):
-            test_env.get_pool(pool_alias=name)
+            test_env.get_pool(pool_alias=alias)
         pool.close()
-        self.assertIsNone(oracledb.get_pool(name))
+        self.assertIsNone(oracledb.get_pool(alias))
 
     def test_2451(self):
-        "2451 - test connect() with pool name"
-        name = "pool_alias_2451"
-        pool = test_env.get_pool(pool_alias=name)
+        "2451 - test connect() with pool alias"
+        alias = "pool_alias_2451"
+        pool = test_env.get_pool(pool_alias=alias)
         with self.assertRaisesFullCode("DPY-2014"):
-            test_env.get_connection(pool=pool, pool_alias=name)
-        with oracledb.connect(pool_alias=name) as conn:
+            test_env.get_connection(pool=pool, pool_alias=alias)
+        with oracledb.connect(pool_alias=alias) as conn:
             with conn.cursor() as cursor:
                 cursor.execute("select user from dual")
                 (value,) = cursor.fetchone()
                 self.assertEqual(value, test_env.get_main_user().upper())
         pool.close()
         with self.assertRaisesFullCode("DPY-2054"):
-            oracledb.connect(pool_alias=name)
+            oracledb.connect(pool_alias=alias)
+
+    def test_2452(self):
+        "2452 - test acquire() with pool alias and stmtcachesize"
+        alias = "pool_2452"
+        stmtcachesize = 35
+        test_env.get_pool(pool_alias=alias, stmtcachesize=stmtcachesize)
+        with oracledb.connect(pool_alias=alias) as conn:
+            self.assertEqual(conn.stmtcachesize, stmtcachesize)
+        oracledb.get_pool(alias).close()
+
+    def test_2453(self):
+        "2453 - test pool alias is case sensitive"
+        alias = "pool_2458"
+        test_env.get_pool(pool_alias=alias)
+        self.assertIsNone(oracledb.get_pool(alias.upper()))
+        with self.assertRaisesFullCode("DPY-2054"):
+            test_env.get_connection(pool_alias=alias.upper())
+        oracledb.get_pool(alias).close()
+
+    def test_2454(self):
+        "2454 - test pool alias with invalid types"
+        aliases = [5, set(), dict(), bytearray(1)]
+        for alias in aliases:
+            with self.subTest(alias=alias):
+                with self.assertRaises(TypeError):
+                    test_env.get_pool(pool_alias=alias)
 
 
 if __name__ == "__main__":

tests/test_5500_pool_async.py

@@ -590,37 +590,46 @@ async def test_5536(self):
         await pool.close()
 
     async def test_5537(self):
-        "5537 - test create_pool() and get_pool() with name"
-        name = "pool_alias_5537"
-        pool = test_env.get_pool_async(pool_alias=name)
-        self.assertIs(pool, oracledb.get_pool(name))
+        "5537 - test create_pool() and get_pool() with alias"
+        alias = "pool_alias_5537"
+        pool = test_env.get_pool_async(pool_alias=alias)
+        self.assertIs(pool, oracledb.get_pool(alias))
         await pool.close()
 
     async def test_5538(self):
-        "5538 - test create_pool() twice with the same name"
-        name = "pool_alias_5538"
-        pool = test_env.get_pool_async(pool_alias=name)
+        "5538 - test create_pool() twice with the same alias"
+        alias = "pool_alias_5538"
+        pool = test_env.get_pool_async(pool_alias=alias)
         with self.assertRaisesFullCode("DPY-2055"):
-            test_env.get_pool_async(pool_alias=name)
+            test_env.get_pool_async(pool_alias=alias)
         await pool.close()
-        self.assertIsNone(oracledb.get_pool(name))
+        self.assertIsNone(oracledb.get_pool(alias))
 
     async def test_5539(self):
-        "5539 - test connect() with pool name"
-        name = "pool_alias_5539"
-        pool = test_env.get_pool_async(pool_alias=name)
-        try:
-            with self.assertRaisesFullCode("DPY-2014"):
-                test_env.get_connection_async(pool=pool, pool_alias=name)
-            async with oracledb.connect_async(pool_alias=name) as conn:
-                with conn.cursor() as cursor:
-                    await cursor.execute("select user from dual")
-                    (value,) = await cursor.fetchone()
-                    self.assertEqual(value, test_env.get_main_user().upper())
-        finally:
-            await pool.close(force=True)
+        "5539 - test acquire() with pool alias and stmtcachesize"
+        alias = "pool_5539"
+        stmtcachesize = 35
+        test_env.get_pool_async(pool_alias=alias, stmtcachesize=stmtcachesize)
+        async with oracledb.connect_async(pool_alias=alias) as conn:
+            self.assertEqual(conn.stmtcachesize, stmtcachesize)
+        await oracledb.get_pool(alias).close()
+
+    async def test_5540(self):
+        "5540 - test pool alias is case sensitive"
+        alias = "pool_5540"
+        test_env.get_pool_async(pool_alias=alias)
+        self.assertIsNone(oracledb.get_pool(alias.upper()))
         with self.assertRaisesFullCode("DPY-2054"):
-            await oracledb.connect_async(pool_alias=name)
+            await test_env.get_connection_async(pool_alias=alias.upper())
+        await oracledb.get_pool(alias).close()
+
+    async def test_5541(self):
+        "5541 - test pool alias with invalid types"
+        aliases = [5, set(), dict(), bytearray(1)]
+        for alias in aliases:
+            with self.subTest(alias=alias):
+                with self.assertRaises(TypeError):
+                    test_env.get_pool_async(pool_alias=alias)
 
 
 if __name__ == "__main__":

utils/templates/pool.py

@@ -864,6 +864,8 @@ def add_pool(self, alias, pool):
         Adds a pool to the cache. An exception is raised if a pool is already
         cached with the given alias.
         """
+        if not isinstance(alias, str):
+            raise TypeError("pool_alias must be a string")
         with self.lock:
             if alias in self.pools:
                 errors._raise_err(errors.ERR_NAMED_POOL_EXISTS, alias=alias)