mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-10-14 10:53:40 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 8

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

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

Automerge-Triggered-By: GH:pablogsal
This commit is contained in:
Erlend Egeberg Aasland 2021-09-21 15:15:54 +02:00 • committed by GitHub
parent 050d103595
commit 3e3ff09058
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 36 additions and 30 deletions
Download patch file Download diff file Expand all files Collapse all files

5
Lib/sqlite3/test/test_dbapi.py
View file

@ -38,13 +38,14 @@ from test.support.os_helper import TESTFN, unlink, temp_dir
# Helper for tests using TESTFN # Helper for tests using TESTFN
@contextlib.contextmanager @contextlib.contextmanager
def managed_connect(*args, **kwargs): def managed_connect(*args, in_mem=False, **kwargs):
cx = sqlite.connect(*args, **kwargs) cx = sqlite.connect(*args, **kwargs)
try: try:
yield cx yield cx
finally: finally:
cx.close() cx.close()
unlink(TESTFN) if not in_mem:
unlink(TESTFN)
class ModuleTests(unittest.TestCase): class ModuleTests(unittest.TestCase):

61
Lib/sqlite3/test/test_regression.py
View file

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