Enforce strict catalog verification on reads

Author: TonisOrmisson

src/openstatspec/sql/wide.py

@@ -1000,7 +1000,7 @@ def read_wide_dataset(
     engine = create_engine(database_url)
     normative = normative_catalog(MetaData())
     with engine.connect() as connection:
-        _verify_normative_catalog(connection, normative)
+        require_verified_catalog(connection)
         dataset_row = _resolve_normative_dataset(connection, normative, dataset_id)
         core_id = str(dataset_row["dataset_id"])
         data_table = Table(
@@ -1331,7 +1331,7 @@ def read_fidelity_events(
     engine = create_engine(database_url)
     normative = normative_catalog(MetaData())
     with engine.connect() as connection:
-        _verify_normative_catalog(connection, normative)
+        require_verified_catalog(connection)
         dataset = _resolve_normative_dataset(connection, normative, dataset_id)
         events = connection.execute(
             select(normative.fidelity_event)
@@ -1367,6 +1367,7 @@ def _bound_export_audit_transaction(
         audit_relations={normative.operation.name, normative.fidelity_event.name},
         phase=phase,
     ) as connection:
+        require_verified_catalog(connection)
         yield connection, normative
 
 
@@ -1464,7 +1465,7 @@ def read_export_operation_state(
     )
     normative = normative_catalog(MetaData())
     with create_engine(database_url).connect() as connection:
-        _verify_normative_catalog(connection, normative)
+        require_verified_catalog(connection)
         row = _export_operation_row(connection, normative, operation_id)
     classification = {
         "started": "running",

tests/test_strict_catalog_review.py

@@ -75,6 +75,26 @@ def test_existing_normative_shape_drift_is_rejected(tmp_path):
         openstatspec.initialize_catalog(database_url=database)
 
 
+def test_read_and_export_reject_obsolete_catalog_relations(tmp_path):
+    path = tmp_path / "obsolete-read.sqlite"
+    database = f"sqlite:///{path}"
+    _create(database, "sample")
+    connection = sqlite3.connect(path)
+    connection.execute("create table dataset_catalog (dataset_id text primary key)")
+    connection.commit()
+    connection.close()
+
+    with pytest.raises(UnsupportedOperationError, match="obsolete"):
+        wide.read_wide_dataset(database_url=database, dataset_id="sample")
+    with pytest.raises(UnsupportedOperationError, match="obsolete"):
+        wide.record_export_operation(
+            database_url=database,
+            dataset_id="sample",
+            destination="sample.sav",
+            allowed_fidelity_events=(),
+        )
+
+
 def test_verified_workflow_profile_remains_catalog_owned(tmp_path):
     path = tmp_path / "workflow-owned.sqlite"
     database = f"sqlite:///{path}"