508 lines
23 KiB
Python
508 lines
23 KiB
Python
"""
|
|
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)
|