Harden persistent and bound catalog operations

Author: TonisOrmisson

src/openstatspec/sql/database_urls.py

@@ -0,0 +1,22 @@
+"""Database URL invariants shared by persistent catalog operations."""
+
+from sqlalchemy.engine import make_url
+
+from ..core import UnsupportedOperationError
+
+
+def require_persistent_database_url(database_url: str) -> None:
+    """Reject SQLite URLs whose catalog disappears with a connection or engine."""
+    parsed_url = make_url(database_url)
+    if parsed_url.get_backend_name() != "sqlite":
+        return
+    database = parsed_url.database or ""
+    mode = str(parsed_url.query.get("mode", "")).lower()
+    if (
+        database in {"", ":memory:"}
+        or database.lower() == "file::memory:"
+        or mode == "memory"
+    ):
+        raise UnsupportedOperationError(
+            "OpenStatSpec catalogs require a persistent SQLite database URL."
+        )

src/openstatspec/sql/wide.py

@@ -32,6 +32,7 @@
     store_imported_dataset as store_normative_dataset,
 )
 from .catalog_verification import verify_catalog_relations
+from .database_urls import require_persistent_database_url
 
 _IDENTIFIER = re.compile(r"[^a-zA-Z0-9_]+")
 
@@ -669,12 +670,20 @@ def initialize_wide_catalog(
 ) -> dict[str, Any]:
     """Initialize or verify the singular normative OpenStatSpec catalog."""
     validate_connection_url(database_url)
-    profile, _active = effective_profile(
+    require_persistent_database_url(database_url)
+    profile, active = effective_profile(
         database_url, dolt_conformance_source=dolt_conformance_source,
     )
     engine = create_engine(database_url)
     normative = normative_catalog(MetaData())
-    with engine.begin() as connection:
+    audit_relations = {
+        normative.fidelity_event.name,
+        normative.operation.name,
+    }
+    with _bound_catalog_transaction(
+        engine=engine, profile_name=profile.name, active=active,
+        audit_relations=audit_relations, phase="catalog initialization",
+    ) as connection:
         inspector = inspect(connection)
         views = set(inspector.get_view_names())
         if views and not inspector.has_table(normative.catalog_identity.name):
@@ -755,8 +764,9 @@ def create_wide_dataset(
     dolt_conformance_source: Any | None = None,
 ) -> dict[str, Any]:
     del source_table_name, source_created_at, source_modified_at
+    require_persistent_database_url(database_url)
     validate_connection_url(database_url)
-    profile, _active_connection = (
+    profile, active_connection = (
         effective_profile(database_url)
         if dolt_conformance_source is None
         else effective_profile(
@@ -765,7 +775,14 @@ def create_wide_dataset(
     )
     engine = create_engine(database_url)
     normative = normative_catalog(MetaData())
-    with engine.begin() as catalog_connection:
+    audit_relations = {
+        normative.fidelity_event.name,
+        normative.operation.name,
+    }
+    with _bound_catalog_transaction(
+        engine=engine, profile_name=profile.name, active=active_connection,
+        audit_relations=audit_relations, phase="catalog initialization",
+    ) as catalog_connection:
         catalog_existed = inspect(catalog_connection).has_table(
             normative.catalog_identity.name
         )
@@ -827,7 +844,10 @@ def create_wide_dataset(
         with engine.begin() as setup:
             _verify_normative_catalog(setup, normative)
         namespace_owned = True
-        with engine.begin() as connection:
+        with _bound_catalog_transaction(
+            engine=engine, profile_name=profile.name, active=active_connection,
+            audit_relations=audit_relations, phase="import",
+        ) as connection:
             if connection.execute(
                 select(normative.dataset.c.dataset_id).where(or_(
                     normative.dataset.c.dataset_name == dataset_id,
@@ -1333,7 +1353,7 @@ def record_export_operation(
     dolt_conformance_source: Any | None = None,
 ) -> str:
     """Persist a completed export only in the normative audit catalog."""
-    del destination, operation_details
+    operation_details = dict(operation_details or {})
     engine = create_engine(database_url)
     effective_profile(
         database_url, dolt_conformance_source=dolt_conformance_source,
@@ -1354,14 +1374,26 @@ def record_export_operation(
         record_normative_fidelity_events(
             connection, normative, operation_id=operation_id,
             dataset_id=str(dataset["dataset_id"]), direction="export",
-            events=({
-                **event,
-                "severity": event.get("severity", "warning"),
-                "details": {
-                    **event.get("details", {}),
-                    "accepted_by_user": True,
-                },
-            } for event in events),
+            events=(
+                *(
+                    {
+                        **event,
+                        "severity": event.get("severity", "warning"),
+                        "details": {
+                            **event.get("details", {}),
+                            "accepted_by_user": True,
+                        },
+                    }
+                    for event in events
+                ),
+                *(({
+                    "code": "operation-engine-identity",
+                    "detail": "Export engine identity recorded for audit.",
+                    "severity": "info",
+                    "source_item": destination,
+                    "details": operation_details,
+                },) if operation_details else ()),
+            ),
         )
     return operation_id
 

tests/test_atomic_import.py

@@ -284,6 +284,9 @@ def test_dolt_nonfinite_preflight_creates_no_dataset_or_physical_table(
     database_path = tmp_path / "dolt-nonfinite.sqlite"
     database = f"sqlite:///{database_path}"
     monkeypatch.setattr(wide, "effective_profile", lambda _url: (DOLT, {}))
+    monkeypatch.setattr(
+        wide, "_capture_dolt_state", lambda *_args, **_kwargs: None,
+    )
     variables = [{
         "ordinal": 1, "source_name": "value", "physical_name": "value",
         "storage_kind": "numeric", "string_width": None, "label": "",
@@ -320,6 +323,9 @@ def test_empty_namespace_dolt_width_failure_initializes_identity_and_one_audit(
     database_path = tmp_path / "dolt-preflight.sqlite"
     database = f"sqlite:///{database_path}"
     monkeypatch.setattr(wide, "effective_profile", lambda _url: (DOLT, {}))
+    monkeypatch.setattr(
+        wide, "_capture_dolt_state", lambda *_args, **_kwargs: None,
+    )
 
     with pytest.raises(Exception, match="Target capability exceeded"):
         create_wide_dataset(

tests/test_catalog_persistence_review.py

@@ -0,0 +1,98 @@
+"""Regression tests for persistent URLs and operation-bound audit writes."""
+
+import json
+import sqlite3
+from contextlib import contextmanager
+
+import pytest
+
+from openstatspec.core import UnsupportedOperationError
+from openstatspec.sql import wide
+
+
+def _variables():
+    return [{
+        "ordinal": 1,
+        "source_name": "name",
+        "physical_name": "name",
+        "storage_kind": "string",
+        "string_width": 8,
+        "label": "",
+        "format": "A8",
+        "measure": "nominal",
+        "alignment": "left",
+        "display_width": 8,
+        "value_labels": "{}",
+        "missing_ranges": "[]",
+    }]
+
+
+def _create(database_url, dataset_id="sample"):
+    return wide.create_wide_dataset(
+        database_url=database_url,
+        dataset_id=dataset_id,
+        source_name=f"{dataset_id}.sav",
+        source_format="SAV",
+        rows=[{"name": "ok"}],
+        variables=_variables(),
+    )
+
+
+@pytest.mark.parametrize(
+    "database_url",
+    [
+        "sqlite://",
+        "sqlite:///:memory:",
+        "sqlite:///file:shared-memory?mode=memory&uri=true",
+    ],
+)
+def test_import_rejects_ephemeral_sqlite_catalogs(database_url):
+    with pytest.raises(UnsupportedOperationError, match="persistent SQLite"):
+        _create(database_url)
+
+
+def test_import_routes_catalog_and_dataset_mutations_through_binding_guard(
+    tmp_path, monkeypatch,
+):
+    database_url = f"sqlite:///{tmp_path / 'bound.sqlite'}"
+    real_bound_transaction = wide._bound_catalog_transaction
+    phases = []
+
+    @contextmanager
+    def observe(**kwargs):
+        phases.append(kwargs["phase"])
+        with real_bound_transaction(**kwargs) as connection:
+            yield connection
+
+    monkeypatch.setattr(wide, "_bound_catalog_transaction", observe)
+    _create(database_url)
+
+    assert phases == ["catalog initialization", "import"]
+
+
+def test_export_engine_identity_is_persisted_as_non_loss_audit_metadata(tmp_path):
+    path = tmp_path / "export-audit.sqlite"
+    database_url = f"sqlite:///{path}"
+    _create(database_url)
+    engine_identity = {"name": "pyspssio", "commit": "export-commit"}
+
+    operation_id = wide.record_export_operation(
+        database_url=database_url,
+        dataset_id="sample",
+        destination="out.sav",
+        allowed_fidelity_events=(),
+        operation_details={"engine": engine_identity},
+    )
+
+    connection = sqlite3.connect(path)
+    row = connection.execute(
+        "select severity, event_code, source_item, detail_json "
+        "from fidelity_event where operation_id = ?",
+        (operation_id,),
+    ).fetchone()
+    connection.close()
+    assert row[:3] == ("info", "operation-engine-identity", "out.sav")
+    assert json.loads(row[3])["engine"] == engine_identity
+    assert wide.read_fidelity_events(
+        database_url=database_url, dataset_id="sample",
+    ) == ()

tests/test_loss_reports.py

@@ -55,7 +55,7 @@ def test_loss_allowed_export_persists_accepted_diagnostics(tmp_path) -> None:
     result = openstatspec.export_sav(database_url=database, dataset_id="accepted", destination=destination, allow_loss=_REQUIRED_ENGINE_LOSS)
 
     connection = sqlite3.connect(database_path)
-    rows = connection.execute("select direction, severity, event_code, detail_json from fidelity_event where operation_id = ? order by event_code", (result["operation_id"],)).fetchall()
+    rows = connection.execute("select direction, severity, event_code, detail_json from fidelity_event where operation_id = ? and severity != 'info' order by event_code", (result["operation_id"],)).fetchall()
     assert [(row[0], row[1], row[2]) for row in rows] == [("export", "warning", code) for code in _REQUIRED_ENGINE_LOSS]
     assert all('"accepted_by_user": true' in row[3] for row in rows)