mirror of
https://github.com/python/cpython.git
synced 2025-09-12 11:46:52 +00:00
gh-105539: Explict resource management for connection objects in sqlite3 tests (#108017)
- Use memory_database() helper - Move test utility functions to util.py - Add convenience memory database mixin - Add check() helper for closed connection tests
This commit is contained in:
Erlend E. Aasland
GitHub
parent
c9d83f93d8
commit
1344cfac43
9 changed files with 371 additions and 385 deletions
|
|
@ -33,26 +33,13 @@ from test.support import (
|
|||
SHORT_TIMEOUT, check_disallow_instantiation, requires_subprocess,
|
||||
is_emscripten, is_wasi
|
||||
)
|
||||
from test.support import gc_collect
|
||||
from test.support import threading_helper
|
||||
from _testcapi import INT_MAX, ULLONG_MAX
|
||||
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 .util import memory_database, cx_limit
|
||||
|
||||
|
||||
class ModuleTests(unittest.TestCase):
|
||||
|
|
@ -326,9 +313,9 @@ class ModuleTests(unittest.TestCase):
|
|||
self.assertEqual(exc.sqlite_errorname, "SQLITE_CONSTRAINT_CHECK")
|
||||
|
||||
def test_disallow_instantiation(self):
|
||||
cx = sqlite.connect(":memory:")
|
||||
check_disallow_instantiation(self, type(cx("select 1")))
|
||||
check_disallow_instantiation(self, sqlite.Blob)
|
||||
with memory_database() as cx:
|
||||
check_disallow_instantiation(self, type(cx("select 1")))
|
||||
check_disallow_instantiation(self, sqlite.Blob)
|
||||
|
||||
def test_complete_statement(self):
|
||||
self.assertFalse(sqlite.complete_statement("select t"))
|
||||
|
|
@ -342,6 +329,7 @@ class ConnectionTests(unittest.TestCase):
|
|||
cu = self.cx.cursor()
|
||||
cu.execute("create table test(id integer primary key, name text)")
|
||||
cu.execute("insert into test(name) values (?)", ("foo",))
|
||||
cu.close()
|
||||
|
||||
def tearDown(self):
|
||||
self.cx.close()
|
||||
|
|
@ -412,21 +400,22 @@ class ConnectionTests(unittest.TestCase):
|
|||
|
||||
def test_in_transaction(self):
|
||||
# Can't use db from setUp because we want to test initial state.
|
||||
cx = sqlite.connect(":memory:")
|
||||
cu = cx.cursor()
|
||||
self.assertEqual(cx.in_transaction, False)
|
||||
cu.execute("create table transactiontest(id integer primary key, name text)")
|
||||
self.assertEqual(cx.in_transaction, False)
|
||||
cu.execute("insert into transactiontest(name) values (?)", ("foo",))
|
||||
self.assertEqual(cx.in_transaction, True)
|
||||
cu.execute("select name from transactiontest where name=?", ["foo"])
|
||||
row = cu.fetchone()
|
||||
self.assertEqual(cx.in_transaction, True)
|
||||
cx.commit()
|
||||
self.assertEqual(cx.in_transaction, False)
|
||||
cu.execute("select name from transactiontest where name=?", ["foo"])
|
||||
row = cu.fetchone()
|
||||
self.assertEqual(cx.in_transaction, False)
|
||||
with memory_database() as cx:
|
||||
cu = cx.cursor()
|
||||
self.assertEqual(cx.in_transaction, False)
|
||||
cu.execute("create table transactiontest(id integer primary key, name text)")
|
||||
self.assertEqual(cx.in_transaction, False)
|
||||
cu.execute("insert into transactiontest(name) values (?)", ("foo",))
|
||||
self.assertEqual(cx.in_transaction, True)
|
||||
cu.execute("select name from transactiontest where name=?", ["foo"])
|
||||
row = cu.fetchone()
|
||||
self.assertEqual(cx.in_transaction, True)
|
||||
cx.commit()
|
||||
self.assertEqual(cx.in_transaction, False)
|
||||
cu.execute("select name from transactiontest where name=?", ["foo"])
|
||||
row = cu.fetchone()
|
||||
self.assertEqual(cx.in_transaction, False)
|
||||
cu.close()
|
||||
|
||||
def test_in_transaction_ro(self):
|
||||
with self.assertRaises(AttributeError):
|
||||
|
|
@ -450,10 +439,9 @@ class ConnectionTests(unittest.TestCase):
|
|||
self.assertIs(getattr(sqlite, exc), getattr(self.cx, exc))
|
||||
|
||||
def test_interrupt_on_closed_db(self):
|
||||
cx = sqlite.connect(":memory:")
|
||||
cx.close()
|
||||
self.cx.close()
|
||||
with self.assertRaises(sqlite.ProgrammingError):
|
||||
cx.interrupt()
|
||||
self.cx.interrupt()
|
||||
|
||||
def test_interrupt(self):
|
||||
self.assertIsNone(self.cx.interrupt())
|
||||
|
|
@ -521,29 +509,29 @@ class ConnectionTests(unittest.TestCase):
|
|||
self.assertEqual(cx.isolation_level, level)
|
||||
|
||||
def test_connection_reinit(self):
|
||||
db = ":memory:"
|
||||
cx = sqlite.connect(db)
|
||||
cx.text_factory = bytes
|
||||
cx.row_factory = sqlite.Row
|
||||
cu = cx.cursor()
|
||||
cu.execute("create table foo (bar)")
|
||||
cu.executemany("insert into foo (bar) values (?)",
|
||||
((str(v),) for v in range(4)))
|
||||
cu.execute("select bar from foo")
|
||||
with memory_database() as cx:
|
||||
cx.text_factory = bytes
|
||||
cx.row_factory = sqlite.Row
|
||||
cu = cx.cursor()
|
||||
cu.execute("create table foo (bar)")
|
||||
cu.executemany("insert into foo (bar) values (?)",
|
||||
((str(v),) for v in range(4)))
|
||||
cu.execute("select bar from foo")
|
||||
|
||||
rows = [r for r in cu.fetchmany(2)]
|
||||
self.assertTrue(all(isinstance(r, sqlite.Row) for r in rows))
|
||||
self.assertEqual([r[0] for r in rows], [b"0", b"1"])
|
||||
rows = [r for r in cu.fetchmany(2)]
|
||||
self.assertTrue(all(isinstance(r, sqlite.Row) for r in rows))
|
||||
self.assertEqual([r[0] for r in rows], [b"0", b"1"])
|
||||
|
||||
cx.__init__(db)
|
||||
cx.execute("create table foo (bar)")
|
||||
cx.executemany("insert into foo (bar) values (?)",
|
||||
((v,) for v in ("a", "b", "c", "d")))
|
||||
cx.__init__(":memory:")
|
||||
cx.execute("create table foo (bar)")
|
||||
cx.executemany("insert into foo (bar) values (?)",
|
||||
((v,) for v in ("a", "b", "c", "d")))
|
||||
|
||||
# This uses the old database, old row factory, but new text factory
|
||||
rows = [r for r in cu.fetchall()]
|
||||
self.assertTrue(all(isinstance(r, sqlite.Row) for r in rows))
|
||||
self.assertEqual([r[0] for r in rows], ["2", "3"])
|
||||
# This uses the old database, old row factory, but new text factory
|
||||
rows = [r for r in cu.fetchall()]
|
||||
self.assertTrue(all(isinstance(r, sqlite.Row) for r in rows))
|
||||
self.assertEqual([r[0] for r in rows], ["2", "3"])
|
||||
cu.close()
|
||||
|
||||
def test_connection_bad_reinit(self):
|
||||
cx = sqlite.connect(":memory:")
|
||||
|
|
@ -591,11 +579,11 @@ class ConnectionTests(unittest.TestCase):
|
|||
"parameters in Python 3.15."
|
||||
)
|
||||
with self.assertWarnsRegex(DeprecationWarning, regex) as cm:
|
||||
sqlite.connect(":memory:", 1.0)
|
||||
cx = sqlite.connect(":memory:", 1.0)
|
||||
cx.close()
|
||||
self.assertEqual(cm.filename, __file__)
|
||||
|
||||
|
||||
|
||||
class UninitialisedConnectionTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.cx = sqlite.Connection.__new__(sqlite.Connection)
|
||||
|
|
@ -1571,12 +1559,12 @@ class ThreadTests(unittest.TestCase):
|
|||
except sqlite.Error:
|
||||
err.append("multi-threading not allowed")
|
||||
|
||||
con = sqlite.connect(":memory:", check_same_thread=False)
|
||||
err = []
|
||||
t = threading.Thread(target=run, kwargs={"con": con, "err": err})
|
||||
t.start()
|
||||
t.join()
|
||||
self.assertEqual(len(err), 0, "\n".join(err))
|
||||
with memory_database(check_same_thread=False) as con:
|
||||
err = []
|
||||
t = threading.Thread(target=run, kwargs={"con": con, "err": err})
|
||||
t.start()
|
||||
t.join()
|
||||
self.assertEqual(len(err), 0, "\n".join(err))
|
||||
|
||||
|
||||
class ConstructorTests(unittest.TestCase):
|
||||
|
|
@ -1602,9 +1590,16 @@ class ConstructorTests(unittest.TestCase):
|
|||
b = sqlite.Binary(b"\0'")
|
||||
|
||||
class ExtensionTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.con = sqlite.connect(":memory:")
|
||||
self.cur = self.con.cursor()
|
||||
|
||||
def tearDown(self):
|
||||
self.cur.close()
|
||||
self.con.close()
|
||||
|
||||
def test_script_string_sql(self):
|
||||
con = sqlite.connect(":memory:")
|
||||
cur = con.cursor()
|
||||
cur = self.cur
|
||||
cur.executescript("""
|
||||
-- bla bla
|
||||
/* a stupid comment */
|
||||
|
|
@ -1616,40 +1611,40 @@ class ExtensionTests(unittest.TestCase):
|
|||
self.assertEqual(res, 5)
|
||||
|
||||
def test_script_syntax_error(self):
|
||||
con = sqlite.connect(":memory:")
|
||||
cur = con.cursor()
|
||||
with self.assertRaises(sqlite.OperationalError):
|
||||
cur.executescript("create table test(x); asdf; create table test2(x)")
|
||||
self.cur.executescript("""
|
||||
CREATE TABLE test(x);
|
||||
asdf;
|
||||
CREATE TABLE test2(x)
|
||||
""")
|
||||
|
||||
def test_script_error_normal(self):
|
||||
con = sqlite.connect(":memory:")
|
||||
cur = con.cursor()
|
||||
with self.assertRaises(sqlite.OperationalError):
|
||||
cur.executescript("create table test(sadfsadfdsa); select foo from hurz;")
|
||||
self.cur.executescript("""
|
||||
CREATE TABLE test(sadfsadfdsa);
|
||||
SELECT foo FROM hurz;
|
||||
""")
|
||||
|
||||
def test_cursor_executescript_as_bytes(self):
|
||||
con = sqlite.connect(":memory:")
|
||||
cur = con.cursor()
|
||||
with self.assertRaises(TypeError):
|
||||
cur.executescript(b"create table test(foo); insert into test(foo) values (5);")
|
||||
self.cur.executescript(b"""
|
||||
CREATE TABLE test(foo);
|
||||
INSERT INTO test(foo) VALUES (5);
|
||||
""")
|
||||
|
||||
def test_cursor_executescript_with_null_characters(self):
|
||||
con = sqlite.connect(":memory:")
|
||||
cur = con.cursor()
|
||||
with self.assertRaises(ValueError):
|
||||
cur.executescript("""
|
||||
create table a(i);\0
|
||||
insert into a(i) values (5);
|
||||
""")
|
||||
self.cur.executescript("""
|
||||
CREATE TABLE a(i);\0
|
||||
INSERT INTO a(i) VALUES (5);
|
||||
""")
|
||||
|
||||
def test_cursor_executescript_with_surrogates(self):
|
||||
con = sqlite.connect(":memory:")
|
||||
cur = con.cursor()
|
||||
with self.assertRaises(UnicodeEncodeError):
|
||||
cur.executescript("""
|
||||
create table a(s);
|
||||
insert into a(s) values ('\ud8ff');
|
||||
""")
|
||||
self.cur.executescript("""
|
||||
CREATE TABLE a(s);
|
||||
INSERT INTO a(s) VALUES ('\ud8ff');
|
||||
""")
|
||||
|
||||
def test_cursor_executescript_too_large_script(self):
|
||||
msg = "query string is too large"
|
||||
|
|
@ -1659,19 +1654,18 @@ class ExtensionTests(unittest.TestCase):
|
|||
cx.executescript("select 'too large'".ljust(lim+1))
|
||||
|
||||
def test_cursor_executescript_tx_control(self):
|
||||
con = sqlite.connect(":memory:")
|
||||
con = self.con
|
||||
con.execute("begin")
|
||||
self.assertTrue(con.in_transaction)
|
||||
con.executescript("select 1")
|
||||
self.assertFalse(con.in_transaction)
|
||||
|
||||
def test_connection_execute(self):
|
||||
con = sqlite.connect(":memory:")
|
||||
result = con.execute("select 5").fetchone()[0]
|
||||
result = self.con.execute("select 5").fetchone()[0]
|
||||
self.assertEqual(result, 5, "Basic test of Connection.execute")
|
||||
|
||||
def test_connection_executemany(self):
|
||||
con = sqlite.connect(":memory:")
|
||||
con = self.con
|
||||
con.execute("create table test(foo)")
|
||||
con.executemany("insert into test(foo) values (?)", [(3,), (4,)])
|
||||
result = con.execute("select foo from test order by foo").fetchall()
|
||||
|
|
@ -1679,47 +1673,44 @@ class ExtensionTests(unittest.TestCase):
|
|||
self.assertEqual(result[1][0], 4, "Basic test of Connection.executemany")
|
||||
|
||||
def test_connection_executescript(self):
|
||||
con = sqlite.connect(":memory:")
|
||||
con.executescript("create table test(foo); insert into test(foo) values (5);")
|
||||
con = self.con
|
||||
con.executescript("""
|
||||
CREATE TABLE test(foo);
|
||||
INSERT INTO test(foo) VALUES (5);
|
||||
""")
|
||||
result = con.execute("select foo from test").fetchone()[0]
|
||||
self.assertEqual(result, 5, "Basic test of Connection.executescript")
|
||||
|
||||
|
||||
class ClosedConTests(unittest.TestCase):
|
||||
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):
|
||||
con = sqlite.connect(":memory:")
|
||||
con.close()
|
||||
with self.assertRaises(sqlite.ProgrammingError):
|
||||
cur = con.cursor()
|
||||
self.check(self.con.cursor)
|
||||
|
||||
def test_closed_con_commit(self):
|
||||
con = sqlite.connect(":memory:")
|
||||
con.close()
|
||||
with self.assertRaises(sqlite.ProgrammingError):
|
||||
con.commit()
|
||||
self.check(self.con.commit)
|
||||
|
||||
def test_closed_con_rollback(self):
|
||||
con = sqlite.connect(":memory:")
|
||||
con.close()
|
||||
with self.assertRaises(sqlite.ProgrammingError):
|
||||
con.rollback()
|
||||
self.check(self.con.rollback)
|
||||
|
||||
def test_closed_cur_execute(self):
|
||||
con = sqlite.connect(":memory:")
|
||||
cur = con.cursor()
|
||||
con.close()
|
||||
with self.assertRaises(sqlite.ProgrammingError):
|
||||
cur.execute("select 4")
|
||||
self.check(self.cur.execute, "select 4")
|
||||
|
||||
def test_closed_create_function(self):
|
||||
con = sqlite.connect(":memory:")
|
||||
con.close()
|
||||
def f(x): return 17
|
||||
with self.assertRaises(sqlite.ProgrammingError):
|
||||
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):
|
||||
con = sqlite.connect(":memory:")
|
||||
con.close()
|
||||
class Agg:
|
||||
def __init__(self):
|
||||
pass
|
||||
|
|
@ -1727,29 +1718,21 @@ class ClosedConTests(unittest.TestCase):
|
|||
pass
|
||||
def finalize(self):
|
||||
return 17
|
||||
with self.assertRaises(sqlite.ProgrammingError):
|
||||
con.create_aggregate("foo", 1, Agg)
|
||||
self.check(self.con.create_aggregate, "foo", 1, Agg)
|
||||
|
||||
def test_closed_set_authorizer(self):
|
||||
con = sqlite.connect(":memory:")
|
||||
con.close()
|
||||
def authorizer(*args):
|
||||
return sqlite.DENY
|
||||
with self.assertRaises(sqlite.ProgrammingError):
|
||||
con.set_authorizer(authorizer)
|
||||
self.check(self.con.set_authorizer, authorizer)
|
||||
|
||||
def test_closed_set_progress_callback(self):
|
||||
con = sqlite.connect(":memory:")
|
||||
con.close()
|
||||
def progress(): pass
|
||||
with self.assertRaises(sqlite.ProgrammingError):
|
||||
con.set_progress_handler(progress, 100)
|
||||
def progress():
|
||||
pass
|
||||
self.check(self.con.set_progress_handler, progress, 100)
|
||||
|
||||
def test_closed_call(self):
|
||||
con = sqlite.connect(":memory:")
|
||||
con.close()
|
||||
with self.assertRaises(sqlite.ProgrammingError):
|
||||
con()
|
||||
self.check(self.con)
|
||||
|
||||
|
||||
class ClosedCurTests(unittest.TestCase):
|
||||
def test_closed(self):
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue