New autouse fixture to help with test warnings

Refs https://github.com/simonw/sqlite-utils/issues/692#issuecomment-3644371889
This commit is contained in:
Simon Willison 2025-12-11 16:30:07 -08:00
commit 77359bea30
5 changed files with 46 additions and 46 deletions

View file

@ -14,11 +14,29 @@ def pytest_configure(config):
sys._called_from_test = True sys._called_from_test = True
@pytest.fixture(autouse=True)
def close_all_databases():
"""Automatically close all Database objects created during a test."""
databases = []
original_init = Database.__init__
def tracking_init(self, *args, **kwargs):
original_init(self, *args, **kwargs)
databases.append(self)
Database.__init__ = tracking_init
yield
Database.__init__ = original_init
for db in databases:
try:
db.close()
except Exception:
pass
@pytest.fixture @pytest.fixture
def fresh_db(): def fresh_db():
db = Database(memory=True) return Database(memory=True)
yield db
db.close()
@pytest.fixture @pytest.fixture
@ -32,8 +50,7 @@ def existing_db():
INSERT INTO foo (text) values ("three"); INSERT INTO foo (text) values ("three");
""" """
) )
yield database return database
database.close()
@pytest.fixture @pytest.fixture

View file

@ -44,7 +44,6 @@ def big_db_to_analyze_path(tmpdir):
} }
) )
db["stuff"].insert_all(to_insert) db["stuff"].insert_all(to_insert)
db.close()
return path return path
@ -313,7 +312,6 @@ def test_analyze_table_validate_columns(tmpdir, args, expected_error):
"age": 5, "age": 5,
} }
) )
db.close()
result = CliRunner().invoke( result = CliRunner().invoke(
cli.cli, cli.cli,
["analyze-tables", path] + args, ["analyze-tables", path] + args,

View file

@ -82,7 +82,6 @@ def test_tables_counts_and_columns(db_path):
db = Database(db_path) db = Database(db_path)
with db.conn: with db.conn:
db["lots"].insert_all([{"id": i, "age": i + 1} for i in range(30)]) db["lots"].insert_all([{"id": i, "age": i + 1} for i in range(30)])
db.close()
result = CliRunner().invoke(cli.cli, ["tables", "--counts", "--columns", db_path]) result = CliRunner().invoke(cli.cli, ["tables", "--counts", "--columns", db_path])
assert ( assert (
'[{"table": "Gosh", "count": 0, "columns": ["c1", "c2", "c3"]},\n' '[{"table": "Gosh", "count": 0, "columns": ["c1", "c2", "c3"]},\n'
@ -118,7 +117,6 @@ def test_tables_counts_and_columns_csv(db_path, format, expected):
db = Database(db_path) db = Database(db_path)
with db.conn: with db.conn:
db["lots"].insert_all([{"id": i, "age": i + 1} for i in range(30)]) db["lots"].insert_all([{"id": i, "age": i + 1} for i in range(30)])
db.close()
result = CliRunner().invoke( result = CliRunner().invoke(
cli.cli, ["tables", "--counts", "--columns", format, db_path] cli.cli, ["tables", "--counts", "--columns", format, db_path]
) )
@ -129,7 +127,6 @@ def test_tables_schema(db_path):
db = Database(db_path) db = Database(db_path)
with db.conn: with db.conn:
db["lots"].insert_all([{"id": i, "age": i + 1} for i in range(30)]) db["lots"].insert_all([{"id": i, "age": i + 1} for i in range(30)])
db.close()
result = CliRunner().invoke(cli.cli, ["tables", "--schema", db_path]) result = CliRunner().invoke(cli.cli, ["tables", "--schema", db_path])
assert ( assert (
'[{"table": "Gosh", "schema": "CREATE TABLE Gosh (c1 text, c2 text, c3 text)"},\n' '[{"table": "Gosh", "schema": "CREATE TABLE Gosh (c1 text, c2 text, c3 text)"},\n'
@ -191,7 +188,6 @@ def test_output_table(db_path, options, expected):
for i in range(4) for i in range(4)
] ]
) )
db.close()
result = CliRunner().invoke(cli.cli, ["rows", db_path, "rows"] + options) result = CliRunner().invoke(cli.cli, ["rows", db_path, "rows"] + options)
assert result.exit_code == 0 assert result.exit_code == 0
assert expected == result.output.strip() assert expected == result.output.strip()
@ -247,7 +243,6 @@ def test_create_index(db_path):
CliRunner().invoke(cli.cli, create_index_unique_args + [option]).exit_code CliRunner().invoke(cli.cli, create_index_unique_args + [option]).exit_code
== 0 == 0
) )
db.close()
def test_create_index_analyze(db_path): def test_create_index_analyze(db_path):
@ -627,7 +622,6 @@ def test_optimize(db_path, tables):
) )
db["Gosh"].enable_fts(["c1", "c2", "c3"], fts_version="FTS4") db["Gosh"].enable_fts(["c1", "c2", "c3"], fts_version="FTS4")
db["Gosh2"].enable_fts(["c1", "c2", "c3"], fts_version="FTS5") db["Gosh2"].enable_fts(["c1", "c2", "c3"], fts_version="FTS5")
db.close()
size_before_optimize = os.stat(db_path).st_size size_before_optimize = os.stat(db_path).st_size
result = CliRunner().invoke(cli.cli, ["optimize", db_path] + tables) result = CliRunner().invoke(cli.cli, ["optimize", db_path] + tables)
assert result.exit_code == 0 assert result.exit_code == 0
@ -1456,7 +1450,6 @@ def test_drop_table_error():
with runner.isolated_filesystem(): with runner.isolated_filesystem():
db = Database("test.db") db = Database("test.db")
db["t"].create({"pk": int}, pk="pk") db["t"].create({"pk": int}, pk="pk")
db.close()
result = runner.invoke( result = runner.invoke(
cli.cli, cli.cli,
[ [
@ -1481,7 +1474,6 @@ def test_drop_view():
db = Database("test.db") db = Database("test.db")
db.create_view("hello", "select 1") db.create_view("hello", "select 1")
assert "hello" in db.view_names() assert "hello" in db.view_names()
db.close()
result = runner.invoke( result = runner.invoke(
cli.cli, cli.cli,
[ [
@ -1491,9 +1483,7 @@ def test_drop_view():
], ],
) )
assert result.exit_code == 0 assert result.exit_code == 0
db = Database("test.db")
assert "hello" not in db.view_names() assert "hello" not in db.view_names()
db.close()
def test_drop_view_error(): def test_drop_view_error():
@ -1501,7 +1491,6 @@ def test_drop_view_error():
with runner.isolated_filesystem(): with runner.isolated_filesystem():
db = Database("test.db") db = Database("test.db")
db["t"].create({"pk": int}, pk="pk") db["t"].create({"pk": int}, pk="pk")
db.close()
result = runner.invoke( result = runner.invoke(
cli.cli, cli.cli,
[ [
@ -1739,13 +1728,10 @@ def test_transform(db_path, args, expected_schema):
defaults={"age": 1}, defaults={"age": 1},
pk="id", pk="id",
) )
db.close()
result = CliRunner().invoke(cli.cli, ["transform", db_path, "dogs"] + args) result = CliRunner().invoke(cli.cli, ["transform", db_path, "dogs"] + args)
print(result.output) print(result.output)
assert result.exit_code == 0 assert result.exit_code == 0
db = Database(db_path)
schema = db["dogs"].schema schema = db["dogs"].schema
db.close()
assert schema == expected_schema assert schema == expected_schema

View file

@ -335,4 +335,3 @@ def test_memory_return_db(tmpdir):
db = ctx.invoke(cli.commands["memory"], paths=(path,), return_db=True) db = ctx.invoke(cli.commands["memory"], paths=(path,), return_db=True)
assert db.table_names() == ["dogs"] assert db.table_names() == ["dogs"]
db.close()

View file

@ -6,39 +6,39 @@ import sqlite_utils
# SQLite integers are -(2^63) to 2^63 - 1 # SQLite integers are -(2^63) to 2^63 - 1
@given(st.integers(-9223372036854775808, 9223372036854775807)) @given(st.integers(-9223372036854775808, 9223372036854775807))
def test_roundtrip_integers(integer): def test_roundtrip_integers(integer):
with sqlite_utils.Database(memory=True) as db: db = sqlite_utils.Database(memory=True)
row = { row = {
"integer": integer, "integer": integer,
} }
db["test"].insert(row) db["test"].insert(row)
assert list(db["test"].rows) == [row] assert list(db["test"].rows) == [row]
@given(st.text()) @given(st.text())
def test_roundtrip_text(text): def test_roundtrip_text(text):
with sqlite_utils.Database(memory=True) as db: db = sqlite_utils.Database(memory=True)
row = { row = {
"text": text, "text": text,
} }
db["test"].insert(row) db["test"].insert(row)
assert list(db["test"].rows) == [row] assert list(db["test"].rows) == [row]
@given(st.binary(max_size=1024 * 1024)) @given(st.binary(max_size=1024 * 1024))
def test_roundtrip_binary(binary): def test_roundtrip_binary(binary):
with sqlite_utils.Database(memory=True) as db: db = sqlite_utils.Database(memory=True)
row = { row = {
"binary": binary, "binary": binary,
} }
db["test"].insert(row) db["test"].insert(row)
assert list(db["test"].rows) == [row] assert list(db["test"].rows) == [row]
@given(st.floats(allow_nan=False)) @given(st.floats(allow_nan=False))
def test_roundtrip_floats(floats): def test_roundtrip_floats(floats):
with sqlite_utils.Database(memory=True) as db: db = sqlite_utils.Database(memory=True)
row = { row = {
"floats": floats, "floats": floats,
} }
db["test"].insert(row) db["test"].insert(row)
assert list(db["test"].rows) == [row] assert list(db["test"].rows) == [row]