mirror of
https://github.com/simonw/sqlite-utils.git
synced 2026-07-23 17:34:32 +02:00
c2a1774409
SQLite 3.23.1 rejects a UTF-8 byte order mark before the first token, so the BOM variant of the execute()-prefixed-BEGIN test now skips when the SQLite version does not accept a leading BOM. And versions before 3.36 allowed selecting rowid from a view, returning NULL, rather than raising an error - the pks_and_rows_where() view test now accepts either behavior. Refs https://github.com/simonw/sqlite-utils/issues/769#issuecomment-4900034150 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
382 lines
13 KiB
Python
382 lines
13 KiB
Python
import pytest
|
|
|
|
from sqlite_utils.db import Database, _iter_complete_sql_statements
|
|
from sqlite_utils.utils import sqlite3
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"sql,expected",
|
|
(
|
|
(
|
|
"CREATE TABLE t(id); INSERT INTO t VALUES (1)",
|
|
["CREATE TABLE t(id);", "INSERT INTO t VALUES (1)"],
|
|
),
|
|
(
|
|
"INSERT INTO t VALUES ('a;b');",
|
|
["INSERT INTO t VALUES ('a;b');"],
|
|
),
|
|
(
|
|
"-- comment;\nCREATE TABLE t(id);",
|
|
["-- comment;\nCREATE TABLE t(id);"],
|
|
),
|
|
(
|
|
"""
|
|
CREATE TRIGGER t_ai AFTER INSERT ON t
|
|
BEGIN
|
|
UPDATE t SET value = 'a;b' WHERE id = new.id;
|
|
INSERT INTO log VALUES ('x;y');
|
|
END;
|
|
""",
|
|
[
|
|
"CREATE TRIGGER t_ai AFTER INSERT ON t\n"
|
|
" BEGIN\n"
|
|
" UPDATE t SET value = 'a;b' WHERE id = new.id;\n"
|
|
" INSERT INTO log VALUES ('x;y');\n"
|
|
" END;"
|
|
],
|
|
),
|
|
),
|
|
)
|
|
def test_iter_complete_sql_statements(sql, expected):
|
|
assert list(_iter_complete_sql_statements(sql)) == expected
|
|
|
|
|
|
def test_atomic_commits(fresh_db):
|
|
with fresh_db.atomic():
|
|
fresh_db["dogs"].insert({"id": 1, "name": "Cleo"}, pk="id")
|
|
|
|
assert list(fresh_db["dogs"].rows) == [{"id": 1, "name": "Cleo"}]
|
|
|
|
|
|
def test_atomic_rolls_back(fresh_db):
|
|
with pytest.raises(RuntimeError):
|
|
with fresh_db.atomic():
|
|
fresh_db["dogs"].insert({"id": 1, "name": "Cleo"}, pk="id")
|
|
raise RuntimeError("boom")
|
|
|
|
assert not fresh_db["dogs"].exists()
|
|
|
|
|
|
def test_nested_atomic_rolls_back_to_savepoint(fresh_db):
|
|
fresh_db["dogs"].create({"id": int, "name": str}, pk="id")
|
|
|
|
with fresh_db.atomic():
|
|
fresh_db["dogs"].insert({"id": 1, "name": "Cleo"})
|
|
with pytest.raises(RuntimeError):
|
|
with fresh_db.atomic():
|
|
fresh_db["dogs"].insert({"id": 2, "name": "Pancakes"})
|
|
raise RuntimeError("boom")
|
|
fresh_db["dogs"].insert({"id": 3, "name": "Marnie"})
|
|
|
|
assert list(fresh_db["dogs"].rows) == [
|
|
{"id": 1, "name": "Cleo"},
|
|
{"id": 3, "name": "Marnie"},
|
|
]
|
|
|
|
|
|
def test_outer_atomic_rolls_back_released_savepoint(fresh_db):
|
|
with pytest.raises(RuntimeError):
|
|
with fresh_db.atomic():
|
|
fresh_db["dogs"].insert({"id": 1, "name": "Cleo"}, pk="id")
|
|
with fresh_db.atomic():
|
|
fresh_db["dogs"].insert({"id": 2, "name": "Pancakes"})
|
|
raise RuntimeError("boom")
|
|
|
|
assert not fresh_db["dogs"].exists()
|
|
|
|
|
|
def test_executescript_does_not_commit_open_atomic_block(fresh_db):
|
|
with pytest.raises(RuntimeError):
|
|
with fresh_db.atomic():
|
|
fresh_db.executescript("""
|
|
CREATE TABLE dogs(id INTEGER PRIMARY KEY, name TEXT);
|
|
CREATE TRIGGER dogs_ai AFTER INSERT ON dogs
|
|
BEGIN
|
|
UPDATE dogs SET name = upper(new.name) || '; updated' WHERE id = new.id;
|
|
END;
|
|
-- This comment has a semicolon;
|
|
INSERT INTO dogs VALUES (1, 'Cleo; the first');
|
|
""")
|
|
raise RuntimeError("boom")
|
|
|
|
assert not fresh_db["dogs"].exists()
|
|
|
|
|
|
def test_transform_does_not_commit_open_atomic_block(fresh_db):
|
|
fresh_db["dogs"].insert({"id": 1, "name": "Cleo", "age": "5"}, pk="id")
|
|
|
|
with pytest.raises(RuntimeError):
|
|
with fresh_db.atomic():
|
|
fresh_db["dogs"].insert({"id": 2, "name": "Pancakes", "age": "6"})
|
|
fresh_db["dogs"].transform(rename={"age": "dog_age"})
|
|
raise RuntimeError("boom")
|
|
|
|
assert (
|
|
fresh_db["dogs"].schema
|
|
== 'CREATE TABLE "dogs" (\n "id" INTEGER PRIMARY KEY,\n "name" TEXT,\n "age" TEXT\n)'
|
|
)
|
|
assert list(fresh_db["dogs"].rows) == [
|
|
{"id": 1, "name": "Cleo", "age": "5"},
|
|
]
|
|
|
|
|
|
def test_transform_parent_table_with_foreign_keys_in_atomic(fresh_db):
|
|
fresh_db.conn.execute("PRAGMA foreign_keys=ON")
|
|
fresh_db["authors"].insert({"id": 1, "name": "Tina"}, pk="id")
|
|
fresh_db["books"].insert(
|
|
{"id": 1, "title": "Book", "author_id": 1},
|
|
pk="id",
|
|
foreign_keys={"author_id"},
|
|
)
|
|
|
|
with fresh_db.atomic():
|
|
fresh_db["authors"].transform(rename={"name": "full_name"})
|
|
assert fresh_db.conn.execute("PRAGMA foreign_keys").fetchone()[0]
|
|
|
|
assert (
|
|
fresh_db["authors"].schema
|
|
== 'CREATE TABLE "authors" (\n "id" INTEGER PRIMARY KEY,\n "full_name" TEXT\n)'
|
|
)
|
|
assert fresh_db.execute("PRAGMA foreign_key_check").fetchall() == []
|
|
|
|
|
|
def test_transform_parent_table_with_foreign_keys_rolls_back(fresh_db):
|
|
fresh_db.conn.execute("PRAGMA foreign_keys=ON")
|
|
fresh_db["authors"].insert({"id": 1, "name": "Tina"}, pk="id")
|
|
fresh_db["books"].insert(
|
|
{"id": 1, "title": "Book", "author_id": 1},
|
|
pk="id",
|
|
foreign_keys={"author_id"},
|
|
)
|
|
|
|
with pytest.raises(RuntimeError):
|
|
with fresh_db.atomic():
|
|
fresh_db["authors"].transform(rename={"name": "full_name"})
|
|
raise RuntimeError("boom")
|
|
|
|
assert (
|
|
fresh_db["authors"].schema
|
|
== 'CREATE TABLE "authors" (\n "id" INTEGER PRIMARY KEY,\n "name" TEXT\n)'
|
|
)
|
|
assert fresh_db.conn.execute("PRAGMA foreign_keys").fetchone()[0]
|
|
assert fresh_db.execute("PRAGMA foreign_key_check").fetchall() == []
|
|
|
|
|
|
def test_transform_detects_foreign_key_check_violations(fresh_db):
|
|
fresh_db.conn.execute("PRAGMA foreign_keys=ON")
|
|
fresh_db["authors"].insert({"id": 1, "name": "Tina"}, pk="id")
|
|
fresh_db["books"].insert({"id": 1, "author_id": 2}, pk="id")
|
|
|
|
with pytest.raises(sqlite3.IntegrityError):
|
|
fresh_db["books"].transform(add_foreign_keys=(("author_id", "authors", "id"),))
|
|
|
|
assert fresh_db["books"].foreign_keys == []
|
|
assert fresh_db.conn.execute("PRAGMA foreign_keys").fetchone()[0]
|
|
|
|
|
|
def test_atomic_inside_manual_transaction_uses_savepoint(fresh_db):
|
|
fresh_db["t"].insert({"id": 1}, pk="id")
|
|
fresh_db.execute("begin")
|
|
with fresh_db.atomic():
|
|
fresh_db["t"].insert({"id": 2}, pk="id")
|
|
# Nothing is committed until the user's own transaction commits
|
|
assert fresh_db.conn.in_transaction
|
|
fresh_db.rollback()
|
|
assert [r["id"] for r in fresh_db["t"].rows] == [1]
|
|
# And with a commit instead, the atomic block's writes persist
|
|
fresh_db.execute("begin")
|
|
with fresh_db.atomic():
|
|
fresh_db["t"].insert({"id": 3}, pk="id")
|
|
fresh_db.commit()
|
|
assert [r["id"] for r in fresh_db["t"].rows] == [1, 3]
|
|
|
|
|
|
def test_begin_commit_rollback(tmpdir):
|
|
path = str(tmpdir / "test.db")
|
|
db = Database(path)
|
|
db["t"].insert({"id": 1}, pk="id")
|
|
db.begin()
|
|
db["t"].insert({"id": 2}, pk="id")
|
|
assert db.conn.in_transaction
|
|
db.rollback()
|
|
assert not db.conn.in_transaction
|
|
assert [r["id"] for r in db["t"].rows] == [1]
|
|
db.begin()
|
|
db["t"].insert({"id": 3}, pk="id")
|
|
db.commit()
|
|
db.close()
|
|
db2 = Database(path)
|
|
assert [r["id"] for r in db2["t"].rows] == [1, 3]
|
|
db2.close()
|
|
|
|
|
|
def test_begin_inside_transaction_errors(fresh_db):
|
|
fresh_db.begin()
|
|
with pytest.raises(sqlite3.OperationalError):
|
|
fresh_db.begin()
|
|
fresh_db.rollback()
|
|
|
|
|
|
def test_commit_and_rollback_without_transaction_are_noops(fresh_db):
|
|
fresh_db.commit()
|
|
fresh_db.rollback()
|
|
assert not fresh_db.conn.in_transaction
|
|
|
|
|
|
def test_execute_write_commits_immediately(tmpdir):
|
|
path = str(tmpdir / "test.db")
|
|
db = Database(path)
|
|
db["t"].insert({"id": 1}, pk="id")
|
|
db.execute("insert into t (id) values (2)")
|
|
# No implicit transaction is left open
|
|
assert not db.conn.in_transaction
|
|
# A completely separate connection sees the row straight away
|
|
other = sqlite3.connect(path)
|
|
assert other.execute("select count(*) from t").fetchone()[0] == 2
|
|
other.close()
|
|
db.close()
|
|
|
|
|
|
def test_execute_write_respects_explicit_transaction(fresh_db):
|
|
fresh_db["t"].insert({"id": 1}, pk="id")
|
|
fresh_db.begin()
|
|
fresh_db.execute("insert into t (id) values (2)")
|
|
# Still inside the explicit transaction - not committed
|
|
assert fresh_db.conn.in_transaction
|
|
fresh_db.rollback()
|
|
assert [r["id"] for r in fresh_db["t"].rows] == [1]
|
|
|
|
|
|
def test_execute_comment_prefixed_begin_leaves_transaction_open(fresh_db):
|
|
# A BEGIN hidden behind a leading comment must not be auto-committed
|
|
# out from under the caller
|
|
fresh_db["t"].insert({"id": 1}, pk="id")
|
|
fresh_db.execute("-- start a transaction\nbegin")
|
|
assert fresh_db.conn.in_transaction
|
|
fresh_db.execute("insert into t (id) values (2)")
|
|
fresh_db.rollback()
|
|
assert [r["id"] for r in fresh_db["t"].rows] == [1]
|
|
|
|
|
|
def _sqlite_accepts_bom():
|
|
try:
|
|
sqlite3.connect(":memory:").execute("\ufeffselect 1")
|
|
return True
|
|
except sqlite3.OperationalError:
|
|
return False
|
|
|
|
|
|
@pytest.mark.parametrize("begin_sql", ["; begin", "\ufeffbegin"])
|
|
def test_execute_prefixed_begin_leaves_transaction_open(fresh_db, begin_sql):
|
|
# sqlite3 tolerates empty statements and a UTF-8 BOM before the first
|
|
# real token, so a BEGIN behind either must not be auto-committed
|
|
# out from under the caller
|
|
if begin_sql.startswith("\ufeff") and not _sqlite_accepts_bom():
|
|
pytest.skip("This SQLite version rejects a leading byte order mark")
|
|
fresh_db["t"].insert({"id": 1}, pk="id")
|
|
fresh_db.execute(begin_sql)
|
|
assert fresh_db.conn.in_transaction
|
|
fresh_db.execute("insert into t (id) values (2)")
|
|
fresh_db.rollback()
|
|
assert [r["id"] for r in fresh_db["t"].rows] == [1]
|
|
|
|
|
|
def test_execute_failed_write_rolls_back_implicit_transaction(tmpdir):
|
|
# A failed write must not leave the driver's implicit transaction open -
|
|
# that would silently disable auto-commit for every subsequent write
|
|
path = str(tmpdir / "test.db")
|
|
db = Database(path)
|
|
db["t"].insert({"id": 1}, pk="id")
|
|
with pytest.raises(sqlite3.IntegrityError):
|
|
db.execute("insert into t (id) values (1)")
|
|
assert not db.conn.in_transaction
|
|
# Subsequent writes commit as normal and survive closing the connection
|
|
db["other"].insert({"id": 2})
|
|
db.close()
|
|
db2 = Database(path)
|
|
assert db2["other"].exists()
|
|
db2.close()
|
|
|
|
|
|
def test_execute_failed_write_preserves_explicit_transaction(fresh_db):
|
|
# A failed write inside an explicit transaction must not roll back
|
|
# the caller's earlier work - only the caller decides that
|
|
fresh_db["t"].insert({"id": 1}, pk="id")
|
|
fresh_db.begin()
|
|
fresh_db.execute("insert into t (id) values (2)")
|
|
with pytest.raises(sqlite3.IntegrityError):
|
|
fresh_db.execute("insert into t (id) values (1)")
|
|
assert fresh_db.conn.in_transaction
|
|
fresh_db.commit()
|
|
assert [r["id"] for r in fresh_db["t"].rows] == [1, 2]
|
|
|
|
|
|
def test_execute_failed_write_inside_atomic_preserves_block(fresh_db):
|
|
# A caught failure inside an atomic() block must leave the block's
|
|
# transaction open so its other work still commits
|
|
fresh_db["t"].insert({"id": 1}, pk="id")
|
|
with fresh_db.atomic():
|
|
fresh_db.execute("insert into t (id) values (2)")
|
|
with pytest.raises(sqlite3.IntegrityError):
|
|
fresh_db.execute("insert into t (id) values (1)")
|
|
assert [r["id"] for r in fresh_db["t"].rows] == [1, 2]
|
|
|
|
|
|
def test_query_returning_commits_after_iteration(tmpdir):
|
|
if sqlite3.sqlite_version_info < (3, 35, 0):
|
|
import pytest as _pytest
|
|
|
|
_pytest.skip("RETURNING requires SQLite 3.35.0 or higher")
|
|
path = str(tmpdir / "test.db")
|
|
db = Database(path)
|
|
db["t"].insert({"id": 1}, pk="id")
|
|
rows = list(db.query("insert into t (id) values (2) returning id"))
|
|
assert rows == [{"id": 2}]
|
|
assert not db.conn.in_transaction
|
|
other = sqlite3.connect(path)
|
|
assert other.execute("select count(*) from t").fetchone()[0] == 2
|
|
other.close()
|
|
db.close()
|
|
|
|
|
|
TRIGGER_SQL = """
|
|
create trigger no_bad before insert on t
|
|
when new.v = 'bad'
|
|
begin
|
|
select raise(rollback, 'trigger says no');
|
|
end
|
|
"""
|
|
|
|
|
|
def test_atomic_preserves_error_from_transaction_destroying_trigger(fresh_db):
|
|
# RAISE(ROLLBACK) rolls back the whole transaction and destroys every
|
|
# savepoint - atomic()'s cleanup must not mask the IntegrityError
|
|
# with "cannot rollback - no transaction is active"
|
|
fresh_db.execute("create table t (id integer primary key, v text)")
|
|
fresh_db.execute(TRIGGER_SQL)
|
|
with pytest.raises(sqlite3.IntegrityError, match="trigger says no"):
|
|
with fresh_db.atomic():
|
|
fresh_db.execute("insert into t (v) values ('bad')")
|
|
assert not fresh_db.conn.in_transaction
|
|
|
|
|
|
def test_nested_atomic_preserves_error_from_transaction_destroying_trigger(
|
|
fresh_db,
|
|
):
|
|
# The nested savepoint branch previously raised
|
|
# "no such savepoint" from ROLLBACK TO SAVEPOINT
|
|
fresh_db.execute("create table t (id integer primary key, v text)")
|
|
fresh_db.execute(TRIGGER_SQL)
|
|
with pytest.raises(sqlite3.IntegrityError, match="trigger says no"):
|
|
with fresh_db.atomic():
|
|
with fresh_db.atomic():
|
|
fresh_db.execute("insert into t (v) values ('bad')")
|
|
assert not fresh_db.conn.in_transaction
|
|
|
|
|
|
def test_atomic_preserves_error_from_insert_or_rollback(fresh_db):
|
|
fresh_db["t"].insert({"id": 1}, pk="id")
|
|
with pytest.raises(sqlite3.IntegrityError):
|
|
with fresh_db.atomic():
|
|
fresh_db.execute("insert or rollback into t (id) values (1)")
|
|
assert not fresh_db.conn.in_transaction
|