"""Execute a disposable file-backed SQLite workload; never inspect a user database.""" from __future__ import annotations import argparse import hashlib import json from pathlib import Path import platform import sqlite3 import tempfile ROOT = Path(__file__).resolve().parent SELECT = "SELECT id, label, amount_cents FROM ledger ORDER BY id" INSERT = "INSERT INTO ledger (id, label, amount_cents) VALUES (?, ?, ?)" MODES = ("broken", "fixed", "begin-too-late", "isolation-none", "commit-control", "unreleased-control", "autocommit-false") class ForcedOuterFailure(RuntimeError): pass def check(result, name, condition): if not condition: raise AssertionError(name) result["checks"].append(name) def run(mode): fixture = json.loads((ROOT / "fixture.json").read_text(encoding="utf-8")) result = { "mode": mode, "status": "running", "phase": "initialize", "checks": [], "versions": {"python": platform.python_version(), "sqlite": sqlite3.sqlite_version}, "sql_trace": [], "reopen_sql_trace": [], "transaction_states": [], } # Exposed trace entries are actual sqlite3.set_trace_callback events. Python # context-manager entry and forced exceptions are recorded separately. with tempfile.TemporaryDirectory(prefix="lab-savepoint-") as temporary: path = Path(temporary) / "synthetic.sqlite3" initializer = sqlite3.connect(path, isolation_level=None) try: initializer.execute("CREATE TABLE ledger (id INTEGER PRIMARY KEY, label TEXT NOT NULL, amount_cents INTEGER)") initializer.executemany(INSERT, fixture["seed"]) finally: initializer.close() kwargs = {"isolation_level": None if mode == "isolation-none" else ""} if hasattr(sqlite3, "LEGACY_TRANSACTION_CONTROL"): kwargs["autocommit"] = sqlite3.LEGACY_TRANSACTION_CONTROL connection = sqlite3.connect(path, **kwargs) connection.set_trace_callback(result["sql_trace"].append) def state(point): result["transaction_states"].append({"point": point, "in_transaction": connection.in_transaction}) return connection.in_transaction try: check(result, "fresh_connection_has_no_pending_transaction", not state("fresh_connection")) if mode == "autocommit-false": if not hasattr(connection, "autocommit"): result.update(status="unsupported", phase="capability_check", reason="Connection.autocommit requires supported Python runtime (added in 3.12)") return result, 3 # Set on a fresh connection after registering the trace so the # driver-emitted opening BEGIN is actually observable. connection.autocommit = False result["configuration"] = { "isolation_level": connection.isolation_level, "autocommit": getattr(connection, "autocommit", "unavailable"), "explicit_begin_before_savepoint": mode in ("fixed", "commit-control"), "autocommit_false_application": "fresh Connection.autocommit = False after trace registration" if mode == "autocommit-false" else None, } result["phase"] = "outer_context" forced = None try: with connection: state("python_context_entered") check(result, "seed_rows_match_public_fixture", [list(row) for row in connection.execute(SELECT)] == fixture["seed"]) if mode in ("fixed", "commit-control"): connection.execute("BEGIN") before_savepoint = state("before_savepoint") result["outer_begin_preceded_savepoint"] = before_savepoint if mode in ("fixed", "commit-control", "autocommit-false"): check(result, "real_outer_transaction_active_before_savepoint", before_savepoint) else: check(result, "python_context_and_select_did_not_open_outer_transaction", not before_savepoint) connection.execute("SAVEPOINT nested_work") check(result, "savepoint_opens_or_joins_real_transaction", state("after_savepoint")) connection.execute(INSERT, fixture["savepoint_row"]) check(result, "savepoint_write_visible_inside_connection", [list(row) for row in connection.execute(SELECT)] == fixture["seed"] + [fixture["savepoint_row"]]) if mode != "unreleased-control": connection.execute("RELEASE SAVEPOINT nested_work") after_release = state("after_release") check(result, "release_retains_outer_transaction" if before_savepoint else "outermost_release_ends_transaction", after_release == before_savepoint) if mode == "begin-too-late": connection.execute("BEGIN") connection.execute(INSERT, fixture["later_row"]) state("after_later_write") if mode != "commit-control": raise ForcedOuterFailure("LAB forced outer failure after nested work") except ForcedOuterFailure as error: forced = {"type": type(error).__name__, "message": str(error)} state("after_context_exit") result["forced_exception"] = forced check(result, "intentional_success_reached" if mode == "commit-control" else "forced_outer_failure_caught", forced is None if mode == "commit-control" else forced is not None) except Exception as error: result.update(status="check_failed", error_type=type(error).__name__, error=str(error)) return result, 2 finally: connection.close() result["phase"] = "reopen_database" # mode=ro forbids silently making a fresh empty database at the wrong path. reopened = sqlite3.connect(path.as_uri() + "?mode=ro", uri=True) reopened.set_trace_callback(result["reopen_sql_trace"].append) try: persisted = [list(row) for row in reopened.execute(SELECT)] integrity = reopened.execute("PRAGMA integrity_check").fetchone()[0] finally: reopened.close() expected = fixture["seed"] + [fixture["savepoint_row"], fixture["later_row"]] if mode == "commit-control" else fixture["seed"] result["summary"] = { "persisted_rows_after_reconnect": persisted, "expected_rows_after_reconnect": expected, "persisted_row_count": len(persisted), "integrity_check": integrity, "database_sha256_after_close": hashlib.sha256(path.read_bytes()).hexdigest(), "database_lifecycle": "new disposable file; writer closed; separate read-only connection queried; temporary directory removed", "transaction_sql": [sql for sql in result["sql_trace"] if sql.split()[0] in ("BEGIN", "SAVEPOINT", "RELEASE", "ROLLBACK", "COMMIT")], } try: check(result, "reopened_database_integrity_is_ok", integrity == "ok") check(result, "seed_null_unicode_and_values_preserved", persisted[:len(fixture["seed"])] == fixture["seed"]) check(result, "reopened_rows_match_independent_fixture_oracle", persisted == expected) if mode in ("fixed", "autocommit-false"): tx = result["summary"]["transaction_sql"] check(result, "trace_confirms_begin_savepoint_release_rollback_order", tx[0].startswith("BEGIN") and tx[1:4] == ["SAVEPOINT nested_work", "RELEASE SAVEPOINT nested_work", "ROLLBACK"]) check(result, "both_attempted_writes_absent_after_outer_rollback", fixture["savepoint_row"] not in persisted and fixture["later_row"] not in persisted) result.update(status="passed", phase="complete") return result, 0 except AssertionError as error: result.update(status="check_failed", error_type="AssertionError", error=str(error)) return result, 2 if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("mode", choices=MODES) args = parser.parse_args() result, code = run(args.mode) print(json.dumps(result, ensure_ascii=False, sort_keys=True)) raise SystemExit(code)