mirror of
https://github.com/simonw/sqlite-utils.git
synced 2026-07-23 09:24:31 +02:00
bcd9a26560Connections created with sqlite3.connect(autocommit=True) make commit() a documented no-op, and autocommit=False connections are permanently inside a transaction - in both modes every write made by this library appeared to work in-process but was silently discarded when the connection closed. Database() now raises TransactionError for these connections instead of losing data. Tests are skipped on Python versions before 3.12, where the autocommit parameter does not exist. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01UnLnhsH25Nnv7LHhekUfPd
86 lines
2.7 KiB
Python
86 lines
2.7 KiB
Python
from sqlite_utils import Database
|
|
from sqlite_utils.db import TransactionError
|
|
from sqlite_utils.utils import sqlite3
|
|
import pytest
|
|
import sys
|
|
|
|
|
|
def test_recursive_triggers():
|
|
db = Database(memory=True)
|
|
assert db.execute("PRAGMA recursive_triggers").fetchone()[0]
|
|
|
|
|
|
def test_recursive_triggers_off():
|
|
db = Database(memory=True, recursive_triggers=False)
|
|
assert not db.execute("PRAGMA recursive_triggers").fetchone()[0]
|
|
|
|
|
|
def test_memory_name():
|
|
db1 = Database(memory_name="shared")
|
|
db2 = Database(memory_name="shared")
|
|
db1["dogs"].insert({"name": "Cleo"})
|
|
assert list(db2["dogs"].rows) == [{"name": "Cleo"}]
|
|
|
|
|
|
def test_sqlite_version():
|
|
db = Database(memory=True)
|
|
version = db.sqlite_version
|
|
assert isinstance(version, tuple)
|
|
as_string = ".".join(map(str, version))
|
|
actual = next(db.query("select sqlite_version() as v"))["v"]
|
|
assert actual == as_string
|
|
|
|
|
|
def test_database_context_manager(tmpdir):
|
|
path = str(tmpdir / "test.db")
|
|
with Database(path) as db:
|
|
db["t"].insert({"id": 1})
|
|
# A raw write left uncommitted on purpose:
|
|
db.execute("insert into t (id) values (2)")
|
|
# The connection is closed...
|
|
with pytest.raises(sqlite3.ProgrammingError):
|
|
db.execute("select 1")
|
|
# ... and the uncommitted change was rolled back, not committed
|
|
db2 = Database(path)
|
|
assert [r["id"] for r in db2["t"].rows] == [1]
|
|
db2.close()
|
|
|
|
|
|
@pytest.mark.parametrize("memory", [True, False])
|
|
def test_database_close(tmpdir, memory):
|
|
if memory:
|
|
db = Database(memory=True)
|
|
else:
|
|
db = Database(str(tmpdir / "test.db"))
|
|
assert db.execute("select 1 + 1").fetchone()[0] == 2
|
|
db.close()
|
|
with pytest.raises(sqlite3.ProgrammingError):
|
|
db.execute("select 1 + 1")
|
|
|
|
|
|
@pytest.mark.skipif(
|
|
sys.version_info < (3, 12),
|
|
reason="sqlite3.connect(autocommit=) requires Python 3.12",
|
|
)
|
|
@pytest.mark.parametrize("autocommit", [True, False])
|
|
def test_autocommit_connections_are_rejected(tmpdir, autocommit):
|
|
# These connection modes break commit()/rollback() in ways that
|
|
# silently lose data, so the constructor refuses them
|
|
conn = sqlite3.connect(str(tmpdir / "test.db"), autocommit=autocommit)
|
|
with pytest.raises(TransactionError):
|
|
Database(conn)
|
|
conn.close()
|
|
|
|
|
|
@pytest.mark.skipif(
|
|
sys.version_info < (3, 12),
|
|
reason="sqlite3.LEGACY_TRANSACTION_CONTROL requires Python 3.12",
|
|
)
|
|
def test_legacy_transaction_control_connection_is_accepted(tmpdir):
|
|
conn = sqlite3.connect(
|
|
str(tmpdir / "test.db"), autocommit=sqlite3.LEGACY_TRANSACTION_CONTROL
|
|
)
|
|
db = Database(conn)
|
|
db["t"].insert({"id": 1}, pk="id")
|
|
assert [r["id"] for r in db["t"].rows] == [1]
|
|
db.close()
|