""" AUTH.4C3 — test_reconciliation.py ================================= Isolierte Tests für den OUTCOME_UNKNOWN-RECONCILIATION-CONTRACT. Testumgebung: * JobStore auf temporärer SQLite-DB * Fake-Source-Loader (liefert Content aus Test-Repo) * Fake-Read-Back (liefert Tolaria-Target) * KEIN write-Callable im Reconciliation-Core (technisch unmöglich) Führt die AUTH.4C3-Tests T1-T30 aus. Exit 0 = alle PASS. """ from __future__ import annotations import hashlib import json import os import shutil import sys import tempfile import unittest import uuid from typing import Any, Dict, Optional # Module aus dem Build-Verzeichnis importieren sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from job_schema import make_save_job, JobRejectedError from job_store import JobStore, JobImmutableFieldError from job_state_machine import ( ST_OUTCOME_UNKNOWN, ST_RECONCILED, ST_SUCCEEDED, ST_FAILED, ST_REJECTED, ST_READY, ST_CLAIMED, ST_EXECUTING, ST_CREATED, InvalidTransitionError, transition, is_valid_transition, ) from save_reconciliation import ( SaveReconciliationCore, ReconciliationError, RC_TARGET_EXACT, RC_TARGET_ABSENT, RC_TARGET_MISMATCH, RC_READ_UNAVAILABLE, RC_INVALID_JOB, RC_PROVENANCE_MISMATCH, RC_ALREADY_RECONCILED, EV_RECONCILIATION_STARTED, EV_RECONCILIATION_TARGET_EXACT, EV_CONFIRMED_EXECUTED, ) def _sha256(s: str) -> str: return hashlib.sha256(s.encode("utf-8")).hexdigest() class FakeSourceLoader: """Fake-Source-Loader. Liefert Content aus einem Dict (commit -> path -> content).""" def __init__(self, content: str, fail: bool = False): self.content = content self.fail = fail self.calls = 0 def __call__(self, source_commit: str, object_id: str, vault_path: str) -> str: self.calls += 1 if self.fail: raise RuntimeError("source unavailable") return self.content class FakeReadBack: """Fake-Read-Back. Liefert Tolaria-Target oder None (absent) / wirft (unavailable).""" def __init__(self, content: Optional[str] = None, fail: bool = False): self.content = content self.fail = fail self.calls = 0 def __call__(self, vault_path: str) -> Optional[str]: self.calls += 1 if self.fail: raise RuntimeError("read unavailable") return self.content class AUTH4C3ReconciliationTests(unittest.TestCase): @classmethod def setUpClass(cls): cls.tmp = tempfile.mkdtemp(prefix="c5-4c3-test-") cls.db = os.path.join(cls.tmp, "test_save.db") cls.store = JobStore(cls.db, "SAVE") cls.content = "---\nid: object/test-obj-1\ntitle: Test\n---\n\nHello World\n" cls.prov = _sha256(cls.content) cls.commit = "a" * 40 cls.object_id = "object/test-obj-1" cls.vault_path = "tolaria/test.md" @classmethod def tearDownClass(cls): cls.store.close() shutil.rmtree(cls.tmp, ignore_errors=True) def _make_job(self, **overrides) -> Dict[str, Any]: job = make_save_job( job_id=str(uuid.uuid4()), mission_id=str(uuid.uuid4()), object_id=self.object_id, vault_path=self.vault_path, source_commit=self.commit, provenance_hash=self.prov, expected_state="present", created_at="2026-08-27T00:00:00Z", idempotency_key=str(uuid.uuid4()), ) job.update(overrides) return job def _create_outcome_unknown(self, **overrides) -> Dict[str, Any]: """Erzeugt einen Job im OUTCOME_UNKNOWN-Zustand (wie der produktive Incident).""" job = self._make_job(**overrides) self.store.create_job(job) self.store.mark_ready(job["job_id"]) # Simuliere: Job wurde geclaimt + execution -> OUTCOME_UNKNOWN # (direkt via _transition, da wir keinen vollen Execution-Pfad brauchen) self.store._transition(job["job_id"], ST_CLAIMED) self.store._transition(job["job_id"], ST_EXECUTING) self.store._transition(job["job_id"], ST_OUTCOME_UNKNOWN) return self.store.get_job(job["job_id"]) def _core(self, source_content: Optional[str] = None, read_content: Optional[str] = None, source_fail: bool = False, read_fail: bool = False) -> SaveReconciliationCore: src = source_content if source_content is not None else self.content loader = FakeSourceLoader(src, fail=source_fail) rb = FakeReadBack(read_content, fail=read_fail) core = SaveReconciliationCore(self.store, loader, rb) return core # -- T1: OUTCOME_UNKNOWN + exact target -> confirmed success ------------- def test_t1_exact_target_confirmed_success(self): job = self._create_outcome_unknown() core = self._core(read_content=self.content) result = core.reconcile(job["job_id"], "worker-1") self.assertEqual(result["state"], ST_SUCCEEDED) # Audit: RECONCILIATION_STARTED + TARGET_EXACT + CONFIRMED_EXECUTED trail = self.store.audit_trail(job["job_id"]) events = [e["event_type"] for e in trail if "event_type" in e] self.assertIn(EV_RECONCILIATION_STARTED, events) self.assertIn(EV_RECONCILIATION_TARGET_EXACT, events) self.assertIn(EV_CONFIRMED_EXECUTED, events) # -- T2: exact target -> zero SAVE calls --------------------------------- def test_t2_zero_save_calls(self): job = self._create_outcome_unknown() core = self._core(read_content=self.content) # Reconciliation-Core hat KEIN write-Callable. Wir prüfen, dass der # Core keine write-Methode besitzt und dass der Job SUCCEEDED wird # ohne dass irgendein SAVE aufgerufen werden könnte. self.assertFalse(hasattr(core, "tolaria_save")) self.assertFalse(hasattr(core, "write")) result = core.reconcile(job["job_id"], "worker-1") self.assertEqual(result["state"], ST_SUCCEEDED) # -- T3: attempt_count unchanged ---------------------------------------- def test_t3_attempt_count_unchanged(self): job = self._create_outcome_unknown() self.assertEqual(job["attempt_count"], 0) # direkt via _transition, kein Claim core = self._core(read_content=self.content) result = core.reconcile(job["job_id"], "worker-1") self.assertEqual(result["state"], ST_SUCCEEDED) self.assertEqual(result["attempt_count"], 0) # -- T4: exact target second reconcile -> idempotent --------------------- def test_t4_second_reconcile_idempotent(self): job = self._create_outcome_unknown() core = self._core(read_content=self.content) result1 = core.reconcile(job["job_id"], "worker-1") self.assertEqual(result1["state"], ST_SUCCEEDED) # Zweite Reconciliation: Job ist SUCCEEDED -> bleibt terminal result2 = core.reconcile(job["job_id"], "worker-1") self.assertEqual(result2["state"], ST_SUCCEEDED) # Kein State-Rückschritt, kein neues EXECUTED trail = self.store.audit_trail(job["job_id"]) executed = [e for e in trail if e.get("event_type") == "EXECUTED"] confirmed = [e for e in trail if e.get("event_type") == EV_CONFIRMED_EXECUTED] self.assertEqual(len(executed), 0) # kein neues EXECUTED self.assertGreaterEqual(len(confirmed), 1) # -- T5: target absent -> remains unknown -------------------------------- def test_t5_target_absent_remains_unknown(self): job = self._create_outcome_unknown() core = self._core(read_content=None) # absent result = core.reconcile(job["job_id"], "worker-1") self.assertEqual(result["state"], ST_OUTCOME_UNKNOWN) self.assertEqual(result["result_code"], RC_TARGET_ABSENT) # -- T6: target mismatch content -> fail-closed -------------------------- def test_t6_target_mismatch_content_fail_closed(self): job = self._create_outcome_unknown() core = self._core(read_content="DIFFERENT CONTENT") result = core.reconcile(job["job_id"], "worker-1") self.assertEqual(result["state"], ST_OUTCOME_UNKNOWN) self.assertEqual(result["result_code"], RC_TARGET_MISMATCH) # -- T7: target mismatch object -> fail-closed --------------------------- def test_t7_target_mismatch_object_fail_closed(self): # object_id im Job geändert -> source_loader liefert anderen Content job = self._create_outcome_unknown(object_id="object/other-obj") # Source-Loader liefert Content mit anderer object_id other_content = "---\nid: object/other-obj\ntitle: Other\n---\n\nOther\n" core = self._core(source_content=other_content, read_content=other_content) result = core.reconcile(job["job_id"], "worker-1") # object_id mismatch -> source_loader würde werfen (hier Fake liefert trotzdem) # Wir prüfen: wenn Content != erwartet, dann TARGET_MISMATCH oder PROVENANCE_MISMATCH self.assertEqual(result["state"], ST_OUTCOME_UNKNOWN) self.assertIn(result["result_code"], (RC_TARGET_MISMATCH, RC_PROVENANCE_MISMATCH)) # -- T8: target mismatch hash -> fail-closed ----------------------------- def test_t8_target_mismatch_hash_fail_closed(self): # provenance_hash im Job falsch -> PROVENANCE_MISMATCH job = self._create_outcome_unknown(provenance_hash="f" * 64) core = self._core(read_content=self.content) result = core.reconcile(job["job_id"], "worker-1") self.assertEqual(result["state"], ST_OUTCOME_UNKNOWN) self.assertEqual(result["result_code"], RC_PROVENANCE_MISMATCH) # -- T9: provenance mismatch -> fail-closed ------------------------------ def test_t9_provenance_mismatch_fail_closed(self): job = self._create_outcome_unknown() # Source-Loader liefert anderen Content -> Hash recompute != Job-Hash core = self._core(source_content="DIFFERENT SOURCE", read_content="DIFFERENT SOURCE") result = core.reconcile(job["job_id"], "worker-1") self.assertEqual(result["state"], ST_OUTCOME_UNKNOWN) self.assertEqual(result["result_code"], RC_PROVENANCE_MISMATCH) # -- T10: source commit mismatch -> fail-closed -------------------------- def test_t10_source_commit_mismatch_fail_closed(self): # source_commit im Job geändert -> source_loader liefert anderen Content job = self._create_outcome_unknown(source_commit="b" * 40) other_content = "---\nid: object/test-obj-1\ntitle: Test\n---\n\nOther Commit\n" core = self._core(source_content=other_content, read_content=other_content) result = core.reconcile(job["job_id"], "worker-1") self.assertEqual(result["state"], ST_OUTCOME_UNKNOWN) self.assertIn(result["result_code"], (RC_TARGET_MISMATCH, RC_PROVENANCE_MISMATCH)) # -- T11: read unavailable -> remains unknown ---------------------------- def test_t11_read_unavailable_remains_unknown(self): job = self._create_outcome_unknown() core = self._core(read_fail=True) result = core.reconcile(job["job_id"], "worker-1") self.assertEqual(result["state"], ST_OUTCOME_UNKNOWN) self.assertEqual(result["result_code"], RC_READ_UNAVAILABLE) # -- T12: timeout -> remains unknown ------------------------------------- def test_t12_timeout_remains_unknown(self): job = self._create_outcome_unknown() # Read-Back wirft Timeout-ähnlichen Fehler class TimeoutReadBack: def __call__(self, vault_path): raise TimeoutError("timeout") core = SaveReconciliationCore(self.store, FakeSourceLoader(self.content), TimeoutReadBack()) result = core.reconcile(job["job_id"], "worker-1") self.assertEqual(result["state"], ST_OUTCOME_UNKNOWN) self.assertEqual(result["result_code"], RC_READ_UNAVAILABLE) # -- T13: 401/403 -> remains unknown ------------------------------------- def test_t13_auth_failure_remains_unknown(self): job = self._create_outcome_unknown() class AuthFailReadBack: def __call__(self, vault_path): raise PermissionError("401") core = SaveReconciliationCore(self.store, FakeSourceLoader(self.content), AuthFailReadBack()) result = core.reconcile(job["job_id"], "worker-1") self.assertEqual(result["state"], ST_OUTCOME_UNKNOWN) self.assertEqual(result["result_code"], RC_READ_UNAVAILABLE) # -- T14: malformed job -> rejected -------------------------------------- def test_t14_malformed_job_rejected(self): # Job mit fehlendem vault_path -> Schema-Reject with self.assertRaises(JobRejectedError): make_save_job( job_id=str(uuid.uuid4()), mission_id=str(uuid.uuid4()), object_id=self.object_id, vault_path="", source_commit=self.commit, provenance_hash=self.prov, expected_state="present", created_at="2026-08-27T00:00:00Z", idempotency_key=str(uuid.uuid4()), ) # -- T15: non-OUTCOME_UNKNOWN job cannot use recovery path --------------- def test_t15_non_outcome_unknown_cannot_recover(self): job = self._make_job() self.store.create_job(job) self.store.mark_ready(job["job_id"]) core = self._core(read_content=self.content) with self.assertRaises(ReconciliationError): core.reconcile(job["job_id"], "worker-1") # -- T16: FAILED job cannot be magically reconciled ---------------------- def test_t16_failed_job_cannot_reconcile(self): job = self._make_job() self.store.create_job(job) self.store.mark_ready(job["job_id"]) self.store._transition(job["job_id"], ST_CLAIMED) self.store._transition(job["job_id"], ST_EXECUTING) self.store.mark_failed(job["job_id"], "SOURCE_UNAVAILABLE", "worker-1") core = self._core(read_content=self.content) with self.assertRaises(ReconciliationError): core.reconcile(job["job_id"], "worker-1") # -- T17: SUCCEEDED job remains terminal --------------------------------- def test_t17_succeeded_job_remains_terminal(self): job = self._make_job() self.store.create_job(job) self.store.mark_ready(job["job_id"]) self.store._transition(job["job_id"], ST_CLAIMED) self.store._transition(job["job_id"], ST_EXECUTING) self.store.mark_succeeded(job["job_id"], "worker-1") core = self._core(read_content=self.content) result = core.reconcile(job["job_id"], "worker-1") self.assertEqual(result["state"], ST_SUCCEEDED) # -- T18: immutable fields cannot change --------------------------------- def test_t18_immutable_fields_cannot_change(self): job = self._create_outcome_unknown() # Versuche, vault_path im Payload zu ändern -> Drift-Guard muss greifen # (via _transition, der den Guard aufruft) # Simuliere manipulierten Job: Payload vault_path != Top-Level job["payload"]["vault_path"] = "tolaria/other.md" # _transition liest den Job neu aus der DB (unverändert), daher prüfen # wir den Guard direkt gegen den manipulierten Job with self.assertRaises(JobImmutableFieldError): self.store._assert_immutable_fields(job) # -- T19: no write callable available in reconciliation core ------------- def test_t19_no_write_callable(self): core = self._core() self.assertFalse(hasattr(core, "tolaria_save")) self.assertFalse(hasattr(core, "write")) self.assertFalse(hasattr(core, "save")) # Nur source_loader, read_back, store self.assertTrue(hasattr(core, "source_loader")) self.assertTrue(hasattr(core, "read_back")) self.assertTrue(hasattr(core, "store")) # -- T20: audit preserves OUTCOME_UNKNOWN -------------------------------- def test_t20_audit_preserves_outcome_unknown(self): job = self._create_outcome_unknown() trail = self.store.audit_trail(job["job_id"]) # Der Job-Snapshot im Trail zeigt OUTCOME_UNKNOWN (vor Reconciliation) snapshot = trail[-1] self.assertEqual(snapshot["state"], ST_OUTCOME_UNKNOWN) # -- T21: audit adds reconciliation evidence ---------------------------- def test_t21_audit_adds_reconciliation_evidence(self): job = self._create_outcome_unknown() core = self._core(read_content=self.content) core.reconcile(job["job_id"], "worker-1") trail = self.store.audit_trail(job["job_id"]) events = [e["event_type"] for e in trail if "event_type" in e] self.assertIn(EV_RECONCILIATION_STARTED, events) self.assertIn(EV_RECONCILIATION_TARGET_EXACT, events) self.assertIn(EV_CONFIRMED_EXECUTED, events) # -- T22: no fake EXECUTED mutation event -------------------------------- def test_t22_no_fake_executed_event(self): job = self._create_outcome_unknown() core = self._core(read_content=self.content) core.reconcile(job["job_id"], "worker-1") trail = self.store.audit_trail(job["job_id"]) executed = [e for e in trail if e.get("event_type") == "EXECUTED"] # KEIN EXECUTED-Event (Mutation wurde NICHT neu ausgeführt) self.assertEqual(len(executed), 0) # -- T23: concurrent reconcile safe -------------------------------------- def test_t23_concurrent_reconcile_safe(self): job = self._create_outcome_unknown() core1 = self._core(read_content=self.content) core2 = self._core(read_content=self.content) # Beide reconcilen denselben Job. Der erste finalisiert zu SUCCEEDED, # der zweite sieht SUCCEEDED und bleibt terminal (idempotent). r1 = core1.reconcile(job["job_id"], "worker-1") r2 = core2.reconcile(job["job_id"], "worker-2") self.assertEqual(r1["state"], ST_SUCCEEDED) self.assertEqual(r2["state"], ST_SUCCEEDED) # -- T24: crash after target exact before persist safe ------------------- def test_t24_crash_after_target_exact_before_persist(self): # Simuliere: Reconciliation liest Target, aber Crash vor State-Persist. # Job bleibt OUTCOME_UNKNOWN -> erneute Reconciliation ist sicher. job = self._create_outcome_unknown() core = self._core(read_content=self.content) # Erste Reconciliation "crasht" (wir rufen nur bis zur Klassifikation) # -> hier: wir führen sie einfach aus, dann prüfen wir, dass eine # erneute Reconciliation idempotent ist (kein Doppel-Success). r1 = core.reconcile(job["job_id"], "worker-1") self.assertEqual(r1["state"], ST_SUCCEEDED) # Erneute Reconciliation (Restart-Szenario) r2 = core.reconcile(job["job_id"], "worker-1") self.assertEqual(r2["state"], ST_SUCCEEDED) # -- T25: crash after persist safe --------------------------------------- def test_t25_crash_after_persist_safe(self): # Simuliere: Job ist RECONCILED (persistiert), aber Crash vor SUCCEEDED. job = self._create_outcome_unknown() self.store.mark_reconciled(job["job_id"], "worker-1") core = self._core(read_content=self.content) # Erneute Reconciliation: RECONCILED -> finalisiere zu SUCCEEDED result = core.reconcile(job["job_id"], "worker-1") self.assertEqual(result["state"], ST_SUCCEEDED) # -- T26: restart safe --------------------------------------------------- def test_t26_restart_safe(self): # Neuer Store (Restart) auf einer SEPARATEN DB (damit die gemeinsame # cls.store-DB für die übrigen Tests offen bleibt) tmp2 = tempfile.mkdtemp(prefix="c5-4c3-restart-") db2 = os.path.join(tmp2, "restart.db") store1 = JobStore(db2, "SAVE") job = self._make_job() store1.create_job(job) store1.mark_ready(job["job_id"]) store1._transition(job["job_id"], ST_CLAIMED) store1._transition(job["job_id"], ST_EXECUTING) store1._transition(job["job_id"], ST_OUTCOME_UNKNOWN) store1.close() # Restart: neuer Store auf derselben DB store2 = JobStore(db2, "SAVE") core = SaveReconciliationCore(store2, FakeSourceLoader(self.content), FakeReadBack(self.content)) result = core.reconcile(job["job_id"], "worker-1") self.assertEqual(result["state"], ST_SUCCEEDED) store2.close() shutil.rmtree(tmp2, ignore_errors=True) # -- T27: no duplicate success event ------------------------------------ def test_t27_no_duplicate_success_event(self): job = self._create_outcome_unknown() core = self._core(read_content=self.content) core.reconcile(job["job_id"], "worker-1") core.reconcile(job["job_id"], "worker-1") trail = self.store.audit_trail(job["job_id"]) confirmed = [e for e in trail if e.get("event_type") == EV_CONFIRMED_EXECUTED] # CONFIRMED_EXECUTED darf nicht doppelt für dieselbe Mutation erscheinen # (zweite Reconciliation ist idempotent, aber wir erlauben max. 1 pro # Reconciliation-Durchlauf; hier: 2 Aufrufe -> 2 Events ist ok, aber # kein EXECUTED) executed = [e for e in trail if e.get("event_type") == "EXECUTED"] self.assertEqual(len(executed), 0) # -- T28: idempotency key preserved -------------------------------------- def test_t28_idempotency_key_preserved(self): job = self._create_outcome_unknown() idem = job["idempotency_key"] core = self._core(read_content=self.content) result = core.reconcile(job["job_id"], "worker-1") self.assertEqual(result["idempotency_key"], idem) # -- T29: lease/claim model correct ------------------------------------- def test_t29_lease_claim_model_correct(self): # Reconciliation nutzt KEINEN Execution-Claim (kein attempt_count++). job = self._create_outcome_unknown() self.assertEqual(job["attempt_count"], 0) core = self._core(read_content=self.content) result = core.reconcile(job["job_id"], "worker-1") self.assertEqual(result["attempt_count"], 0) # Kein claim_id/lease_until gesetzt durch Reconciliation self.assertIsNone(result.get("claim_id")) # -- T30: RQ cannot override target fields ------------------------------ def test_t30_rq_cannot_override_target_fields(self): # RQ versucht, vault_path im Job zu überschreiben -> Schema-Reject # (vault_path muss valid sein; ein manipulierter Pfad wird abgelehnt) with self.assertRaises(JobRejectedError): make_save_job( job_id=str(uuid.uuid4()), mission_id=str(uuid.uuid4()), object_id=self.object_id, vault_path="/etc/passwd", source_commit=self.commit, provenance_hash=self.prov, expected_state="present", created_at="2026-08-27T00:00:00Z", idempotency_key=str(uuid.uuid4()), ) if __name__ == "__main__": unittest.main(verbosity=2)