mirror of
https://github.com/simonw/sqlite-utils.git
synced 2026-10-07 18:07:03 +02:00
Fix two tests that assumed modern SQLite behavior
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>
This commit is contained in:
parent
93640a7dde
commit
c2a1774409
2 changed files with 64 additions and 5 deletions
|
|
@ -258,11 +258,21 @@ def test_execute_comment_prefixed_begin_leaves_transaction_open(fresh_db):
|
||||||
assert [r["id"] for r in fresh_db["t"].rows] == [1]
|
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"])
|
@pytest.mark.parametrize("begin_sql", ["; begin", "\ufeffbegin"])
|
||||||
def test_execute_prefixed_begin_leaves_transaction_open(fresh_db, begin_sql):
|
def test_execute_prefixed_begin_leaves_transaction_open(fresh_db, begin_sql):
|
||||||
# sqlite3 tolerates empty statements and a UTF-8 BOM before the first
|
# sqlite3 tolerates empty statements and a UTF-8 BOM before the first
|
||||||
# real token, so a BEGIN behind either must not be auto-committed
|
# real token, so a BEGIN behind either must not be auto-committed
|
||||||
# out from under the caller
|
# 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["t"].insert({"id": 1}, pk="id")
|
||||||
fresh_db.execute(begin_sql)
|
fresh_db.execute(begin_sql)
|
||||||
assert fresh_db.conn.in_transaction
|
assert fresh_db.conn.in_transaction
|
||||||
|
|
@ -327,3 +337,46 @@ def test_query_returning_commits_after_iteration(tmpdir):
|
||||||
assert other.execute("select count(*) from t").fetchone()[0] == 2
|
assert other.execute("select count(*) from t").fetchone()[0] == 2
|
||||||
other.close()
|
other.close()
|
||||||
db.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
|
||||||
|
|
|
||||||
|
|
@ -115,15 +115,21 @@ def test_rows_where_duplicate_select_columns_are_deduped(fresh_db):
|
||||||
|
|
||||||
def test_pks_and_rows_where_view(fresh_db):
|
def test_pks_and_rows_where_view(fresh_db):
|
||||||
# pks_and_rows_where() lives on Queryable so views expose it, but
|
# pks_and_rows_where() lives on Queryable so views expose it, but
|
||||||
# SQLite views have no rowid - it has always failed with an
|
# SQLite views have no rowid. Modern SQLite (3.36+) raises an
|
||||||
# OperationalError from the generated SQL. Guard against it failing
|
# OperationalError from the generated SQL; older versions returned
|
||||||
# earlier with an AttributeError from View lacking Table properties
|
# NULL for a view's rowid. Either way it must not fail earlier with
|
||||||
|
# an AttributeError from View lacking Table-only properties
|
||||||
from sqlite_utils.utils import sqlite3
|
from sqlite_utils.utils import sqlite3
|
||||||
|
|
||||||
fresh_db["dogs"].insert({"id": 1, "name": "Cleo"}, pk="id")
|
fresh_db["dogs"].insert({"id": 1, "name": "Cleo"}, pk="id")
|
||||||
fresh_db.create_view("dog_names", "select name from dogs")
|
fresh_db.create_view("dog_names", "select name from dogs")
|
||||||
with pytest.raises(sqlite3.OperationalError):
|
try:
|
||||||
list(fresh_db["dog_names"].pks_and_rows_where())
|
result = list(fresh_db["dog_names"].pks_and_rows_where())
|
||||||
|
except sqlite3.OperationalError:
|
||||||
|
pass # SQLite 3.36+: no such column: rowid
|
||||||
|
else:
|
||||||
|
# Older SQLite returns NULL rowids for views
|
||||||
|
assert result == [(None, {"rowid": None, "name": "Cleo"})]
|
||||||
|
|
||||||
|
|
||||||
def test_pks_and_rows_where_compound_pk_declaration_order(fresh_db):
|
def test_pks_and_rows_where_compound_pk_declaration_order(fresh_db):
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue