mirror of
https://github.com/simonw/sqlite-utils.git
synced 2026-09-18 06:24:23 +02:00
6df6da1a8e
The ensure_autocommit_off() context manager was misleadingly named: it sets isolation_level = None on the underlying sqlite3 connection, which puts the driver INTO autocommit mode (no implicit transactions) - the opposite of what the name and docstring claimed. The behavior was always correct for its call sites, which need autocommit to run PRAGMA statements like journal_mode=wal. This renames the method to ensure_autocommit_on() with an accurate docstring, updates the three internal call sites, adds tests, and documents the rename in the changelog and the 3.x to 4.0 upgrading guide. The old name is removed outright since 4.0 permits breaking changes. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
71 lines
2.4 KiB
Python
71 lines
2.4 KiB
Python
import pytest
|
|
from sqlite_utils import Database
|
|
from sqlite_utils.db import TransactionError
|
|
|
|
|
|
@pytest.fixture
|
|
def db_path_tmpdir(tmpdir):
|
|
path = tmpdir / "test.db"
|
|
db = Database(str(path))
|
|
return db, path, tmpdir
|
|
|
|
|
|
def test_enable_disable_wal(db_path_tmpdir):
|
|
db, path, tmpdir = db_path_tmpdir
|
|
assert len(tmpdir.listdir()) == 1
|
|
assert "delete" == db.journal_mode
|
|
assert "test.db-wal" not in [f.basename for f in tmpdir.listdir()]
|
|
db.enable_wal()
|
|
assert "wal" == db.journal_mode
|
|
db["test"].insert({"foo": "bar"})
|
|
assert "test.db-wal" in [f.basename for f in tmpdir.listdir()]
|
|
db.disable_wal()
|
|
assert "delete" == db.journal_mode
|
|
assert "test.db-wal" not in [f.basename for f in tmpdir.listdir()]
|
|
|
|
|
|
def test_enable_wal_inside_transaction_raises(db_path_tmpdir):
|
|
db, path, tmpdir = db_path_tmpdir
|
|
db["test"].insert({"id": 1}, pk="id")
|
|
with pytest.raises(TransactionError):
|
|
with db.atomic():
|
|
db["test"].insert({"id": 2}, pk="id")
|
|
db.enable_wal()
|
|
# The atomic() block must have rolled back cleanly and the
|
|
# journal mode must be unchanged
|
|
assert db.journal_mode == "delete"
|
|
assert [r["id"] for r in db["test"].rows] == [1]
|
|
|
|
|
|
def test_disable_wal_inside_transaction_raises(db_path_tmpdir):
|
|
db, path, tmpdir = db_path_tmpdir
|
|
db.enable_wal()
|
|
db["test"].insert({"id": 1}, pk="id")
|
|
with pytest.raises(TransactionError):
|
|
with db.atomic():
|
|
db["test"].insert({"id": 2}, pk="id")
|
|
db.disable_wal()
|
|
assert db.journal_mode == "wal"
|
|
assert [r["id"] for r in db["test"].rows] == [1]
|
|
|
|
|
|
def test_ensure_autocommit_on(db_path_tmpdir):
|
|
db, path, tmpdir = db_path_tmpdir
|
|
previous_isolation_level = db.conn.isolation_level
|
|
assert previous_isolation_level is not None
|
|
with db.ensure_autocommit_on():
|
|
# isolation_level of None means driver-level autocommit mode
|
|
assert db.conn.isolation_level is None
|
|
# Restored afterwards
|
|
assert db.conn.isolation_level == previous_isolation_level
|
|
|
|
|
|
def test_enable_wal_noop_inside_transaction_is_allowed(db_path_tmpdir):
|
|
# Calling enable_wal() when WAL is already enabled is a no-op,
|
|
# so it is fine inside a transaction
|
|
db, path, tmpdir = db_path_tmpdir
|
|
db.enable_wal()
|
|
with db.atomic():
|
|
db["test"].insert({"id": 1}, pk="id")
|
|
db.enable_wal()
|
|
assert [r["id"] for r in db["test"].rows] == [1]
|