#!/usr/bin/env python3 """ Red Queen — C5G.1: ACCEPTANCE / EXIT-CRITERIA TEST. Prueft die dokumentierten C5G-Exit-Criteria aus C5_SYNC_ARCHITECTURE_DESIGN.md §28/§29 deterministisch gegen die tatsaechlich implementierten C5A-C5F-Contracts. WICHTIG: Dieser Test erfindet KEINE neuen Anforderungen. Fuer jedes §29-Kriterium wird die echte implementierte Semantik geprueft (State-Machine, Guarantee-Funktionen, Health-Contract, DELETE-Human-Gate, Provenance/Adoption, Secret-Detection) — NICHT bloss Strings, Dateiexistenz oder hardcodierte PASS-Werte. Jeder Test nutzt eine frische temp-DB (tempfile.mkdtemp), niemals die Produkt-DB. Keine Netzwerk-/Tolaria-/Search-/Forgejo-Writes. Keine Mutation produktiver State. """ from __future__ import annotations import os import sys import tempfile import traceback from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) from rq_c5a import ( C5AStore, SyncStateMachine, assert_no_write_guarantee, ST_APPLIED, ST_HUMAN_REVIEW_REQUIRED, ST_WAITING_FOR_PREDECESSOR, OP_CREATE, OP_DELETE_REQUEST, IDEM_ALREADY_APPLIED, RC_OUT_OF_ORDER_COMMIT, RC_SECRET_DETECTED, RC_DANGLING_DERIVED_FROM, RC_AMBIGUOUS_DELETE, RC_UNEXPECTED_TOLARIA_DRIFT, ) from rq_c5c import assert_no_search_calls, assert_no_master_write as c5c_no_master from rq_c5d import ( assert_no_master_write as c5d_no_master, assert_no_production_activation, evaluate_external_adoption, ) from rq_c5e import health_contract, HEALTH_BLOCKED, HEALTH_HEALTHY def _new_store(): tmp = tempfile.mkdtemp(prefix="c5g1_") return C5AStore(os.path.join(tmp, "c5g1.db")) def _commit(sha, parent=None, seq=1, objects=None): return { "commit_sha": sha, "parent_sha": parent, "discovered_at": 1000, "sequence": seq, "changed_objects": objects or [], } def _obj(object_id, op=OP_CREATE, **kw): d = {"object_id": object_id, "operation": op} d.update(kw) return d def _active_store(): """Frische temp-DB, gebootstrapt bis ACTIVE (nie Produkt-DB).""" store = _new_store() store.bootstrap_transition("RECONCILING") store.bootstrap_transition("BASELINE_READY") store.set_baseline("base") store.bootstrap_transition("ACTIVE") return store # --------------------------------------------------------------------------- # §29.1 — Forgejo bleibt Master (kein Rückschreib) # --------------------------------------------------------------------------- def test_forgejo_remains_master(): """C5C und C5D duerfen NIE nach Forgejo zurueckschreiben.""" assert c5c_no_master()["no_master_write"] is True, "C5C darf nicht nach Forgejo schreiben" assert c5d_no_master()["no_master_write"] is True, "C5D darf nicht nach Forgejo schreiben" # --------------------------------------------------------------------------- # §29.2 — Deterministische Propagation (Forgejo→Tolaria→Search) # --------------------------------------------------------------------------- def test_deterministic_propagation_chain(): """Normaler Commit-Lebenszyklus endet deterministisch in APPLIED.""" store = _active_store() sm = SyncStateMachine(store) try: c = _commit("c1", objects=[_obj("obj-1")]) store.add_object_change({"commit_sha": "c1", "object_id": "obj-1", "operation": OP_CREATE}) r = sm.process_commit(c) assert r["status"] == ST_APPLIED, f"erwartet APPLIED, got {r}" assert store.commit_status("c1") == ST_APPLIED assert store.health()["last_applied_commit"] == "c1" finally: store.close() # --------------------------------------------------------------------------- # §29.3 — Idempotent (doppelte Events = ALREADY_APPLIED) # --------------------------------------------------------------------------- def test_idempotent_duplicate_event(): """Bereits angewendeter Commit -> ALREADY_APPLIED, kein zweiter Durchlauf.""" store = _active_store() sm = SyncStateMachine(store) try: c = _commit("c1", objects=[_obj("obj-1")]) store.add_object_change({"commit_sha": "c1", "object_id": "obj-1", "operation": OP_CREATE}) r1 = sm.process_commit(c) assert r1["status"] == ST_APPLIED r2 = sm.process_commit(c) assert r2["status"] == ST_APPLIED assert r2["idempotency"] == IDEM_ALREADY_APPLIED, f"erwartet ALREADY_APPLIED, got {r2}" finally: store.close() # --------------------------------------------------------------------------- # §29.4 — Ordering-safe (out-of-order zurückgestellt) # --------------------------------------------------------------------------- def test_ordering_out_of_order_held(): """Commit ohne angewendeten Vorgaenger -> WAITING_FOR_PREDECESSOR.""" store = _active_store() sm = SyncStateMachine(store) try: c = _commit("c2", parent="c1", seq=2, objects=[_obj("obj-1")]) store.add_object_change({"commit_sha": "c2", "object_id": "obj-1", "operation": OP_CREATE}) r = sm.process_commit(c) assert r["status"] == ST_WAITING_FOR_PREDECESSOR, f"erwartet WAITING, got {r}" assert r["reason_code"] == RC_OUT_OF_ORDER_COMMIT finally: store.close() # --------------------------------------------------------------------------- # §29.5 — Replaybar (last_applied_commit nur nach vollständigem PASS) # --------------------------------------------------------------------------- def test_replayable_last_applied_only_after_pass(): """last_applied_commit wird NUR nach APPLIED fortgeschrieben.""" store = _active_store() sm = SyncStateMachine(store) try: # Out-of-order Commit -> nicht angewendet, last_applied bleibt base c = _commit("c2", parent="c1", seq=2, objects=[_obj("obj-1")]) store.add_object_change({"commit_sha": "c2", "object_id": "obj-1", "operation": OP_CREATE}) sm.process_commit(c) assert store.health()["last_applied_commit"] == "base" # Normaler Commit -> APPLIED, last_applied = c1 c1 = _commit("c1", objects=[_obj("obj-1")]) store.add_object_change({"commit_sha": "c1", "object_id": "obj-1", "operation": OP_CREATE}) sm.process_commit(c1) assert store.health()["last_applied_commit"] == "c1" finally: store.close() # --------------------------------------------------------------------------- # §29.6 — Drift-aware (Erkennung, kein blindes Überschreiben) # --------------------------------------------------------------------------- def test_drift_aware_fail_closed(): """Unerwarteter Drift -> HUMAN_REVIEW_REQUIRED (kein blindes Ueberschreiben).""" store = _active_store() sm = SyncStateMachine(store) try: c = _commit("c1", objects=[_obj("obj-1", drift=True)]) store.add_object_change({"commit_sha": "c1", "object_id": "obj-1", "operation": OP_CREATE, "drift": True}) r = sm.process_commit(c) assert r["status"] == ST_HUMAN_REVIEW_REQUIRED, f"erwartet HUMAN_REVIEW, got {r}" assert r["reason_code"] == RC_UNEXPECTED_TOLARIA_DRIFT finally: store.close() # --------------------------------------------------------------------------- # §29.7 — Fail-closed bei Konflikten/unexpected drift # --------------------------------------------------------------------------- def test_fail_closed_without_allow_writes(): """C5E-Replay ohne allow_writes=True -> blockiert (kein produktiver Write).""" from rq_c5e import C5EEngine, C5EStore tmp = tempfile.mkdtemp(prefix="c5g1_") store = C5EStore(os.path.join(tmp, "c5g1.db")) try: rec = C5EEngine(store) # allow_writes=False Default # Replay eines nicht existenten Commits -> fail-closed, kein Write r = rec.recover("nicht_existent") assert r.get("decision") in ("HUMAN_REVIEW", "BLOCKED", "FAIL_CLOSED"), f"got {r}" assert r.get("state") == "UNKNOWN", f"got {r}" finally: store.close() # --------------------------------------------------------------------------- # §29.8 — Search erst nach Tolaria (kein vorzeitiger Search-Stand) # --------------------------------------------------------------------------- def test_search_after_tolaria_only(): """C5C darf keine Search-Calls ausfuehren (Search erst nach Tolaria-PASS).""" assert assert_no_search_calls()["no_search_calls"] is True, "C5C darf Search nicht aufrufen" # --------------------------------------------------------------------------- # §29.9 — Keine dangling relations (derived_from-Policy) # --------------------------------------------------------------------------- def test_no_dangling_relations(): """Dangling derived_from -> HUMAN_GATE (RC_DANGLING_DERIVED_FROM).""" store = _active_store() sm = SyncStateMachine(store) try: c = _commit("c1", objects=[_obj("obj-1", derived_from="nonexistent-obj")]) store.add_object_change({"commit_sha": "c1", "object_id": "obj-1", "operation": OP_CREATE, "derived_from": "nonexistent-obj"}) r = sm.process_commit(c) assert r["status"] == ST_HUMAN_REVIEW_REQUIRED, f"erwartet HUMAN_REVIEW, got {r}" assert r["reason_code"] == RC_DANGLING_DERIVED_FROM finally: store.close() # --------------------------------------------------------------------------- # §29.10 — Keine ID-Neuvergabe # --------------------------------------------------------------------------- def test_no_id_reassignment(): """Rename/Move behalten die object_id (keine Neuvergabe).""" store = _active_store() try: store.add_object_change({ "commit_sha": "c1", "object_id": "object/abc", "path_before": "/app/vault/a.md", "path_after": "/app/vault/b.md", "operation": "RENAME", "state": "current", }) oc = store.get_object_change("c1", "object/abc", "RENAME") assert oc is not None and oc["object_id"] == "object/abc", "ID darf nicht neu vergeben werden" finally: store.close() # --------------------------------------------------------------------------- # §29.11 — Keine Secrets in Logs/Reports/Image # --------------------------------------------------------------------------- def test_no_secrets(): """Secret-Detection (C5B detect_secret) -> FAIL_CLOSED (RC_SECRET_DETECTED).""" from rq_c5b import detect_secret # Klartext-Secret (OpenAI-Key-Pattern) wird erkannt assert detect_secret("api_key=sk-abcdefghijklmnopqrstuvwxyz123456") is not None, "Secret muss erkannt werden" # Sauberer Inhalt ohne Secret -> None assert detect_secret("nur normaler dokumentationstext") is None, "kein Secret erwartet" # C5B-Propagation blockiert bei Secret (RC_SECRET_DETECTED) assert RC_SECRET_DETECTED == "SECRET_DETECTED" # --------------------------------------------------------------------------- # §29.12 — Recovery getestet (Replay + Full-Reconciliation) # --------------------------------------------------------------------------- def test_recovery_replay_reconcile(): """C5E-Health-Contract: HEALTHY bei sauberem Zustand, BLOCKED bei DEAD/Human.""" store = _active_store() try: # Sauberer Zustand -> HEALTHY h = health_contract(store) assert h["status"] == HEALTH_HEALTHY, f"erwartet HEALTHY, got {h}" # DEAD-Commit -> BLOCKED store.upsert_commit({"commit_sha": "dead1", "parent_sha": "base", "status": "DEAD", "retry_count": 0}) h2 = health_contract(store) assert h2["status"] == HEALTH_BLOCKED, f"erwartet BLOCKED, got {h2}" finally: store.close() # --------------------------------------------------------------------------- # §29.13 — Canary PASS (C5F FINAL CLOSED Baseline) # --------------------------------------------------------------------------- def test_canary_pass_baseline(): """C5F-Canary ist geloescht (Baseline FINAL CLOSED). Kein Canary im produktiven Index.""" # Der produktive Canary-Status ist in der C5F-Baseline verifiziert (FINAL CLOSED). # Dieser Test prueft die Adoption-Semantik: ein Index OHNE Canary ist adoptierbar, # ein Index MIT Canary (abweichend) ist NICHT adoptierbar. expected = { "expected_object_ids": {"object/1", "object/2"}, "expected_paths": {"/app/vault/a.md", "/app/vault/b.md"}, "expected_object_count": 2, "source_head": "abc123", } # Ohne Canary -> adoptierbar clean = { "index_built": True, "integrity_ok": True, "object_count": 2, "failed_objects": [], "stale_objects": [], "secret_blocked_objects": 0, "indexed_paths": {"/app/vault/a.md", "/app/vault/b.md"}, "indexed_object_ids": {"object/1", "object/2"}, "source_head": "abc123", } r = evaluate_external_adoption(expected, clean) assert r["ok"] is True, f"sauberer Index muss adoptierbar sein: {r}" # Mit Canary (extra ID+Path) -> NICHT adoptierbar with_canary = dict(clean) with_canary["indexed_object_ids"] = {"object/1", "object/2", "object/canary"} with_canary["indexed_paths"] = {"/app/vault/a.md", "/app/vault/b.md", "/app/vault/canary.md"} with_canary["object_count"] = 3 r2 = evaluate_external_adoption(expected, with_canary) assert r2["ok"] is False, "Index mit Canary darf NICHT adoptiert werden" # --------------------------------------------------------------------------- # ZUSATZ — DELETE bleibt HUMAN_GATED (Invariante A/B/C, §6/§7) # --------------------------------------------------------------------------- def test_delete_stays_human_gated(): """DELETE-ObjectChange ohne Approval -> HUMAN_GATE (RC_AMBIGUOUS_DELETE).""" store = _active_store() sm = SyncStateMachine(store) try: c = _commit("c1", objects=[_obj("obj-1", op=OP_DELETE_REQUEST)]) store.add_object_change({"commit_sha": "c1", "object_id": "obj-1", "operation": OP_DELETE_REQUEST}) r = sm.process_commit(c) assert r["status"] == ST_HUMAN_REVIEW_REQUIRED, f"erwartet HUMAN_REVIEW, got {r}" assert r["reason_code"] == RC_AMBIGUOUS_DELETE # Kein Auto-Approval: delete_approvals leer assert store.list_delete_approvals() == [], "kein Auto-Approval erlaubt" finally: store.close() # --------------------------------------------------------------------------- # ZUSATZ — Provenance/Adoption (C5D source_provenance Contract) # --------------------------------------------------------------------------- def test_provenance_persist_and_readback(): """Source-Provenance wird commit-spezifisch persistiert und read-back verifiziert.""" store = _active_store() try: store.persist_search_source_provenance( "c1", "abc123", ["object/1", "object/2"], ["/app/vault/a.md", "/app/vault/b.md"], ) prov = store.get_source_provenance("c1") assert prov is not None, "Provenance muss persistiert sein" assert prov["source_head"] == "abc123" assert set(prov["source_object_ids"]) == {"object/1", "object/2"} assert set(prov["source_paths"]) == {"/app/vault/a.md", "/app/vault/b.md"} finally: store.close() # --------------------------------------------------------------------------- # ZUSATZ — No-Write-Guarantee (C5A) # --------------------------------------------------------------------------- def test_c5a_no_write_guarantee(): """C5A importiert keine Netzwerk-/Mutationsmodule (kein externer Write).""" g = assert_no_write_guarantee() assert g["no_write_guarantee"] is True, f"C5A darf keine externen Writes: {g}" # --------------------------------------------------------------------------- # ZUSATZ — No-Production-Activation (C5D) # --------------------------------------------------------------------------- def test_c5d_no_production_activation(): """C5D darf keine produktive Aktivierung ohne expliziten Pfad ausfuehren.""" g = assert_no_production_activation() assert g["no_production_activation"] is True, f"C5D darf nicht produktiv aktivieren: {g}" ALL_TESTS = [ test_forgejo_remains_master, test_deterministic_propagation_chain, test_idempotent_duplicate_event, test_ordering_out_of_order_held, test_replayable_last_applied_only_after_pass, test_drift_aware_fail_closed, test_fail_closed_without_allow_writes, test_search_after_tolaria_only, test_no_dangling_relations, test_no_id_reassignment, test_no_secrets, test_recovery_replay_reconcile, test_canary_pass_baseline, test_delete_stays_human_gated, test_provenance_persist_and_readback, test_c5a_no_write_guarantee, test_c5d_no_production_activation, ] def main() -> int: passed = 0 failed = 0 failures = [] for t in ALL_TESTS: try: t() passed += 1 print(f"PASS {t.__name__}") except Exception as e: failed += 1 failures.append((t.__name__, e)) print(f"FAIL {t.__name__}: {e}") traceback.print_exc() print(f"\n=== C5G.1 ACCEPTANCE: {passed} PASS / {failed} FAIL ===") if failures: for name, e in failures: print(f" FAILED: {name} -> {e}") return 1 if failed else 0 if __name__ == "__main__": sys.exit(main())