bpo-44958: Fix ref. leak introduced in GH-27844 (GH-28490)

Modify managed_connect() helper to support in-memory databases. Use it
for the regression tests added in GH-27844.

Automerge-Triggered-By: GH:pablogsal

diff --git a/Lib/sqlite3/test/test_dbapi.py b/Lib/sqlite3/test/test_dbapi.py
index 34cadeebfa2627b1a4f1d6913cc3106720867df0..732e21dd3a285ee1ca1530e7beef6573883c2a7a 100644 (file)
--- a/Lib/sqlite3/test/test_dbapi.py
+++ b/Lib/sqlite3/test/test_dbapi.py
@@ -38,13 +38,14 @@ from test.support.os_helper import TESTFN, unlink, temp_dir
 
 # Helper for tests using TESTFN
 @contextlib.contextmanager
-def managed_connect(*args, **kwargs):
+def managed_connect(*args, in_mem=False, **kwargs):
     cx = sqlite.connect(*args, **kwargs)
     try:
         yield cx
     finally:
         cx.close()
-        unlink(TESTFN)
+        if not in_mem:
+            unlink(TESTFN)
 
 
 class ModuleTests(unittest.TestCase):

diff --git a/Lib/sqlite3/test/test_regression.py b/Lib/sqlite3/test/test_regression.py
index d41f11814133a52f55e11e99546ae27fb29c699d..ff356734860b645c504b4a3e9023ba9320800bf2 100644 (file)
--- a/Lib/sqlite3/test/test_regression.py
+++ b/Lib/sqlite3/test/test_regression.py
@@ -28,6 +28,8 @@ import weakref
 import functools
 from test import support
 
+from .test_dbapi import managed_connect
+
 class RegressionTests(unittest.TestCase):
     def setUp(self):
         self.con = sqlite.connect(":memory:")
@@ -437,38 +439,41 @@ class RegressionTests(unittest.TestCase):
         self.assertEqual(val, b'')
 
     def test_table_lock_cursor_replace_stmt(self):
-        con = sqlite.connect(":memory:")
-        cur = con.cursor()
-        cur.execute("create table t(t)")
-        cur.executemany("insert into t values(?)", ((v,) for v in range(5)))
-        con.commit()
-        cur.execute("select t from t")
-        cur.execute("drop table t")
-        con.commit()
+        with managed_connect(":memory:", in_mem=True) as con:
+            cur = con.cursor()
+            cur.execute("create table t(t)")
+            cur.executemany("insert into t values(?)",
+                            ((v,) for v in range(5)))
+            con.commit()
+            cur.execute("select t from t")
+            cur.execute("drop table t")
+            con.commit()
 
     def test_table_lock_cursor_dealloc(self):
-        con = sqlite.connect(":memory:")
-        con.execute("create table t(t)")
-        con.executemany("insert into t values(?)", ((v,) for v in range(5)))
-        con.commit()
-        cur = con.execute("select t from t")
-        del cur
-        con.execute("drop table t")
-        con.commit()
+        with managed_connect(":memory:", in_mem=True) as con:
+            con.execute("create table t(t)")
+            con.executemany("insert into t values(?)",
+                            ((v,) for v in range(5)))
+            con.commit()
+            cur = con.execute("select t from t")
+            del cur
+            con.execute("drop table t")
+            con.commit()
 
     def test_table_lock_cursor_non_readonly_select(self):
-        con = sqlite.connect(":memory:")
-        con.execute("create table t(t)")
-        con.executemany("insert into t values(?)", ((v,) for v in range(5)))
-        con.commit()
-        def dup(v):
-            con.execute("insert into t values(?)", (v,))
-            return
-        con.create_function("dup", 1, dup)
-        cur = con.execute("select dup(t) from t")
-        del cur
-        con.execute("drop table t")
-        con.commit()
+        with managed_connect(":memory:", in_mem=True) as con:
+            con.execute("create table t(t)")
+            con.executemany("insert into t values(?)",
+                            ((v,) for v in range(5)))
+            con.commit()
+            def dup(v):
+                con.execute("insert into t values(?)", (v,))
+                return
+            con.create_function("dup", 1, dup)
+            cur = con.execute("select dup(t) from t")
+            del cur
+            con.execute("drop table t")
+            con.commit()
 
 
 if __name__ == "__main__":