Address review

- Move test utility functions to util.py - Use memory database mixin - Add check() helper for closed connection tests - Address other remarks by Nikita
python · erlend-aasland · Aug 17, 2023 · Aug 16, 2023 · Aug 16, 2023 · Aug 16, 2023
commit a92c095c2ca9edc99be00dc7d793a272f301a1c1
@@ -1,7 +1,7 @@
 import sqlite3 as sqlite
 import unittest
 
-from .test_dbapi import memory_database
+from test.test_sqlite3.util import memory_database
 
 
 class BackupTests(unittest.TestCase):

@@ -39,21 +39,7 @@
 from os import SEEK_SET, SEEK_CUR, SEEK_END
 from test.support.os_helper import TESTFN, TESTFN_UNDECODABLE, unlink, temp_dir, FakePath
 
-
-# Helper for temporary memory databases
-def memory_database(*args, **kwargs):
-    cx = sqlite.connect(":memory:", *args, **kwargs)
-    return contextlib.closing(cx)
-
-
-# Temporarily limit a database connection parameter
-@contextlib.contextmanager
-def cx_limit(cx, category=sqlite.SQLITE_LIMIT_SQL_LENGTH, limit=128):
-    try:
-        _prev = cx.setlimit(category, limit)
-        yield limit
    finally:
-        cx.setlimit(category, _prev)
+from test.test_sqlite3.util import memory_database, cx_limit
 
 
 class ModuleTests(unittest.TestCase):
@@ -593,8 +579,8 @@ def test_connect_positional_arguments(self):
             "parameters in Python 3.15."
         )
         with self.assertWarnsRegex(DeprecationWarning, regex) as cm:
-            with memory_database(1.0):
-                pass
+            cx = sqlite.connect(":memory:", 1.0)
+            cx.close()
         self.assertEqual(cm.filename, __file__)
 
 
@@ -1695,34 +1681,34 @@ def test_connection_executescript(self):
         result = con.execute("select foo from test").fetchone()[0]
         self.assertEqual(result, 5, "Basic test of Connection.executescript")
 
+
 class ClosedConTests(unittest.TestCase):
-    REGEX = "Cannot operate on a closed database."
+    def check(self, fn, *args, **kwds):
+        regex = "Cannot operate on a closed database."
+        with self.assertRaisesRegex(sqlite.ProgrammingError, regex):
+            fn(*args, **kwds)
 
     def setUp(self):
         self.con = sqlite.connect(":memory:")
         self.cur = self.con.cursor()
         self.con.close()
 
     def test_closed_con_cursor(self):
-        with self.assertRaisesRegex(sqlite.ProgrammingError, self.REGEX):
-            self.con.cursor()
+        self.check(self.con.cursor)
 
     def test_closed_con_commit(self):
-        with self.assertRaisesRegex(sqlite.ProgrammingError, self.REGEX):
-            self.con.commit()
+        self.check(self.con.commit)
 
     def test_closed_con_rollback(self):
-        with self.assertRaisesRegex(sqlite.ProgrammingError, self.REGEX):
-            self.con.rollback()
+        self.check(self.con.rollback)
 
     def test_closed_cur_execute(self):
-        with self.assertRaisesRegex(sqlite.ProgrammingError, self.REGEX):
-            self.cur.execute("select 4")
+        self.check(self.cur.execute, "select 4")
 
     def test_closed_create_function(self):
-        def f(x): return 17
-        with self.assertRaisesRegex(sqlite.ProgrammingError, self.REGEX):
-            self.con.create_function("foo", 1, f)
+        def f(x):
+            return 17
+        self.check(self.con.create_function, "foo", 1, f)
 
     def test_closed_create_aggregate(self):
         class Agg:
@@ -1732,23 +1718,21 @@ def step(self, x):
                 pass
             def finalize(self):
                 return 17
-        with self.assertRaisesRegex(sqlite.ProgrammingError, self.REGEX):
-            self.con.create_aggregate("foo", 1, Agg)
+        self.check(self.con.create_aggregate, "foo", 1, Agg)
 
     def test_closed_set_authorizer(self):
         def authorizer(*args):
             return sqlite.DENY
-        with self.assertRaisesRegex(sqlite.ProgrammingError, self.REGEX):
-            self.con.set_authorizer(authorizer)
+        self.check(self.con.set_authorizer, authorizer)
 
     def test_closed_set_progress_callback(self):
-        def progress(): pass
-        with self.assertRaisesRegex(sqlite.ProgrammingError, self.REGEX):
-            self.con.set_progress_handler(progress, 100)
+        def progress():
+            pass
+        self.check(self.con.set_progress_handler, progress, 100)
 
     def test_closed_call(self):
-        with self.assertRaisesRegex(sqlite.ProgrammingError, self.REGEX):
-            self.con()
+        self.check(self.con)
+
 
 class ClosedCurTests(unittest.TestCase):
     def test_closed(self):

@@ -2,16 +2,12 @@
 
 import unittest
 import sqlite3 as sqlite
-from .test_dbapi import memory_database
 
+from test.test_sqlite3.util import memory_database
+from test.test_sqlite3.util import MemoryDatabaseMixin
 
-class DumpTests(unittest.TestCase):
-    def setUp(self):
-        self.cx = sqlite.connect(":memory:")
-        self.cu = self.cx.cursor()
 
-    def tearDown(self):
-        self.cx.close()
+class DumpTests(MemoryDatabaseMixin, unittest.TestCase):
 
     def test_table_dump(self):
         expected_sqls = [

@@ -24,7 +24,8 @@
 import sqlite3 as sqlite
 from collections.abc import Sequence
 
-from .test_dbapi import memory_database
+from test.test_sqlite3.util import memory_database
+from test.test_sqlite3.util import MemoryDatabaseMixin
 
 
 def dict_factory(cursor, row):
@@ -84,12 +85,7 @@ def __init__(self, *args, **kwargs):
         self.assertEqual(cm.filename, __file__)
 
 
-class CursorFactoryTests(unittest.TestCase):
-    def setUp(self):
-        self.con = sqlite.connect(":memory:")
-
-    def tearDown(self):
-        self.con.close()
+class CursorFactoryTests(MemoryDatabaseMixin, unittest.TestCase):
 
     def test_is_instance(self):
         cur = self.con.cursor()
@@ -107,9 +103,8 @@ def test_invalid_factory(self):
         # invalid callable returning non-cursor
         self.assertRaises(TypeError, self.con.cursor, lambda con: None)
 
-class RowFactoryTestsBackwardsCompat(unittest.TestCase):
-    def setUp(self):
-        self.con = sqlite.connect(":memory:")
+
+class RowFactoryTestsBackwardsCompat(MemoryDatabaseMixin, unittest.TestCase):
 
     def test_is_produced_by_factory(self):
         cur = self.con.cursor(factory=MyCursor)
@@ -118,12 +113,8 @@ def test_is_produced_by_factory(self):
         self.assertIsInstance(row, dict)
         cur.close()
 
-    def tearDown(self):
-        self.con.close()
 
-class RowFactoryTests(unittest.TestCase):
-    def setUp(self):
-        self.con = sqlite.connect(":memory:")
+class RowFactoryTests(MemoryDatabaseMixin, unittest.TestCase):
 
     def test_custom_factory(self):
         self.con.row_factory = lambda cur, row: list(row)
@@ -269,12 +260,8 @@ class FakeCursor(str):
         self.assertRaises(TypeError, self.con.cursor, FakeCursor)
         self.assertRaises(TypeError, sqlite.Row, FakeCursor(), ())
 
-    def tearDown(self):
-        self.con.close()
 
-class TextFactoryTests(unittest.TestCase):
-    def setUp(self):
-        self.con = sqlite.connect(":memory:")
+class TextFactoryTests(MemoryDatabaseMixin, unittest.TestCase):
 
     def test_unicode(self):
         austria = "Österreich"
@@ -295,15 +282,17 @@ def test_custom(self):
         self.assertEqual(type(row[0]), str, "type of row[0] must be unicode")
         self.assertTrue(row[0].endswith("reich"), "column must contain original data")
 
-    def tearDown(self):
-        self.con.close()
 
 class TextFactoryTestsWithEmbeddedZeroBytes(unittest.TestCase):
+
     def setUp(self):
         self.con = sqlite.connect(":memory:")
         self.con.execute("create table test (value text)")
         self.con.execute("insert into test (value) values (?)", ("a\x00b",))
 
+    def tearDown(self):
+        self.con.close()
+
     def test_string(self):
         # text_factory defaults to str
         row = self.con.execute("select value from test").fetchone()
@@ -329,9 +318,6 @@ def test_custom(self):
         self.assertIs(type(row[0]), bytes)
         self.assertEqual(row[0], b"a\x00b")
 
-    def tearDown(self):
-        self.con.close()
-
 
 if __name__ == "__main__":
     unittest.main()
@@ -26,16 +26,11 @@
 
 from test.support.os_helper import TESTFN, unlink
 
-from test.test_sqlite3.test_dbapi import memory_database, cx_limit
-from test.test_sqlite3.test_userfunctions import with_tracebacks
+from test.test_sqlite3.util import memory_database, cx_limit, with_tracebacks
+from test.test_sqlite3.util import MemoryDatabaseMixin
 
 
-class CollationTests(unittest.TestCase):
-    def setUp(self):
-        self.con = sqlite.connect(":memory:")
-
-    def tearDown(self):
-        self.con.close()
+class CollationTests(MemoryDatabaseMixin, unittest.TestCase):
 
     def test_create_collation_not_string(self):
         with self.assertRaises(TypeError):
@@ -133,12 +128,8 @@ def test_deregister_collation(self):
             con.execute("select 'a' as x union select 'b' as x order by x collate mycoll")
         self.assertEqual(str(cm.exception), 'no such collation sequence: mycoll')
 
-class ProgressTests(unittest.TestCase):
-    def setUp(self):
-        self.con = sqlite.connect(":memory:")
 
-    def tearDown(self):
-        self.con.close()
+class ProgressTests(MemoryDatabaseMixin, unittest.TestCase):
 
     def test_progress_handler_used(self):
         """
@@ -229,12 +220,7 @@ def bad_progress():
                 """)
 
 
-class TraceCallbackTests(unittest.TestCase):
-    def setUp(self):
-        self.con = sqlite.connect(":memory:")
-
-    def tearDown(self):
-        self.con.close()
+class TraceCallbackTests(MemoryDatabaseMixin, unittest.TestCase):
 
     @contextlib.contextmanager
     def check_stmt_trace(self, cx, expected):

@@ -28,15 +28,12 @@
 
 from test import support
 from unittest.mock import patch
-from test.test_sqlite3.test_dbapi import memory_database, cx_limit
 
+from test.test_sqlite3.util import memory_database, cx_limit
+from test.test_sqlite3.util import MemoryDatabaseMixin
 
-class RegressionTests(unittest.TestCase):
-    def setUp(self):
-        self.con = sqlite.connect(":memory:")
 
-    def tearDown(self):
-        self.con.close()
+class RegressionTests(MemoryDatabaseMixin, unittest.TestCase):
 
     def test_pragma_user_version(self):
         # This used to crash pysqlite because this pragma command returns NULL for the column name
@@ -232,7 +229,8 @@ def test_auto_commit(self):
         be created.
         """
         with memory_database(isolation_level=None) as con:
-            pass
+            self.assertIsNone(con.isolation_level)
+            self.assertFalse(con.in_transaction)
 
     def test_pragma_autocommit(self):
         """
@@ -310,8 +308,7 @@ def test_invalid_isolation_level_type(self):
         # isolation level is a string, not an integer
         regex = "isolation_level must be str or None"
         with self.assertRaisesRegex(TypeError, regex):
-            with memory_database(isolation_level=123):
-                pass
+            memory_database(isolation_level=123).__enter__()
 
 
     def test_null_character(self):

@@ -28,7 +28,8 @@
 from test.support.os_helper import TESTFN, unlink
 from test.support.script_helper import assert_python_ok
 
-from test.test_sqlite3.test_dbapi import memory_database
+from test.test_sqlite3.util import memory_database
+from test.test_sqlite3.util import MemoryDatabaseMixin
 
 
 TIMEOUT = LOOPBACK_TIMEOUT / 10
@@ -218,10 +219,7 @@ def test_no_duplicate_rows_after_rollback_new_query(self):
 
 
 
-class SpecialCommandTests(unittest.TestCase):
-    def setUp(self):
-        self.con = sqlite.connect(":memory:")
-        self.cur = self.con.cursor()
+class SpecialCommandTests(MemoryDatabaseMixin, unittest.TestCase):
 
     def test_drop_table(self):
         self.cur.execute("create table test(i)")
@@ -233,14 +231,8 @@ def test_pragma(self):
         self.cur.execute("insert into test(i) values (5)")
         self.cur.execute("pragma count_changes=1")
 
-    def tearDown(self):
-        self.cur.close()
-        self.con.close()
-
 
-class TransactionalDDL(unittest.TestCase):
-    def setUp(self):
-        self.con = sqlite.connect(":memory:")
+class TransactionalDDL(MemoryDatabaseMixin, unittest.TestCase):
 
     def test_ddl_does_not_autostart_transaction(self):
         # For backwards compatibility reasons, DDL statements should not
@@ -268,9 +260,6 @@ def test_transactional_ddl(self):
         with self.assertRaises(sqlite.OperationalError):
             self.con.execute("select * from test")
 
-    def tearDown(self):
-        self.con.close()
-
 
 class IsolationLevelFromInit(unittest.TestCase):
     CREATE = "create table t(t)"