8000 gh-105539: Emit ResourceWarning if sqlite3 database is not closed explicitly by erlend-aasland · Pull Request #108015 · python/cpython · GitHub
[go: up one dir, main page]

Skip to content

gh-105539: Emit ResourceWarning if sqlite3 database is not closed explicitly #108015

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 12 commits into from
Aug 22, 2023
Merged
Prev Previous commit
Next Next commit
Pull in main
  • Loading branch information
erlend-aasland committed Aug 17, 2023
commit 8075485b0ca4c70fc29b00897f65bf932bd361a6
2 changes: 1 addition & 1 deletion Lib/test/test_sqlite3/test_backup.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
import sqlite3 as sqlite
import unittest

from .test_dbapi import memory_database
from .util import memory_database


class BackupTests(unittest.TestCase):
Expand Down
49 changes: 20 additions & 29 deletions Lib/test/test_sqlite3/test_dbapi.py
Original file line number Diff line number Diff line change
Expand Up @@ -579,16 +579,10 @@ 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__)

def test_connection_resource_warning(self):
with self.assertWarns(ResourceWarning):
cx = sqlite.connect(":memory:")
del cx
gc_collect()


class UninitialisedConnectionTests(unittest.TestCase):
def setUp(self):
Expand Down Expand Up @@ -1689,33 +1683,32 @@ def test_connection_executescript(self):


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:
Expand All @@ -1725,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):
Expand Down
3 changes: 2 additions & 1 deletion Lib/test/test_sqlite3/test_factory.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,8 @@
import sqlite3 as sqlite
from collections.abc import Sequence

from .test_dbapi import memory_database
from .util import memory_database
from .util import MemoryDatabaseMixin


def dict_factory(cursor, row):
Expand Down
20 changes: 3 additions & 17 deletions Lib/test/test_sqlite3/test_hooks.py
8000
Original file line number Diff line number Diff line change
Expand Up @@ -30,12 +30,7 @@
from .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):
Expand Down Expand Up @@ -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):
"""
Expand Down Expand Up @@ -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):
Expand Down
6 changes: 3 additions & 3 deletions Lib/test/test_sqlite3/test_regression.py
Original file line number Diff line number Diff line change
Expand Up @@ -229,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):
"""
Expand Down Expand Up @@ -307,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):
Expand Down
45 changes: 5 additions & 40 deletions Lib/test/test_sqlite3/test_userfunctions.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,43 +28,8 @@
from unittest.mock import Mock, patch
from test.support import bigmemtest, gc_collect

from test.test_sqlite3.test_dbapi import cx_limit, memory_database


def with_tracebacks(exc, regex="", name=""):
"""Convenience decorator for testing callback tracebacks."""
def decorator(func):
_regex = re.compile(regex) if regex else None
@functools.wraps(func)
def wrapper(self, *args, **kwargs):
with catch_unraisable_exception() as cm:
# First, run the test with traceback enabled.
with check_tracebacks(self, cm, exc, _regex, name):
func(self, *args, **kwargs)

# Then run the test with traceback disabled.
func(self, *args, **kwargs)
return wrapper
return decorator


@contextlib.contextmanager
def check_tracebacks(self, cm, exc, regex, obj_name):
"""Convenience context manager for testing callback tracebacks."""
sqlite.enable_callback_tracebacks(True)
try:
buf = io.StringIO()
with contextlib.redirect_stderr(buf):
yield

self.assertEqual(cm.unraisable.exc_type, exc)
if regex:
msg = str(cm.unraisable.exc_value)
self.assertIsNotNone(regex.search(msg))
if obj_name:
self.assertEqual(cm.unraisable.object.__name__, obj_name)
finally:
sqlite.enable_callback_tracebacks(False)
from .util import cx_limit, memory_database
from .util import with_tracebacks, check_tracebacks


def func_returntext():
Expand Down Expand Up @@ -401,10 +366,10 @@ def test_func_deterministic_keyword_only(self):
def test_function_destructor_via_gc(self):
# See bpo-44304: The destructor of the user function can
# crash if is called without the GIL from the gc functions
with memory_database() as dest:
def md5sum(t):
return
def md5sum(t):
return

with memory_database() as dest:
dest.create_function("md5", 1, md5sum)
x = dest("create table lang (name, first_appeared)")
del md5sum, dest
Expand Down
You are viewing a condensed version of this merge commit. You can view the full changes here.
0