fix: allow guarded production Flow v2 read-only audit

tests/test_blif_flow_v2_production_readonly_audit.py

@@ -0,0 +1,110 @@
+from inspect import getsource
+
+import pytest
+
+import scripts.audit_blif_flow_v2_cutover as audit
+
+
+def classify(**overrides):
+    values = dict(
+        v1_state="INFO_SENT", v1_action="SEND_INFO",
+        v2_state="INQUIRY", v2_action="SEND_INFO",
+        v2_projection_present=True, is_duplicate_representation=False,
+        operational_action="SEND_INFO", operational_precedence="business_transition",
+        operational_queue="do_now", v2_confidence="high", v2_diagnostic_status="clear",
+    )
+    values.update(overrides)
+    return audit.classify_semantic_difference(**values)[0]
+
+
+def test_help_performs_zero_database_work(monkeypatch, capsys):
+    monkeypatch.setattr(audit, "run_audit", lambda **kwargs: pytest.fail("help reached DB audit"))
+    with pytest.raises(SystemExit) as exc:
+        audit.main(["--help"])
+    assert exc.value.code == 0
+    assert "--production-readonly-audit" in capsys.readouterr().out
+
+
+def test_production_requires_explicit_opt_in():
+    with pytest.raises(RuntimeError, match="explicit --production-readonly-audit"):
+        audit.validate_execution(production_readonly_audit=False, database="clientflow",
+                                 transaction_read_only="on", mode="shadow")
+
+
+def test_production_database_must_be_clientflow():
+    with pytest.raises(RuntimeError, match="requires database 'clientflow'"):
+        audit.validate_execution(production_readonly_audit=True, database="clientflow_codex_test",
+                                 transaction_read_only="on", mode="shadow")
+
+
+def test_production_transaction_must_be_read_only():
+    with pytest.raises(RuntimeError, match="transaction_read_only=on"):
+        audit.validate_execution(production_readonly_audit=True, database="clientflow",
+                                 transaction_read_only="off", mode="shadow")
+
+
+def test_production_audit_contains_no_sql_writes():
+    source = (getsource(audit._database_snapshot) + getsource(audit.run_audit)).upper()
+    for verb in ("INSERT ", "UPDATE ", "DELETE ", "CREATE ", "ALTER ", "DROP ", "TRUNCATE "):
+        assert verb not in source
+    assert 'BEGIN READ ONLY' in source
+
+
+@pytest.mark.parametrize("mode", ["shadow", "compare"])
+def test_shadow_and_compare_are_allowed(mode):
+    audit.validate_execution(production_readonly_audit=True, database="clientflow",
+                             transaction_read_only="on", mode=mode)
+
+
+@pytest.mark.parametrize("mode", ["", "off", "authoritative"])
+def test_unsafe_production_modes_are_refused(mode):
+    with pytest.raises(RuntimeError, match="requires BLIF_FLOW_V2_MODE=shadow or compare"):
+        audit.validate_execution(production_readonly_audit=True, database="clientflow",
+                                 transaction_read_only="on", mode=mode)
+
+
+def test_missing_projection_is_reported():
+    assert classify(v2_projection_present=False, v2_state=None, v2_action=None) == "MISSING_PROJECTION"
+
+
+def test_semantically_equivalent_is_classified():
+    assert classify() == "SEMANTICALLY_EQUIVALENT"
+    assert classify(v1_action="NO_ACTION", v2_state="COMPLETED", v2_action=None,
+                    operational_action=None, operational_queue="not_current") == "SEMANTICALLY_EQUIVALENT"
+
+
+def test_operational_override_is_classified():
+    assert classify(v1_action="FOLLOW_UP_CUSTOMER_REVIEW", v2_state="AWAITING_CUSTOMER",
+                    v2_action=None, operational_action="FOLLOW_UP_CUSTOMER_REVIEW",
+                    operational_precedence="due_followup") == "V1_OPERATIONAL_OVERRIDE"
+
+
+def test_v2_correction_is_classified():
+    assert classify(v1_action="SEND_INVOICE", v2_state="PROFORMA_REQUIRED",
+                    v2_action="CREATE_PROFORMA") == "V2_CORRECTS_V1"
+    assert classify(v1_action="REVIEW_RECONSTRUCTED_PROCESS", v2_state="REVIEW_REQUIRED",
+                    v2_action="REVIEW_REQUIRED", is_duplicate_representation=True) == "V2_CORRECTS_V1"
+
+
+def test_legacy_only_is_classified():
+    assert classify(v1_action="CREATE_JASMIN_QUOTE", v2_state="AWAITING_CUSTOMER",
+                    v2_action=None, v2_confidence="medium") == "LEGACY_ONLY"
+
+
+def test_real_conflict_is_classified():
+    assert classify(v1_action="UNMAPPED_ACTION", v2_state="INQUIRY",
+                    v2_action=None, v2_confidence="medium",
+                    v2_diagnostic_status="ambiguous", operational_action=None) == "REAL_CONFLICT"
+
+
+def test_named_case_ids_and_expected_semantics_are_frozen():
+    assert audit.NAMED_IDS == {
+        "INSTALBEIRA": "5c33db95-fab8-477a-bddd-0b9cc8f91302",
+        "PANORAMIC": "fd79b9a1-07e6-4f61-95e8-09eab89c155e",
+        "ENGEXICON": "61f1c955-a372-4ea7-b9b0-b8528d74a141",
+        "CONSTRURECUP": "e3b23ac5-84db-4763-8a31-a684e873032c",
+        "X_MAT_CANONICAL": "dc89a466-db24-401b-bfe9-d47644b2d0c8",
+        "X_MAT_DUPLICATE": "1816a06e-9a69-4a9b-9279-1263156892d3",
+        "RZSOLAR_CANONICAL": "fd221608-e007-4043-a23d-07e0c119a345",
+        "RZSOLAR_DUPLICATE": "434124fb-ac19-4d78-909a-55761d7e8daa",
+    }