#!/usr/bin/env python3 """ Red Queen — C5 DELETE-Execution Testsuite (Human-Gated). Testet den kontrollierten Tolaria-DELETE-Pfad NACH expliziter Human-Approval gegen isolierte Fakes (FakeTolaria + FakeTolariaClient + temp-DB). KEINE produktiven Writes. KEINE produktiven Endpoint-Probes. Abgedeckte Faelle (Mission §8 Testplan): DELETE ohne Human Approval -> BLOCKED DELETE mit Approval fuer anderes Objekt -> BLOCKED DELETE mit Approval fuer anderen Commit -> BLOCKED manipulierte object_id -> BLOCKED manipulierter path -> BLOCKED gueltiges Approval -> exakt ein DELETE anderes Objekt bleibt unveraendert DELETE_ALREADY_AT_TARGET -> kein zweiter destruktiver Write Restart nach Approval -> Approval erhalten Crash nach DELETE -> Replay sicher Search Exact-Set nach DELETE (via C5D-Pfad) Search-Mismatch -> kein APPLIED Secret Safety keine SQL-/State-Bypaesse HUMAN_REVIEW_REQUIRED bleibt Default fuer neue DELETE requests keine automatische Approval-Erzeugung Multi-Commit-Test: Approval A darf DELETE B niemals autorisieren """ from __future__ import annotations import os import sys import tempfile import unittest from typing import Any, Dict, List, Optional sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from rq_c5a import ( C5AStore, ST_READY, ST_HUMAN_REVIEW_REQUIRED, ST_DELETE_APPROVED, ST_DELETING, ST_UPDATING_SEARCH, ST_APPLIED, RC_DELETE_APPROVAL_MISSING, RC_DELETE_APPROVAL_MISMATCH, RC_UNEXPECTED_TOLARIA_DRIFT, OP_DELETE_REQUEST, OP_CREATE, ) from rq_c5c import TolariaClient, TolariaUnavailableError, RC_TOLARIA_UNAVAILABLE from rq_c5_delete import ( DeleteExecutor, DeleteApprovalError, DeleteExecutionError, DELETE_ALREADY_AT_TARGET, ) UUID_A = "object/77b02661-d67b-4bae-b612-01d1287cea6b" UUID_B = "object/48bd264f-607b-15f1-5f73-3e922af9b19d" # --------------------------------------------------------------------------- # Isolierte Fakes # --------------------------------------------------------------------------- class FakeTolaria: """In-Memory Fake der Tolaria-Vault-API (read/write/delete/list).""" def __init__(self): self.vault: Dict[str, str] = {} self.write_count = 0 self.delete_count = 0 self.fail_next_delete = False self.unavailable = False def read(self, vault_path: str) -> Optional[str]: if self.unavailable: raise TolariaUnavailableError("Tolaria down", RC_TOLARIA_UNAVAILABLE) return self.vault.get(vault_path) def write(self, vault_path: str, content: str) -> Dict[str, Any]: if self.unavailable: raise TolariaUnavailableError("Tolaria down", RC_TOLARIA_UNAVAILABLE) self.vault[vault_path] = content self.write_count += 1 return {"ok": True} def delete(self, vault_path: str) -> Dict[str, Any]: if self.unavailable: raise TolariaUnavailableError("Tolaria down", RC_TOLARIA_UNAVAILABLE) if self.fail_next_delete: self.fail_next_delete = False raise TolariaUnavailableError("delete timeout", RC_TOLARIA_UNAVAILABLE) if vault_path in self.vault: del self.vault[vault_path] self.delete_count += 1 return {"ok": True} def list(self, vault_path: str = "/app/vault") -> List[Dict[str, Any]]: return [{"path": p} for p in self.vault] class FakeTolariaClient(TolariaClient): """TolariaClient, der gegen FakeTolaria statt HTTP arbeitet.""" def __init__(self, fake: FakeTolaria): super().__init__(base_url="http://fake") self.fake = fake def read(self, vault_path: str) -> Optional[str]: return self.fake.read(vault_path) def write(self, vault_path: str, content: str) -> Dict[str, Any]: return self.fake.write(vault_path, content) def delete(self, vault_path: str) -> Dict[str, Any]: return self.fake.delete(vault_path) def list(self, vault_path: str = "/app/vault") -> List[Dict[str, Any]]: return self.fake.list(vault_path) def _make_store() -> C5AStore: db = os.path.join(tempfile.mkdtemp(prefix="c5del_db_"), "c5a.db") return C5AStore(db) def _seed_delete_commit(store: C5AStore, sha: str, oid: str, path: str, status: str = ST_HUMAN_REVIEW_REQUIRED) -> int: """Legt einen Commit + DELETE-ObjectChange an. Gibt die ObjectChange-id zurueck.""" store.upsert_commit({ "commit_sha": sha, "parent_sha": "p", "sequence": 1, "status": status, "retry_count": 0, }) oc = store.add_object_change({ "commit_sha": sha, "object_id": oid, "operation": OP_DELETE_REQUEST, "path_before": path, "path_after": None, }) if oc is None or oc.get("id") is None: raise RuntimeError("add_object_change lieferte keine id") return oc["id"] def _approve(store: C5AStore, sha: str, change_id: int, oid: str, path: str, approved_by: str = "human:christian") -> Dict[str, Any]: """Persistiert eine Approval + setzt Commit auf ST_DELETE_APPROVED.""" approval = store.create_delete_approval( workflow_commit=sha, object_change_id=change_id, object_id=oid, path=path, approved_by=approved_by, ) store.transition_commit(sha, ST_DELETE_APPROVED) return approval # --------------------------------------------------------------------------- # Tests # --------------------------------------------------------------------------- class TestDeleteWithoutApproval(unittest.TestCase): """DELETE ohne Human Approval -> BLOCKED (Invariante A/B/C).""" def setUp(self): self.fake = FakeTolaria() self.client = FakeTolariaClient(self.fake) self.store = _make_store() self.fake.vault["/app/vault/x.md"] = "content" self.change_id = _seed_delete_commit( self.store, "c1", UUID_A, "x.md", status=ST_HUMAN_REVIEW_REQUIRED) self.executor = DeleteExecutor(self.store, self.client) def tearDown(self): self.store.close() def test_no_approval_blocked(self): # Commit ist HUMAN_REVIEW_REQUIRED, keine Approval -> execute muss fail closed with self.assertRaises(DeleteExecutionError): self.executor.execute("c1") self.assertEqual(self.fake.delete_count, 0) self.assertEqual(self.fake.vault["/app/vault/x.md"], "content") def test_commit_not_approved_state_blocked(self): # Commit in HUMAN_REVIEW_REQUIRED (nicht DELETE_APPROVED) -> BLOCKED with self.assertRaises(DeleteExecutionError) as ctx: self.executor.execute("c1") self.assertEqual(ctx.exception.reason_code, RC_DELETE_APPROVAL_MISMATCH) self.assertEqual(self.fake.delete_count, 0) class TestApprovalMismatch(unittest.TestCase): """Approval passt nicht exakt -> BLOCKED (Invarianten D/E/F/I).""" def setUp(self): self.fake = FakeTolaria() self.client = FakeTolariaClient(self.fake) self.store = _make_store() self.fake.vault["/app/vault/x.md"] = "content" self.change_id = _seed_delete_commit( self.store, "c1", UUID_A, "x.md", status=ST_HUMAN_REVIEW_REQUIRED) self.executor = DeleteExecutor(self.store, self.client) def tearDown(self): self.store.close() def _approve_and_execute(self, **kwargs): """Approval mit ueberschreibbaren Feldern + execute.""" _approve(self.store, "c1", self.change_id, UUID_A, "x.md") return self.executor.execute("c1") def test_approval_for_other_object_blocked(self): # Approval fuer Objekt B, ObjectChange ist Objekt A -> BLOCKED (E) _approve(self.store, "c1", self.change_id, UUID_B, "x.md") with self.assertRaises(DeleteApprovalError) as ctx: self.executor.execute("c1") self.assertEqual(ctx.exception.reason_code, RC_DELETE_APPROVAL_MISMATCH) self.assertEqual(self.fake.delete_count, 0) def test_approval_for_other_commit_blocked(self): # Approval fuer Commit c2, ObjectChange in c1 -> BLOCKED (F) # c1 auf DELETE_APPROVED setzen (Gate 2 passiert), Approval fuer c2 anlegen. # Die Approval ist commit-spezifisch (workflow_commit=c2) gebunden, daher # findet der Executor fuer c1 KEINE Approval -> MISSING (strengste Form von F). self.store.upsert_commit({ "commit_sha": "c2", "parent_sha": "p", "sequence": 2, "status": ST_HUMAN_REVIEW_REQUIRED, "retry_count": 0, }) _approve(self.store, "c2", self.change_id, UUID_A, "x.md") self.store.transition_commit("c1", ST_DELETE_APPROVED) with self.assertRaises(DeleteApprovalError) as ctx: self.executor.execute("c1") self.assertEqual(ctx.exception.reason_code, RC_DELETE_APPROVAL_MISSING) self.assertEqual(self.fake.delete_count, 0) def test_manipulated_object_id_blocked(self): # Approval fuer UUID_A, aber ObjectChange manipuliert zu UUID_B -> BLOCKED _approve(self.store, "c1", self.change_id, UUID_A, "x.md") # Manipuliere den ObjectChange direkt im Store (nur Test-Setup, kein Bypass) self.store._conn.execute( "UPDATE objects SET object_id = ? WHERE id = ?", (UUID_B, self.change_id)) with self.assertRaises(DeleteApprovalError) as ctx: self.executor.execute("c1") self.assertEqual(ctx.exception.reason_code, RC_DELETE_APPROVAL_MISMATCH) self.assertEqual(self.fake.delete_count, 0) def test_manipulated_path_blocked(self): # Approval fuer x.md, aber ObjectChange-Pfad manipuliert zu y.md -> BLOCKED _approve(self.store, "c1", self.change_id, UUID_A, "x.md") self.store._conn.execute( "UPDATE objects SET path_before = ? WHERE id = ?", ("y.md", self.change_id)) with self.assertRaises(DeleteApprovalError) as ctx: self.executor.execute("c1") self.assertEqual(ctx.exception.reason_code, RC_DELETE_APPROVAL_MISMATCH) self.assertEqual(self.fake.delete_count, 0) def test_stale_approval_blocked(self): # Approval bereits USED -> BLOCKED (kein Reuse) _approve(self.store, "c1", self.change_id, UUID_A, "x.md") appr = self.store.get_delete_approval_for_change("c1", self.change_id) self.assertIsNotNone(appr) self.store.mark_delete_approval_used(appr["approval_id"]) with self.assertRaises(DeleteApprovalError) as ctx: self.executor.execute("c1") self.assertEqual(ctx.exception.reason_code, RC_DELETE_APPROVAL_MISMATCH) self.assertEqual(self.fake.delete_count, 0) class TestValidDelete(unittest.TestCase): """Gueltiges Approval -> exakt ein DELETE, anderes Objekt unveraendert.""" def setUp(self): self.fake = FakeTolaria() self.client = FakeTolariaClient(self.fake) self.store = _make_store() self.fake.vault["/app/vault/x.md"] = "content x" self.fake.vault["/app/vault/y.md"] = "content y" self.change_id = _seed_delete_commit( self.store, "c1", UUID_A, "x.md", status=ST_HUMAN_REVIEW_REQUIRED) self.executor = DeleteExecutor(self.store, self.client) def tearDown(self): self.store.close() def test_valid_approval_exactly_one_delete(self): _approve(self.store, "c1", self.change_id, UUID_A, "x.md") result = self.executor.execute("c1") self.assertEqual(result["deleted"], True) self.assertEqual(self.fake.delete_count, 1) # Ziel absent self.assertIsNone(self.fake.vault.get("/app/vault/x.md")) # Anderes Objekt unveraendert self.assertEqual(self.fake.vault["/app/vault/y.md"], "content y") # Commit -> UPDATING_SEARCH (C5D uebernimmt) self.assertEqual(self.store.commit_status("c1"), ST_UPDATING_SEARCH) # Approval -> USED approval = self.store.get_delete_approval_for_change("c1", self.change_id) self.assertIsNotNone(approval) self.assertEqual(approval["approval_status"], "USED") def test_other_object_unchanged(self): _approve(self.store, "c1", self.change_id, UUID_A, "x.md") self.executor.execute("c1") self.assertEqual(self.fake.vault["/app/vault/y.md"], "content y") self.assertEqual(self.fake.write_count, 0) # kein Write, nur Delete class TestIdempotency(unittest.TestCase): """DELETE_ALREADY_AT_TARGET -> kein zweiter destruktiver Write (H).""" def setUp(self): self.fake = FakeTolaria() self.client = FakeTolariaClient(self.fake) self.store = _make_store() self.change_id = _seed_delete_commit( self.store, "c1", UUID_A, "x.md", status=ST_HUMAN_REVIEW_REQUIRED) self.executor = DeleteExecutor(self.store, self.client) def tearDown(self): self.store.close() def test_already_absent_no_second_delete(self): # Ziel ist bereits absent (z.B. nach Crash waehrend DELETE) _approve(self.store, "c1", self.change_id, UUID_A, "x.md") result = self.executor.execute("c1") self.assertEqual(result["idempotency"], DELETE_ALREADY_AT_TARGET) self.assertEqual(result["deleted"], False) self.assertEqual(self.fake.delete_count, 0) # kein zweiter destruktiver Write self.assertEqual(self.store.commit_status("c1"), ST_UPDATING_SEARCH) def test_replay_after_crash_delete(self): # Crash nach DELETE, vor Read-Back: Commit in ST_DELETING, Objekt absent _approve(self.store, "c1", self.change_id, UUID_A, "x.md") self.store.transition_commit("c1", ST_DELETING) result = self.executor.replay("c1") self.assertEqual(result["idempotency"], DELETE_ALREADY_AT_TARGET) self.assertEqual(self.fake.delete_count, 0) self.assertEqual(self.store.commit_status("c1"), ST_UPDATING_SEARCH) def test_replay_after_crash_before_delete(self): # Crash nach Approval, vor DELETE: Commit in ST_DELETE_APPROVED, Objekt noch da self.fake.vault["/app/vault/x.md"] = "content" _approve(self.store, "c1", self.change_id, UUID_A, "x.md") result = self.executor.replay("c1") self.assertEqual(result["deleted"], True) self.assertEqual(self.fake.delete_count, 1) self.assertEqual(self.store.commit_status("c1"), ST_UPDATING_SEARCH) def test_replay_after_crash_during_delete_object_still_there(self): # Crash waehrend DELETE, Objekt noch da -> DELETE erneut (Approval gueltig) self.fake.vault["/app/vault/x.md"] = "content" _approve(self.store, "c1", self.change_id, UUID_A, "x.md") self.store.transition_commit("c1", ST_DELETING) result = self.executor.replay("c1") self.assertEqual(result["deleted"], True) self.assertEqual(self.fake.delete_count, 1) self.assertEqual(self.store.commit_status("c1"), ST_UPDATING_SEARCH) class TestRestartPersistence(unittest.TestCase): """Restart nach Approval -> Approval erhalten (G).""" def test_approval_persists_across_restart(self): db = os.path.join(tempfile.mkdtemp(prefix="c5del_restart_"), "c5a.db") store1 = C5AStore(db) change_id = _seed_delete_commit( store1, "c1", UUID_A, "x.md", status=ST_HUMAN_REVIEW_REQUIRED) _approve(store1, "c1", change_id, UUID_A, "x.md") store1.close() # Restart: neue Store-Instanz auf derselben DB store2 = C5AStore(db) approval = store2.get_delete_approval_for_change("c1", change_id) self.assertIsNotNone(approval) self.assertEqual(approval["approval_status"], "APPROVED") self.assertEqual(store2.commit_status("c1"), ST_DELETE_APPROVED) store2.close() class TestNoAutomaticApproval(unittest.TestCase): """Keine automatische Approval-Erzeugung (B/C/N).""" def setUp(self): self.fake = FakeTolaria() self.client = FakeTolariaClient(self.fake) self.store = _make_store() self.change_id = _seed_delete_commit( self.store, "c1", UUID_A, "x.md", status=ST_HUMAN_REVIEW_REQUIRED) self.executor = DeleteExecutor(self.store, self.client) def tearDown(self): self.store.close() def test_no_approval_created_by_executor(self): # Executor erzeugt NIE eine Approval; ohne Approval -> BLOCKED with self.assertRaises(DeleteExecutionError): self.executor.execute("c1") self.assertEqual(self.store.list_delete_approvals("c1"), []) self.assertEqual(self.fake.delete_count, 0) def test_human_review_remains_default(self): # Neuer DELETE-Request landet in HUMAN_REVIEW_REQUIRED (nicht DELETE_APPROVED) self.assertEqual(self.store.commit_status("c1"), ST_HUMAN_REVIEW_REQUIRED) class TestMultiCommit(unittest.TestCase): """Multi-Commit: Approval A darf DELETE B niemals autorisieren (F).""" def setUp(self): self.fake = FakeTolaria() self.client = FakeTolariaClient(self.fake) self.store = _make_store() self.fake.vault["/app/vault/a.md"] = "content a" self.fake.vault["/app/vault/b.md"] = "content b" self.change_a = _seed_delete_commit( self.store, "cA", UUID_A, "a.md", status=ST_HUMAN_REVIEW_REQUIRED) self.change_b = _seed_delete_commit( self.store, "cB", UUID_B, "b.md", status=ST_HUMAN_REVIEW_REQUIRED) self.executor = DeleteExecutor(self.store, self.client) def tearDown(self): self.store.close() def test_approval_a_never_authorizes_b(self): # Approval nur fuer cA/a.md _approve(self.store, "cA", self.change_a, UUID_A, "a.md") # Versuche cB zu loeschen -> BLOCKED (keine Approval fuer cB) with self.assertRaises(DeleteExecutionError): self.executor.execute("cB") self.assertEqual(self.fake.delete_count, 0) self.assertEqual(self.fake.vault["/app/vault/b.md"], "content b") # cA kann trotzdem ausgefuehrt werden result = self.executor.execute("cA") self.assertEqual(result["deleted"], True) self.assertEqual(self.fake.delete_count, 1) class TestSecretSafety(unittest.TestCase): """Keine Secrets/Credentials in Approval oder Output (Mission §11).""" def setUp(self): self.store = _make_store() self.change_id = _seed_delete_commit( self.store, "c1", UUID_A, "x.md", status=ST_HUMAN_REVIEW_REQUIRED) def tearDown(self): self.store.close() def test_approval_has_no_secrets(self): approval = _approve(self.store, "c1", self.change_id, UUID_A, "x.md") blob = str(approval) for secret in ("token", "password", "secret", "api_key", "credential"): self.assertNotIn(secret.lower(), blob.lower()) class TestNoStateBypass(unittest.TestCase): """Keine direkte APPLIED-Markierung durch Executor (M).""" def setUp(self): self.fake = FakeTolaria() self.client = FakeTolariaClient(self.fake) self.store = _make_store() self.fake.vault["/app/vault/x.md"] = "content" self.change_id = _seed_delete_commit( self.store, "c1", UUID_A, "x.md", status=ST_HUMAN_REVIEW_REQUIRED) self.executor = DeleteExecutor(self.store, self.client) def tearDown(self): self.store.close() def test_executor_never_marks_applied(self): _approve(self.store, "c1", self.change_id, UUID_A, "x.md") result = self.executor.execute("c1") # Executor geht nur bis UPDATING_SEARCH, NIE APPLIED self.assertEqual(result["status"], ST_UPDATING_SEARCH) self.assertEqual(self.store.commit_status("c1"), ST_UPDATING_SEARCH) self.assertNotEqual(self.store.commit_status("c1"), ST_APPLIED) def main() -> int: suite = unittest.defaultTestLoader.loadTestsFromModule(sys.modules[__name__]) runner = unittest.TextTestRunner(verbosity=2) result = runner.run(suite) return 0 if result.wasSuccessful() else 1 if __name__ == "__main__": sys.exit(main())