From c2a17744095553c7fcc4706cc5da1d20045a299e Mon Sep 17 00:00:00 2001
From: Simon Willison <swillison@gmail.com>
Date: Mon, 6 Jul 2026 22:02:41 -0700
Subject: [PATCH] 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>
---
 tests/test_atomic.py | 53 ++++++++++++++++++++++++++++++++++++++++++++
 tests/test_rows.py   | 16 ++++++++-----
 2 files changed, 64 insertions(+), 5 deletions(-)

diff --git a/tests/test_atomic.py b/tests/test_atomic.py
index 0d25b84..c3fd02f 100644
--- a/tests/test_atomic.py
+++ b/tests/test_atomic.py
@@ -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]
 
 
+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
@@ -327,3 +337,46 @@ def test_query_returning_commits_after_iteration(tmpdir):
     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
diff --git a/tests/test_rows.py b/tests/test_rows.py
index 46417ef..46d4f53 100644
--- a/tests/test_rows.py
+++ b/tests/test_rows.py
@@ -115,15 +115,21 @@ def test_rows_where_duplicate_select_columns_are_deduped(fresh_db):
 
 def test_pks_and_rows_where_view(fresh_db):
     # pks_and_rows_where() lives on Queryable so views expose it, but
-    # SQLite views have no rowid - it has always failed with an
-    # OperationalError from the generated SQL. Guard against it failing
-    # earlier with an AttributeError from View lacking Table properties
+    # SQLite views have no rowid. Modern SQLite (3.36+) raises an
+    # OperationalError from the generated SQL; older versions returned
+    # 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
 
     fresh_db["dogs"].insert({"id": 1, "name": "Cleo"}, pk="id")
     fresh_db.create_view("dog_names", "select name from dogs")
-    with pytest.raises(sqlite3.OperationalError):
-        list(fresh_db["dog_names"].pks_and_rows_where())
+    try:
+        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):