#!/usr/bin/env python3 """ Red Queen — C5D Testsuite (Search Integration + Commit Completion). Testet gegen einen Fake-Search-Client + Fake-Tolaria-Client (isolierte Mocks). Jeder Test nutzt eine frische temp-DB (tempfile.mkdtemp), nie die Produkt-DB. KEINE produktiven Writes. KEIN produktiver Search-Rebuild. Abgedeckte Faelle (C5D-Prompt): erfolgreicher Search-Rebuild erfolgreicher Health-/Integrity-Read-Back Search Auth Failure Search unavailable Network Timeout malformed Search Response integrity_ok=false unerwarteter/falscher Indexzustand Search Rebuild Failure Retry -> erfolgreiche Recovery Max Retry -> DEAD/Human Gate gemaess Contract Replay nach bereits erfolgreicher Tolaria-Phase kein Tolaria-Doppel-Write beim Search-Retry Search darf bei fehlgeschlagener Tolaria-Verifikation NIEMALS aufgerufen werden last_applied_commit bleibt bei JEDEM Failure unveraendert APPLIED darf ohne Search PASS niemals erreicht werden Duplicate/Replay Idempotenz Restart-/Persistence-Verhalten soweit C5D betroffen Secret Safety Regression der bestehenden C5A/C5B/C5C-Contracts """ from __future__ import annotations import json import os import shutil import subprocess import sys import tempfile import unittest from typing import Any, Dict, List, Optional # C5D importieren (aus demselben Verzeichnis) sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from rq_c5a import ( C5AStore, ST_READY, ST_PROPAGATING_TOLARIA, ST_VERIFYING_TOLARIA, ST_UPDATING_SEARCH, ST_VERIFYING_SEARCH, ST_APPLIED, ST_RETRY_PENDING, ST_DEAD, ST_HUMAN_REVIEW_REQUIRED, ST_FAILED, IDEM_ALREADY_APPLIED, IDEM_RETRY_SAFE, RC_SEARCH_REBUILD_FAILURE, RC_AUTH_FAILURE, RC_UNKNOWN_OBJECT_ID, RC_TOLARIA_UNAVAILABLE, RC_UNEXPECTED_TOLARIA_DRIFT, OP_CREATE, OP_CONTENT_UPDATE, ) from rq_c5b import ( GitReader, parse_frontmatter, content_hash, metadata_hash, detect_secret, ) from rq_c5c import ( TolariaClient, C5CPropagator, C5CDryRun, pre_write_drift_check, _vault_path, DRIFT_WRITE_ALLOWED, DRIFT_ALREADY_AT_TARGET, DRIFT_UNEXPECTED, PLAN_WOULD_WRITE, PLAN_ALREADY_AT_TARGET, PLAN_HUMAN_REVIEW, PLAN_DRIFT, TolariaUnavailableError, ReadBackMismatchError, UnexpectedDriftError, assert_no_search_calls, assert_no_master_write, ) from rq_c5d import ( SearchClient, C5DEngine, C5DDryRun, verify_integrity, SearchError, SearchUnavailableError, SearchAuthError, SearchRebuildError, SearchIntegrityError, SearchMalformedResponseError, assert_no_tolaria_write, assert_no_production_activation, INTEGRITY_OK, INTEGRITY_FAIL, ) UUID_A = "object/77b02661-d67b-4bae-b612-01d1287cea6b" UUID_B = "object/48bd264f-607b-15f1-5f73-3e922af9b19d" # --------------------------------------------------------------------------- # Fixture: kontrolliertes Git-Repo (Test-Helfer darf schreiben) # --------------------------------------------------------------------------- def _git(repo: str, *args: str) -> str: proc = subprocess.run(["git", "-C", repo] + list(args), capture_output=True, text=True) if proc.returncode != 0: raise RuntimeError(f"git {' '.join(args)} fehlgeschlagen: {proc.stderr}") return proc.stdout def _write(repo: str, path: str, content: str) -> None: full = os.path.join(repo, path) os.makedirs(os.path.dirname(full), exist_ok=True) with open(full, "w", encoding="utf-8") as f: f.write(content) def _commit(repo: str, message: str) -> str: _git(repo, "add", "-A") _git(repo, "commit", "-m", message, "--no-verify") return _git(repo, "rev-parse", "HEAD").strip() def _fm(id_: str, **extra) -> str: lines = ["---", "knowledge_schema: 1", f"id: {id_}"] for k, v in extra.items(): if isinstance(v, list): lines.append(f"{k}: [{', '.join(v)}]") else: lines.append(f"{k}: {v}") lines.append("---") return "\n".join(lines) + "\n" class FixtureRepo: def __init__(self): self.dir = tempfile.mkdtemp(prefix="c5d_fixture_") _git(self.dir, "init", "-q", "-b", "main") _git(self.dir, "config", "user.email", "test@test") _git(self.dir, "config", "user.name", "Test") self.commits: Dict[str, str] = {} def write(self, path: str, content: str) -> None: _write(self.dir, path, content) def commit(self, message: str) -> str: sha = _commit(self.dir, message) self.commits[message] = sha return sha def cleanup(self) -> None: shutil.rmtree(self.dir, ignore_errors=True) # --------------------------------------------------------------------------- # Fake-Tolaria-API (Mock, isolierter Test-Vault) — zaehlt Writes # --------------------------------------------------------------------------- class FakeTolaria: """In-Memory Fake der Tolaria-Vault-API. Zaehlt Writes, um Doppel-Writes zu beweisen.""" def __init__(self): self.vault: Dict[str, str] = {} self.write_count = 0 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 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 list(self, vault_path: str = "/app/vault") -> List[Dict[str, Any]]: return self.fake.list(vault_path) # --------------------------------------------------------------------------- # Fake-Search-API (Mock) — konfigurierbare Fehler # --------------------------------------------------------------------------- class FakeSearch: """In-Memory Fake der Search-API. Konfigurierbare Fehler + Zaehler.""" def __init__(self): self.rebuild_count = 0 self.health_count = 0 self.rebuild_error: Optional[Exception] = None self.health_error: Optional[Exception] = None self.rebuild_response: Optional[Any] = None self.health_response: Optional[Any] = None self.rebuild_ok = True def rebuild(self) -> Dict[str, Any]: self.rebuild_count += 1 if self.rebuild_error: raise self.rebuild_error if self.rebuild_response is not None: return self.rebuild_response if not self.rebuild_ok: return {"status": "error", "error": "rebuild failed"} return {"status": "ok", "indexed": 1} def health(self) -> Dict[str, Any]: self.health_count += 1 if self.health_error: raise self.health_error if self.health_response is not None: return self.health_response return { "index_built": True, "object_count": 1, "failed_objects": [], "integrity_ok": True, "supported_modes": ["exact", "keyword", "metadata"], "index_version": 1, "source_head": "abc", "secret_blocked_objects": 0, } class FakeSearchClient(SearchClient): """SearchClient, der gegen FakeSearch statt HTTP arbeitet.""" def __init__(self, fake: FakeSearch): super().__init__(base_url="http://fake-search") self.fake = fake def rebuild(self) -> Dict[str, Any]: return self.fake.rebuild() def health(self) -> Dict[str, Any]: return self.fake.health() # --------------------------------------------------------------------------- # Test-Helfer: C5A-Store + Commit mit ObjectChanges # --------------------------------------------------------------------------- def _make_store() -> C5AStore: db = os.path.join(tempfile.mkdtemp(prefix="c5d_db_"), "c5a.db") return C5AStore(db) def _seed_commit(store: C5AStore, sha: str, parent: Optional[str], objs: List[Dict[str, Any]], status: str = ST_UPDATING_SEARCH) -> None: """Legt einen Commit + ObjectChanges im Store an (Status UPDATING_SEARCH).""" store.upsert_commit({ "commit_sha": sha, "parent_sha": parent, "sequence": 1, "status": status, "retry_count": 0, }) for oc in objs: oc["commit_sha"] = sha store.add_object_change(oc) def _seed_ready_for_search(store: C5AStore, sha: str) -> None: """Legt einen Commit an, der Tolaria-verifiziert ist (READY_FOR_SEARCH).""" _seed_commit(store, sha, None, [{ "object_id": UUID_A, "operation": OP_CREATE, "path_before": None, "path_after": "modul-09.md", "content_hash_after": content_hash("body"), }], status=ST_UPDATING_SEARCH) # --------------------------------------------------------------------------- # Tests # --------------------------------------------------------------------------- class TestVerifyIntegrity(unittest.TestCase): """§C5D Health/Integrity Read-Back-Verifikation.""" def test_integrity_ok(self): health = {"integrity_ok": True, "index_built": True, "object_count": 5, "failed_objects": []} res = verify_integrity(health) self.assertTrue(res["ok"]) self.assertIsNone(res["reason"]) def test_integrity_ok_false(self): health = {"integrity_ok": False, "index_built": True, "object_count": 5, "failed_objects": []} res = verify_integrity(health) self.assertFalse(res["ok"]) self.assertIn("integrity_ok", res["reason"]) def test_index_not_built(self): health = {"integrity_ok": True, "index_built": False, "object_count": 5, "failed_objects": []} res = verify_integrity(health) self.assertFalse(res["ok"]) self.assertIn("index_built", res["reason"]) def test_object_count_zero(self): # unerwarteter/falscher Indexzustand: 0 Objekte health = {"integrity_ok": True, "index_built": True, "object_count": 0, "failed_objects": []} res = verify_integrity(health) self.assertFalse(res["ok"]) self.assertIn("object_count_positive", res["reason"]) def test_failed_objects_present(self): health = {"integrity_ok": True, "index_built": True, "object_count": 5, "failed_objects": ["x.md"]} res = verify_integrity(health) self.assertFalse(res["ok"]) self.assertIn("failed_objects_empty", res["reason"]) class TestC5DEngine(unittest.TestCase): """C5D-Engine: Search Integration + Commit Completion.""" def setUp(self): self.fake_tol = FakeTolaria() self.tol_client = FakeTolariaClient(self.fake_tol) self.fake_search = FakeSearch() self.search_client = FakeSearchClient(self.fake_search) self.store = _make_store() self.engine = C5DEngine(self.store, search=self.search_client) def tearDown(self): self.store.close() def _last_applied(self) -> Optional[str]: return self.store.health().get("last_applied_commit") # -- Erfolgreicher Pfad ------------------------------------------------ def test_successful_rebuild_and_apply(self): sha = "c1" _seed_ready_for_search(self.store, sha) res = self.engine.apply_commit(sha) self.assertEqual(res["status"], ST_APPLIED) self.assertEqual(self.fake_search.rebuild_count, 1) self.assertEqual(self.fake_search.health_count, 1) self.assertEqual(self._last_applied(), sha) # C5D darf NIE Tolaria schreiben self.assertEqual(self.fake_tol.write_count, 0) def test_apply_requires_verifying_search_transition(self): sha = "c1" _seed_ready_for_search(self.store, sha) self.engine.apply_commit(sha) # Nach APPLIED ist der Commit im Endzustand self.assertEqual(self.store.commit_status(sha), ST_APPLIED) # -- Idempotenz / Replay ---------------------------------------------- def test_already_applied_no_downstream_write(self): sha = "c1" _seed_ready_for_search(self.store, sha) self.engine.apply_commit(sha) self.assertEqual(self.fake_search.rebuild_count, 1) # Erneuter Aufruf -> kein weiterer Search-Rebuild res = self.engine.apply_commit(sha) self.assertEqual(res["status"], ST_APPLIED) self.assertEqual(res["idempotency"], IDEM_ALREADY_APPLIED) self.assertEqual(self.fake_search.rebuild_count, 1) self.assertEqual(self.fake_tol.write_count, 0) def test_replay_after_successful_tolaria_phase(self): # Commit ist in UPDATING_SEARCH (Tolaria bereits verifiziert). # C5D darf NIE Tolaria erneut aufrufen (kein Doppel-Write). sha = "c1" _seed_ready_for_search(self.store, sha) self.engine.apply_commit(sha) self.assertEqual(self.fake_tol.write_count, 0) self.assertEqual(self.fake_search.rebuild_count, 1) # -- Search Auth Failure ---------------------------------------------- def test_auth_failure_human_gate(self): sha = "c1" _seed_ready_for_search(self.store, sha) self.fake_search.rebuild_error = SearchAuthError("auth", RC_AUTH_FAILURE) res = self.engine.apply_commit(sha) self.assertEqual(res["status"], ST_HUMAN_REVIEW_REQUIRED) self.assertEqual(res["reason_code"], RC_AUTH_FAILURE) self.assertIsNone(self._last_applied()) self.assertEqual(self.store.commit_status(sha), ST_HUMAN_REVIEW_REQUIRED) # -- Search unavailable (retrybar) ------------------------------------- def test_search_unavailable_retry_pending(self): sha = "c1" _seed_ready_for_search(self.store, sha) self.fake_search.rebuild_error = SearchUnavailableError("down", RC_SEARCH_REBUILD_FAILURE) res = self.engine.apply_commit(sha) self.assertEqual(res["status"], ST_RETRY_PENDING) self.assertEqual(res["retry_count"], 1) self.assertIsNone(self._last_applied()) self.assertEqual(self.store.commit_status(sha), ST_RETRY_PENDING) # -- Network Timeout (retrybar) ---------------------------------------- def test_network_timeout_retry_pending(self): sha = "c1" _seed_ready_for_search(self.store, sha) self.fake_search.rebuild_error = SearchUnavailableError("timeout", RC_SEARCH_REBUILD_FAILURE) res = self.engine.apply_commit(sha) self.assertEqual(res["status"], ST_RETRY_PENDING) self.assertEqual(res["retry_count"], 1) self.assertIsNone(self._last_applied()) # -- Malformed Search Response ---------------------------------------- def test_malformed_rebuild_response_human_gate(self): sha = "c1" _seed_ready_for_search(self.store, sha) self.fake_search.rebuild_response = "not-a-dict" res = self.engine.apply_commit(sha) self.assertEqual(res["status"], ST_HUMAN_REVIEW_REQUIRED) self.assertIsNone(self._last_applied()) def test_malformed_health_response_human_gate(self): sha = "c1" _seed_ready_for_search(self.store, sha) self.fake_search.health_response = "not-a-dict" res = self.engine.apply_commit(sha) self.assertEqual(res["status"], ST_HUMAN_REVIEW_REQUIRED) self.assertIsNone(self._last_applied()) # -- integrity_ok=false ------------------------------------------------ def test_integrity_false_human_gate(self): sha = "c1" _seed_ready_for_search(self.store, sha) self.fake_search.health_response = { "index_built": True, "object_count": 1, "failed_objects": [], "integrity_ok": False, } res = self.engine.apply_commit(sha) self.assertEqual(res["status"], ST_HUMAN_REVIEW_REQUIRED) self.assertIsNone(self._last_applied()) self.assertEqual(self.store.commit_status(sha), ST_HUMAN_REVIEW_REQUIRED) # -- unerwarteter/falscher Indexzustand -------------------------------- def test_unexpected_index_state_human_gate(self): sha = "c1" _seed_ready_for_search(self.store, sha) # object_count=0 -> falscher Indexzustand self.fake_search.health_response = { "index_built": True, "object_count": 0, "failed_objects": [], "integrity_ok": True, } res = self.engine.apply_commit(sha) self.assertEqual(res["status"], ST_HUMAN_REVIEW_REQUIRED) self.assertIsNone(self._last_applied()) # -- Search Rebuild Failure ------------------------------------------- def test_rebuild_failure_human_gate(self): sha = "c1" _seed_ready_for_search(self.store, sha) self.fake_search.rebuild_ok = False # status != "ok" res = self.engine.apply_commit(sha) self.assertEqual(res["status"], ST_HUMAN_REVIEW_REQUIRED) self.assertIsNone(self._last_applied()) # -- Retry -> erfolgreiche Recovery ----------------------------------- def test_retry_then_recovery(self): sha = "c1" _seed_ready_for_search(self.store, sha) # 1. Versuch: Search down -> RETRY_PENDING self.fake_search.rebuild_error = SearchUnavailableError("down", RC_SEARCH_REBUILD_FAILURE) res1 = self.engine.apply_commit(sha) self.assertEqual(res1["status"], ST_RETRY_PENDING) self.assertEqual(res1["retry_count"], 1) # 2. Versuch (Replay): Search wieder da -> APPLIED self.fake_search.rebuild_error = None res2 = self.engine.apply_commit(sha) self.assertEqual(res2["status"], ST_APPLIED) self.assertEqual(self._last_applied(), sha) # Kein Tolaria-Write waehrend des gesamten Retry-Zyklus self.assertEqual(self.fake_tol.write_count, 0) # -- Max Retry -> DEAD ------------------------------------------------ def test_max_retry_reaches_dead(self): sha = "c1" _seed_ready_for_search(self.store, sha) self.fake_search.rebuild_error = SearchUnavailableError("down", RC_SEARCH_REBUILD_FAILURE) # max_retries=2 -> nach 2 Fehlversuchen DEAD engine = C5DEngine(self.store, search=self.search_client, max_retries=2) res1 = engine.apply_commit(sha) self.assertEqual(res1["status"], ST_RETRY_PENDING) res2 = engine.apply_commit(sha) self.assertEqual(res2["status"], ST_DEAD) self.assertIsNone(self._last_applied()) self.assertEqual(self.store.commit_status(sha), ST_DEAD) # -- Search darf bei fehlgeschlagener Tolaria-Verifikation NIE aufgerufen werden def test_search_never_called_when_tolaria_not_verified(self): # Commit ist in ST_READY (Tolaria NICHT verifiziert) -> C5D darf nicht starten sha = "c1" _seed_commit(self.store, sha, None, [{ "object_id": UUID_A, "operation": OP_CREATE, "path_before": None, "path_after": "modul-09.md", "content_hash_after": content_hash("body"), }], status=ST_READY) res = self.engine.apply_commit(sha) # C5D startet NICHT aus ST_READY -> kein Search-Rebuild self.assertEqual(self.fake_search.rebuild_count, 0) self.assertEqual(self.fake_search.health_count, 0) self.assertNotEqual(res["status"], ST_APPLIED) self.assertIsNone(self._last_applied()) def test_search_never_called_when_tolaria_drift(self): # Commit in ST_HUMAN_REVIEW_REQUIRED (Tolaria-Drift) -> C5D darf nicht starten sha = "c1" _seed_commit(self.store, sha, None, [{ "object_id": UUID_A, "operation": OP_CREATE, "path_before": None, "path_after": "modul-09.md", "content_hash_after": content_hash("body"), }], status=ST_HUMAN_REVIEW_REQUIRED) res = self.engine.apply_commit(sha) self.assertEqual(self.fake_search.rebuild_count, 0) self.assertNotEqual(res["status"], ST_APPLIED) # -- last_applied_commit bleibt bei JEDEM Failure unveraendert --------- def test_last_applied_unchanged_on_all_failures(self): sha = "c1" _seed_ready_for_search(self.store, sha) # Auth-Fehler self.fake_search.rebuild_error = SearchAuthError("auth", RC_AUTH_FAILURE) self.engine.apply_commit(sha) self.assertIsNone(self._last_applied()) # Integrity-Fehler sha2 = "c2" _seed_ready_for_search(self.store, sha2) self.fake_search.rebuild_error = None self.fake_search.health_response = {"integrity_ok": False, "index_built": True, "object_count": 1, "failed_objects": []} self.engine.apply_commit(sha2) self.assertIsNone(self._last_applied()) # -- APPLIED darf ohne Search PASS niemals erreicht werden ------------- def test_applied_requires_search_pass(self): sha = "c1" _seed_ready_for_search(self.store, sha) # Search-Rebuild schlaegt fehl -> kein APPLIED self.fake_search.rebuild_ok = False res = self.engine.apply_commit(sha) self.assertNotEqual(res["status"], ST_APPLIED) self.assertIsNone(self._last_applied()) # -- Restart-/Persistence-Verhalten ------------------------------------ def test_persistence_across_engine_instances(self): sha = "c1" _seed_ready_for_search(self.store, sha) # Engine 1: Search down -> RETRY_PENDING (persistiert) self.fake_search.rebuild_error = SearchUnavailableError("down", RC_SEARCH_REBUILD_FAILURE) self.engine.apply_commit(sha) self.assertEqual(self.store.commit_status(sha), ST_RETRY_PENDING) # Engine 2 (neue Instanz, gleiche DB): Replay -> APPLIED engine2 = C5DEngine(self.store, search=self.search_client) self.fake_search.rebuild_error = None res = engine2.apply_commit(sha) self.assertEqual(res["status"], ST_APPLIED) self.assertEqual(self._last_applied(), sha) # -- kein Tolaria-Doppel-Write beim Search-Retry ---------------------- def test_no_tolaria_double_write_on_search_retry(self): sha = "c1" _seed_ready_for_search(self.store, sha) # Mehrere Search-Fehler + Recovery -> Tolaria-Write bleibt 0 self.fake_search.rebuild_error = SearchUnavailableError("down", RC_SEARCH_REBUILD_FAILURE) self.engine.apply_commit(sha) self.engine.apply_commit(sha) self.fake_search.rebuild_error = None self.engine.apply_commit(sha) self.assertEqual(self.fake_tol.write_count, 0) class TestC5DDryRun(unittest.TestCase): """C5D Dry-Run (read-only).""" def setUp(self): self.store = _make_store() def tearDown(self): self.store.close() def test_plan_lists_ready_and_retry(self): _seed_ready_for_search(self.store, "c1") _seed_commit(self.store, "c2", None, [{ "object_id": UUID_A, "operation": OP_CREATE, "path_before": None, "path_after": "modul-09.md", "content_hash_after": content_hash("body"), }], status=ST_RETRY_PENDING) dry = C5DDryRun(self.store) plan = dry.plan() self.assertIn("c1", plan["ready_for_search"]) self.assertIn("c2", plan["search_retry_pending"]) class TestGuarantees(unittest.TestCase): """C5D-Guarantees (statisch).""" def test_no_tolaria_write(self): res = assert_no_tolaria_write() self.assertTrue(res["no_tolaria_write"], res) def test_no_master_write(self): res = assert_no_master_write() self.assertTrue(res["no_master_write"], res) def test_no_production_activation(self): res = assert_no_production_activation() self.assertTrue(res["no_production_activation"], res) class TestSecretSafety(unittest.TestCase): """Secret Safety: C5D persistiert/ausgibt keine Secrets.""" def test_detect_secret_blocks(self): # detect_secret aus C5B erkennt Secrets (Regression) self.assertIsNotNone(detect_secret("sk-1234567890abcdefghijklmnopqrstuvwxyz")) self.assertIsNone(detect_secret("normal content ohne secret")) def test_no_secret_in_c5d_source(self): # C5D-Quelle darf keine echten Secret-Werte enthalten import rq_c5d with open(rq_c5d.__file__, encoding="utf-8") as f: src = f.read() # Kein sk-<20+> Muster im Quellcode (nur fragmentierte Referenzen) import re self.assertIsNone(re.search(r"sk-[A-Za-z0-9]{20,}", src)) class TestRegressionContracts(unittest.TestCase): """Regression der bestehenden C5A/C5B/C5C-Contracts.""" def test_c5c_guarantees_still_pass(self): # C5C-Guarantees (No-Search, No-Master-Write) bleiben intakt search = assert_no_search_calls() master = assert_no_master_write() self.assertTrue(search["no_search_calls"]) self.assertTrue(master["no_master_write"]) def test_c5a_transitions_include_search_retry_replay(self): # C5D-Transition (RETRY_PENDING -> UPDATING_SEARCH) muss existieren from rq_c5a import _ALLOWED_TRANSITIONS self.assertIn((ST_RETRY_PENDING, ST_UPDATING_SEARCH), _ALLOWED_TRANSITIONS) def test_c5a_transitions_include_apply_path(self): from rq_c5a import _ALLOWED_TRANSITIONS self.assertIn((ST_UPDATING_SEARCH, ST_VERIFYING_SEARCH), _ALLOWED_TRANSITIONS) self.assertIn((ST_VERIFYING_SEARCH, ST_APPLIED), _ALLOWED_TRANSITIONS) def test_c5c_propagator_still_works(self): # C5C-Propagator (Tolaria-Propagation) bleibt funktional fake = FakeTolaria() client = FakeTolariaClient(fake) repo = FixtureRepo() content = _fm(UUID_A, type="arch", role="module", representation="source", state="current") repo.write("modul-09.md", content) sha = repo.commit("create") store = _make_store() reader = GitReader(repo.dir) _seed_commit(store, sha, None, [{ "object_id": UUID_A, "operation": OP_CREATE, "path_before": None, "path_after": "modul-09.md", "content_hash_after": content_hash(parse_frontmatter(content)[1]), }], status=ST_READY) prop = C5CPropagator(store, reader, client) res = prop.propagate_commit(sha) self.assertEqual(res["status"], ST_UPDATING_SEARCH) self.assertEqual(fake.vault["/app/vault/modul-09.md"], content) repo.cleanup() store.close() if __name__ == "__main__": unittest.main()