test! significantly increase test coverage

Author: picnixz

Lib/test/test_sqlite3/test_dbapi.py

@@ -30,7 +30,9 @@
 import unittest
 import urllib.parse
 import warnings
+from collections import Counter
 
+from test import support
 from test.support import (
     SHORT_TIMEOUT, check_disallow_instantiation, requires_subprocess, subTests
 )
@@ -1747,18 +1749,53 @@ def ParamsCxCloseInNext(cx):
 class ExecutionConcurrentlyCloseConnectionBaseTests:
     """Tests when execute() and executemany() concurrently close the connection."""
 
-    def inittest(self):
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        cls.sqlite = import_helper.import_fresh_module("sqlite3", fresh=["_sqlite3"])
+
+    def inittest(self, **adapters):
         """Return a pair (connection, connection or cursor) to use in tests."""
-        cx = sqlite.connect(":memory:")
-        cx.execute("create table tmp(a number)")
+        # Counter for the number of calls to the tracked functions.
+        self.ncalls = Counter()
+        self.colname = "a"
+
+        cx = self.sqlite.connect(":memory:")
+        # table to use to query the database to ensure that it's not closed
+        cx.execute(f"create table canary({self.colname} nunmber)")
+        cx.execute(f"insert into canary({self.colname}) values (?)", (1,))
+        cx.execute(f"create table tmp({self.colname} number)")
         self.addCleanup(cx.close)
-        return cx, self.interface(cx)
+        return cx, self.executor(cx)
 
-    def interface(self, connection):
+    def executor(self, connection):
         """Return a cursor-like interface from a given SQLite3 connection."""
         raise NotImplementedError
 
-    def test_execute_use___getitem__(self):
+    def check_alive(self, executor):
+        # check that the connection is alive by making a dummy query
+        res = executor.execute("SELECT * from canary").fetchall()
+        self.assertEqual(res, [(1,)])
+
+    def check_execute(self, executor, payload, *, named=False):
+        self.assertEqual(self.ncalls.total(), 0)
+        self.check_alive(executor)
+        msg = r"Cannot operate on a closed database\."
+        binding = "(:a)" if named else "(?)"
+        with self.assertRaisesRegex(self.sqlite.ProgrammingError, msg):
+            executor.execute(f"insert into tmp(a) values {binding}", payload)
+
+    def check_executemany(self, executor, payload, *, named=False):
+        self.assertEqual(self.ncalls.total(), 0)
+        self.check_alive(executor)
+        msg = r"Cannot operate on a closed database\."
+        binding = "(:a)" if named else "(?)"
+        with self.assertRaisesRegex(self.sqlite.ProgrammingError, msg):
+            executor.executemany(f"insert into tmp(a) values {binding}", payload)
+
+    # Simple tests
+
+    def test_execute(self):
         # Prevent SIGSEGV when closing the connection while binding parameters.
         #
         # Internally, the connection's state is checked after bind_parameters().
@@ -1769,33 +1806,31 @@ def test_execute_use___getitem__(self):
         # Regression test for https://github.com/python/cpython/issues/143198.
 
         class PT:
-            def __getitem__(self, i):
+            def __getitem__(_, i):
+                self.ncalls[None] += 1
                 cx.close()
                 return 1
             def __len__(self):
                 return 1
 
-        cx, cu = self.inittest()
-        msg = r"Cannot operate on a closed database\."
-        with self.assertRaisesRegex(sqlite.ProgrammingError, msg):
-            cu.execute("insert into tmp(a) values (?)", PT())
+        cx, ex = self.inittest()
+        self.check_execute(ex, PT())
+        self.assertEqual(self.ncalls[None], 1)
 
     @subTests("params_factory", (ParamsCxCloseInIterMany, ParamsCxCloseInNext))
-    def test_executemany_use___iter__(self, params_factory):
+    def test_executemany_iterator(self, params_factory):
         # Prevent SIGSEGV with iterable of parameters closing the connection.
         # Regression test for https://github.com/python/cpython/issues/143198.
-        cx, cu = self.inittest()
-        msg = r"Cannot operate on a closed database\."
-        with self.assertRaisesRegex(sqlite.ProgrammingError, msg):
-            cu.executemany("insert into tmp(a) values (?)", params_factory(cx))
+        cx, ex = self.inittest()
+        self.check_executemany(ex, params_factory(cx))
 
     # The test constructs an iterable of parameters of length 'n'
     # and the connection is closed when we access the j-th one.
     # The iterable is of type 'map' but the test wraps that map
     # with 'iterable_wrapper' to exercise internals.
     @subTests(("j", "n"), ([0, 1], [0, 3], [1, 3], [2, 3]))
     @subTests("iterable_wrapper", (list, lambda x: x, lambda x: iter(x)))
-    def test_executemany_use___getitem__(self, j, n, iterable_wrapper):
+    def test_executemany_iterable(self, j, n, iterable_wrapper):
         # Prevent SIGSEGV when closing the connection while binding parameters.
         #
         # Internally, the connection's state is checked after bind_parameters().
@@ -1805,30 +1840,157 @@ def test_executemany_use___getitem__(self, j, n, iterable_wrapper):
         #
         # Regression test for https://github.com/python/cpython/issues/143198.
 
-        cx, cu = self.inittest()
-
         class PT:
+            case = self
             def __init__(self, value):
                 self.value = value
             def __getitem__(self, i):
                 if self.value == j:
+                    self.case.ncalls[None] = j
                     cx.close()
                 return self.value
             def __len__(self):
                 return 1
 
-        msg = r"Cannot operate on a closed database\."
+        cx, ex = self.inittest()
         items = iterable_wrapper(map(PT, range(n)))
-        with self.assertRaisesRegex(sqlite.ProgrammingError, msg):
-            cu.executemany("insert into tmp(a) values (?)", items)
+        self.check_executemany(ex, items)
+        self.assertEqual(self.ncalls[None], j)
+
+    # Tests when the SQL parameters are given as a sequence.
+
+    def test_invalid_params_sequence_size(self):
+        class I:
+            def __index__(_):
+                self.ncalls["I.__index__"] += 1
+                cx.close()
+                return 1
+
+        class S:  # emulate a non-native sequence object
+            def __getitem__(self):
+                raise RuntimeError("must not be called")
+            def __len__(_):
+                self.ncalls["S.__len__"] += 1
+                return I()
+
+        cx, ex = self.inittest()
+        self.check_execute(ex, S())
+        self.assertEqual(self.ncalls["S.__len__"], 1)
+        self.assertEqual(self.ncalls["I.__index__"], 1)
+
+        cx, ex = self.inittest()
+        self.check_executemany(ex, [S()])
+        self.assertEqual(self.ncalls["S.__len__"], 1)
+        self.assertEqual(self.ncalls["I.__index__"], 1)
+
+    def test_invalid_params_sequence_item(self):
+        class S:  # emulate a non-native sequence object
+            def __getitem__(_, i):
+                self.ncalls[None] += 1
+                cx.close()
+                return 1
+            def __len__(self):
+                return 1
+
+        cx, ex = self.inittest()
+        self.check_execute(ex, S())
+        self.assertEqual(self.ncalls[None], 1)
+
+        cx, ex = self.inittest()
+        self.check_executemany(ex, [S()])
+        self.assertEqual(self.ncalls[None], 1)
+
+    def test_invalid_params_sequence_item_close_in_adapter(self):
+        class S: pass
+        def adapter(s):
+            self.ncalls[None] += 1
+            cx.close()
+            return 1
+
+        self.sqlite.register_adapter(S, adapter)
+
+        cx, ex = self.inittest()
+        self.check_execute(ex, [S()])
+        self.assertEqual(self.ncalls[None], 1)
+
+        cx, ex = self.inittest()
+        self.check_executemany(ex, [[S()]])
+        self.assertEqual(self.ncalls[None], 1)
+
+    def test_invalid_params_sequence_item_close_after_adapted(self):
+        class B(bytearray):
+            def __buffer__(_, flags):
+                self.ncalls[None] += 1
+                cx.close()
+                return super().__buffer__(flags)
+
+        cx, ex = self.inittest()
+        self.check_execute(ex, [B()])
+        self.assertEqual(self.ncalls[None], 1)
+
+        cx, ex = self.inittest()
+        self.check_executemany(ex, [[B()]])
+        self.assertEqual(self.ncalls[None], 1)
+
+    # Tests when the SQL parameters are given as a mapping.
+
+    def test_invalid_params_mapping_item(self):
+        class S(dict):
+            def __getitem__(_, key):
+                self.assertEqual(key, self.colname)
+                self.ncalls[self.colname] += 1
+                cx.close()
+                return 1
+            def __len__(self):
+                return 1
+
+        cx, ex = self.inittest()
+        self.check_execute(ex, S(), named=True)
+        self.assertEqual(self.ncalls[self.colname], 1)
+
+        cx, ex = self.inittest()
+        self.check_executemany(ex, [S()], named=True)
+        self.assertEqual(self.ncalls[self.colname], 1)
+
+    def test_invalid_params_mapping_item_close_in_adapter(self):
+        class S: pass
+        def adapter(s):
+            self.ncalls[None] += 1
+            cx.close()
+            return 1
+
+        self.sqlite.register_adapter(S, adapter)
+
+        cx, ex = self.inittest()
+        self.check_execute(ex, {self.colname: S()}, named=True)
+        self.assertEqual(self.ncalls[None], 1)
+
+        cx, ex = self.inittest()
+        self.check_executemany(ex, [{self.colname: S()}], named=True)
+        self.assertEqual(self.ncalls[None], 1)
+
+    def test_invalid_params_mapping_item_close_after_adapted(self):
+        class B(bytearray):
+            def __buffer__(_, flags):
+                self.ncalls[None] += 1
+                cx.close()
+                return super().__buffer__(flags)
+
+        cx, ex = self.inittest()
+        self.check_execute(ex, {self.colname: B()}, named=True)
+        self.assertEqual(self.ncalls[None], 1)
+
+        cx, ex = self.inittest()
+        self.check_executemany(ex, [{self.colname: B()}], named=True)
+        self.assertEqual(self.ncalls[None], 1)
 
 
 class ConnectionExecutionConcurrentlyCloseConnectionTests(
     ExecutionConcurrentlyCloseConnectionBaseTests,
     unittest.TestCase,
 ):
     """Regression tests for conn.execute() and conn.executemany()."""
-    def interface(self, connection):
+    def executor(self, connection):
         return connection
 
 
@@ -1837,7 +1999,7 @@ class CursorExecutionConcurrentlyCloseConnectionTests(
     unittest.TestCase,
 ):
     """Regression tests for cursor.execute() and cursor.executemany()."""
-    def interface(self, connection):
+    def executor(self, connection):
         return connection.cursor()