[Python-checkins] bpo-44958: Fix ref. leak introduced in GH-27844 (GH-28490)

miss-islington webhook-mailer at python.org
Tue Sep 21 09:16:03 EDT 2021


https://github.com/python/cpython/commit/3e3ff09058a71b92c32d1d7dc32470c0cf49300c
commit: 3e3ff09058a71b92c32d1d7dc32470c0cf49300c
branch: main
author: Erlend Egeberg Aasland <erlend.aasland at innova.no>
committer: miss-islington <31488909+miss-islington at users.noreply.github.com>
date: 2021-09-21T06:15:54-07:00
summary:

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

files:
M Lib/sqlite3/test/test_dbapi.py
M Lib/sqlite3/test/test_regression.py

diff --git a/Lib/sqlite3/test/test_dbapi.py b/Lib/sqlite3/test/test_dbapi.py
index 34cadeebfa262..732e21dd3a285 100644
--- a/Lib/sqlite3/test/test_dbapi.py
+++ b/Lib/sqlite3/test/test_dbapi.py
@@ -38,13 +38,14 @@
 
 # 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 d41f11814133a..ff356734860b6 100644
--- a/Lib/sqlite3/test/test_regression.py
+++ b/Lib/sqlite3/test/test_regression.py
@@ -28,6 +28,8 @@
 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 @@ def test_return_empty_bytestring(self):
         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__":



More information about the Python-checkins mailing list