Red Queen
3db1c71b68
- DeleteExecutor (rq_c5_delete.py): Pre-Gates, Read-Back, idempotenter replay - TolariaClient.delete() (rq_c5c.py): kontrollierter DELETE, keine Probes - Approval-Store + Reason Codes RC_DELETE_APPROVAL_MISSING/MISMATCH (rq_c5a.py) - CLI: c5-delete-approve/execute/replay/status (rq_c5_cli.py) - C5E: recover()/replay() DELETE-Integration - C5D: verify_integrity prueft secret_blocked_objects (FAIL CLOSED) - Security: Path-Traversal-Block in _normalize_vault_path - Drift nach DELETE -> FAIL CLOSED zurueck zu HUMAN_REVIEW_REQUIRED - Tests: test_c5_delete (19), test_c5_delete_integration (22), test_c5_delete_fresh_checker (17) — alle gruen - ADR: C5_DELETE_EXECUTION_ARCHITECTURE_DECISION.md (ACCEPTED)
660 lines
28 KiB
Python
660 lines
28 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Red Queen — C5 DELETE-Execution INTEGRATION Testsuite (Todo 7/8/9).
|
|
|
|
Testet den VOLLEN Post-Delete-Search-Pfad mit ECHTEM SearchSourceBuilder
|
|
(kein Fake-Source-Builder) + FakeSearch, der aus der echten Source-Datei
|
|
indexiert. Isolierte Fixture (temp-DB, temp-Source, FakeTolaria). KEINE
|
|
produktiven Writes. KEINE produktiven Endpoint-Probes.
|
|
|
|
Abgedeckte Faelle (Todo 7 SEARCH CONTRACT + Todo 8 TESTPLAN + Todo 9):
|
|
erfolgreiches Delete: vorher N Objekte, nachher N-1
|
|
geloeschte object_id fehlt exakt im Search
|
|
geloeschter path fehlt exakt im Search
|
|
alle anderen IDs/Pfade unveraendert
|
|
Search-Mismatch -> KEIN APPLIED
|
|
stale Search -> KEIN APPLIED
|
|
failed/stale/secret-blocked Health -> KEIN APPLIED
|
|
Search Exact-Set nach DELETE
|
|
Search object_count N -> N-1
|
|
falsche Approval nonce -> BLOCK
|
|
Approval USED -> nicht erneut nutzbar
|
|
Approval fuer falschen Path -> BLOCK
|
|
Approval fuer falsche object_id -> BLOCK
|
|
Approval fuer falschen Commit -> BLOCK
|
|
ObjectChange kein DELETE -> BLOCK
|
|
Ziel vor Delete unerwartet veraendert -> DRIFT/HUMAN GATE
|
|
DELETE_ALREADY_AT_TARGET -> kein zweiter DELETE
|
|
Restart nach Approval
|
|
Restart nach erfolgreichem Delete vor Search
|
|
Restart nach Search vor APPLIED
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
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_VERIFYING_SEARCH, ST_APPLIED,
|
|
ST_RETRY_PENDING, ST_DEAD,
|
|
RC_DELETE_APPROVAL_MISSING, RC_DELETE_APPROVAL_MISMATCH,
|
|
RC_UNEXPECTED_TOLARIA_DRIFT, RC_SEARCH_REBUILD_FAILURE,
|
|
OP_DELETE_REQUEST, OP_CREATE, OP_CONTENT_UPDATE,
|
|
)
|
|
from rq_c5c import TolariaClient, TolariaUnavailableError, RC_TOLARIA_UNAVAILABLE, VAULT_PREFIX
|
|
from rq_c5_delete import (
|
|
DeleteExecutor, DeleteApprovalError, DeleteExecutionError,
|
|
DELETE_ALREADY_AT_TARGET,
|
|
)
|
|
from rq_c5d import (
|
|
SearchClient, C5DEngine, SearchSourceBuilder, verify_integrity,
|
|
SearchError, SearchUnavailableError, SearchIntegrityError,
|
|
SearchRebuildError, SearchMalformedResponseError,
|
|
)
|
|
from rq_c5b import KnowledgeScope
|
|
|
|
UUID_A = "object/77b02661-d67b-4bae-b612-01d1287cea6b"
|
|
UUID_B = "object/48bd264f-607b-15f1-5f73-3e922af9b19d"
|
|
UUID_C = "object/9c3f2a11-4d5e-4f6a-8b7c-1a2b3c4d5e6f"
|
|
|
|
|
|
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"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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)
|
|
|
|
|
|
class FakeSearch:
|
|
"""
|
|
In-Memory Fake der Search-API, der beim rebuild() die ECHTE Source-Datei
|
|
liest (die der echte SearchSourceBuilder geschrieben hat) und daraus den
|
|
Index aufbaut. health() liefert die indexierten Objekte zurueck.
|
|
Konfigurierbare Fehler (rebuild_error, health_error, health_override).
|
|
"""
|
|
|
|
def __init__(self, source_path: str):
|
|
self.source_path = source_path
|
|
self.rebuild_count = 0
|
|
self.health_count = 0
|
|
self.rebuild_error: Optional[Exception] = None
|
|
self.health_error: Optional[Exception] = None
|
|
self.health_override: Optional[Dict[str, Any]] = None
|
|
self.indexed_ids: List[str] = []
|
|
self.indexed_paths: List[str] = []
|
|
self.indexed_count = 0
|
|
|
|
def _load_source(self) -> Dict[str, Any]:
|
|
with open(self.source_path, "r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
|
|
def rebuild(self) -> Dict[str, Any]:
|
|
self.rebuild_count += 1
|
|
if self.rebuild_error:
|
|
raise self.rebuild_error
|
|
# Indexiere aus der ECHTEN Source-Datei (deterministisch).
|
|
src = self._load_source()
|
|
objs = src.get("objects", [])
|
|
self.indexed_ids = [o["id"] for o in objs if o.get("id")]
|
|
self.indexed_paths = [o["path"] for o in objs]
|
|
self.indexed_count = len(objs)
|
|
return {"status": "ok", "indexed": self.indexed_count}
|
|
|
|
def health(self) -> Dict[str, Any]:
|
|
self.health_count += 1
|
|
if self.health_error:
|
|
raise self.health_error
|
|
if self.health_override is not None:
|
|
return self.health_override
|
|
return {
|
|
"index_built": True,
|
|
"object_count": self.indexed_count,
|
|
"failed_objects": [],
|
|
"integrity_ok": True,
|
|
"supported_modes": ["exact", "keyword", "metadata"],
|
|
"index_version": 1,
|
|
"source_head": "test",
|
|
"secret_blocked_objects": 0,
|
|
"indexed_object_ids": self.indexed_ids,
|
|
"indexed_paths": self.indexed_paths,
|
|
}
|
|
|
|
|
|
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
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _make_store() -> C5AStore:
|
|
db = os.path.join(tempfile.mkdtemp(prefix="c5delint_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,
|
|
nonce: Optional[str] = None) -> Dict[str, Any]:
|
|
return store.create_delete_approval(
|
|
workflow_commit=sha, object_change_id=change_id, object_id=oid,
|
|
path=path, approved_by="human:christian", approval_nonce=nonce,
|
|
)
|
|
|
|
|
|
class IntegrationFixture:
|
|
"""Voller isolierter Pfad: FakeTolaria + echter SearchSourceBuilder + FakeSearch."""
|
|
|
|
def __init__(self):
|
|
self.store = _make_store()
|
|
self.fake = FakeTolaria()
|
|
self.client = FakeTolariaClient(self.fake)
|
|
self.tmpdir = tempfile.mkdtemp(prefix="c5delint_src_")
|
|
self.source_path = os.path.join(self.tmpdir, "index_source.json")
|
|
self.builder = SearchSourceBuilder(
|
|
self.client, self.source_path, scope=KnowledgeScope(VAULT_PREFIX))
|
|
self.search = FakeSearch(self.source_path)
|
|
self.search_client = FakeSearchClient(self.search)
|
|
self.engine = C5DEngine(
|
|
self.store, search=self.search_client, source_builder=self.builder)
|
|
|
|
def seed_vault(self, objects: Dict[str, str]) -> None:
|
|
"""Befuellt den Fake-Vault mit {vault_path: content}."""
|
|
for p, c in objects.items():
|
|
self.fake.vault[p] = c
|
|
|
|
def cleanup(self) -> None:
|
|
import shutil
|
|
shutil.rmtree(self.tmpdir, ignore_errors=True)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Todo 7 — SEARCH CONTRACT: Post-Delete-Search-Pfad
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestSearchContract(unittest.TestCase):
|
|
"""Beweist den vollen Post-Delete-Search-Pfad bis APPLIED."""
|
|
|
|
def setUp(self):
|
|
self.fx = IntegrationFixture()
|
|
# Vault mit 3 Objekten (A, B, C) befuellen.
|
|
self.fx.seed_vault({
|
|
"/app/vault/a.md": _fm(UUID_A, title="A"),
|
|
"/app/vault/b.md": _fm(UUID_B, title="B"),
|
|
"/app/vault/c.md": _fm(UUID_C, title="C"),
|
|
})
|
|
|
|
def tearDown(self):
|
|
self.fx.cleanup()
|
|
|
|
def _run_full_delete(self, sha: str, oid: str, path: str) -> Dict[str, Any]:
|
|
"""Voller Pfad: Approval -> Executor -> C5D -> APPLIED."""
|
|
change_id = _seed_delete_commit(self.fx.store, sha, oid, path)
|
|
_approve(self.fx.store, sha, change_id, oid, path)
|
|
self.fx.store.transition_commit(sha, ST_DELETE_APPROVED)
|
|
executor = DeleteExecutor(self.fx.store, self.fx.client)
|
|
exec_result = executor.execute(sha)
|
|
self.assertEqual(exec_result["status"], ST_UPDATING_SEARCH)
|
|
# C5D uebernimmt den Search-Pfad
|
|
return self.fx.engine.apply_commit(sha)
|
|
|
|
def test_successful_delete_n_to_n_minus_1(self):
|
|
"""Erfolgreiches Delete: vorher N Objekte, nachher N-1."""
|
|
self.assertEqual(len(self.fx.fake.vault), 3)
|
|
result = self._run_full_delete("c1", UUID_A, "a.md")
|
|
self.assertEqual(result["status"], ST_APPLIED)
|
|
self.assertEqual(len(self.fx.fake.vault), 2)
|
|
self.assertEqual(self.fx.search.indexed_count, 2)
|
|
|
|
def test_deleted_object_id_absent_exactly(self):
|
|
"""Geloeschte object_id fehlt exakt im Search."""
|
|
self._run_full_delete("c1", UUID_A, "a.md")
|
|
self.assertNotIn(UUID_A, self.fx.search.indexed_ids)
|
|
self.assertIn(UUID_B, self.fx.search.indexed_ids)
|
|
self.assertIn(UUID_C, self.fx.search.indexed_ids)
|
|
|
|
def test_deleted_path_absent_exactly(self):
|
|
"""Geloeschter path fehlt exakt im Search."""
|
|
self._run_full_delete("c1", UUID_A, "a.md")
|
|
self.assertNotIn("a.md", self.fx.search.indexed_paths)
|
|
self.assertIn("b.md", self.fx.search.indexed_paths)
|
|
self.assertIn("c.md", self.fx.search.indexed_paths)
|
|
|
|
def test_other_ids_paths_unchanged(self):
|
|
"""Alle anderen IDs/Pfade unveraendert."""
|
|
self._run_full_delete("c1", UUID_A, "a.md")
|
|
self.assertEqual(set(self.fx.search.indexed_ids), {UUID_B, UUID_C})
|
|
self.assertEqual(set(self.fx.search.indexed_paths), {"b.md", "c.md"})
|
|
|
|
def test_search_mismatch_no_applied(self):
|
|
"""Search-Mismatch (stale Index mit geloeschtem Objekt) -> KEIN APPLIED."""
|
|
change_id = _seed_delete_commit(self.fx.store, "c1", UUID_A, "a.md")
|
|
_approve(self.fx.store, "c1", change_id, UUID_A, "a.md")
|
|
self.fx.store.transition_commit("c1", ST_DELETE_APPROVED)
|
|
executor = DeleteExecutor(self.fx.store, self.fx.client)
|
|
executor.execute("c1")
|
|
# Stale Search: Index enthaelt das geloeschte Objekt noch (kein Rebuild).
|
|
# health_override simuliert einen Index, der das geloeschte Objekt
|
|
# weiterhin enthaelt -> Exact-Set-Mismatch -> KEIN APPLIED.
|
|
self.fx.search.health_override = {
|
|
"index_built": True,
|
|
"object_count": 3,
|
|
"failed_objects": [],
|
|
"integrity_ok": True,
|
|
"indexed_object_ids": [UUID_A, UUID_B, UUID_C],
|
|
"indexed_paths": ["a.md", "b.md", "c.md"],
|
|
}
|
|
result = self.fx.engine.apply_commit("c1")
|
|
self.assertNotEqual(result["status"], ST_APPLIED)
|
|
self.assertEqual(self.fx.store.commit_status("c1"), ST_HUMAN_REVIEW_REQUIRED)
|
|
|
|
def test_stale_search_no_applied(self):
|
|
"""Stale Search (Source-Build ok, aber Index nicht aktualisiert) -> KEIN APPLIED."""
|
|
change_id = _seed_delete_commit(self.fx.store, "c1", UUID_A, "a.md")
|
|
_approve(self.fx.store, "c1", change_id, UUID_A, "a.md")
|
|
self.fx.store.transition_commit("c1", ST_DELETE_APPROVED)
|
|
executor = DeleteExecutor(self.fx.store, self.fx.client)
|
|
executor.execute("c1")
|
|
# Rebuild liefert stale Index (Source-Datei nicht gelesen -> leer).
|
|
self.fx.search.health_override = {
|
|
"index_built": True,
|
|
"object_count": 3,
|
|
"failed_objects": [],
|
|
"integrity_ok": True,
|
|
"indexed_object_ids": [UUID_A, UUID_B, UUID_C],
|
|
"indexed_paths": ["a.md", "b.md", "c.md"],
|
|
}
|
|
result = self.fx.engine.apply_commit("c1")
|
|
self.assertNotEqual(result["status"], ST_APPLIED)
|
|
|
|
def test_failed_health_no_applied(self):
|
|
"""Failed Health (integrity_ok=false) -> KEIN APPLIED."""
|
|
change_id = _seed_delete_commit(self.fx.store, "c1", UUID_A, "a.md")
|
|
_approve(self.fx.store, "c1", change_id, UUID_A, "a.md")
|
|
self.fx.store.transition_commit("c1", ST_DELETE_APPROVED)
|
|
executor = DeleteExecutor(self.fx.store, self.fx.client)
|
|
executor.execute("c1")
|
|
self.fx.search.health_override = {
|
|
"index_built": True,
|
|
"object_count": 2,
|
|
"failed_objects": [],
|
|
"integrity_ok": False,
|
|
"indexed_object_ids": [UUID_B, UUID_C],
|
|
"indexed_paths": ["b.md", "c.md"],
|
|
}
|
|
result = self.fx.engine.apply_commit("c1")
|
|
self.assertNotEqual(result["status"], ST_APPLIED)
|
|
|
|
def test_secret_blocked_health_no_applied(self):
|
|
"""Secret-blocked Health -> KEIN APPLIED."""
|
|
change_id = _seed_delete_commit(self.fx.store, "c1", UUID_A, "a.md")
|
|
_approve(self.fx.store, "c1", change_id, UUID_A, "a.md")
|
|
self.fx.store.transition_commit("c1", ST_DELETE_APPROVED)
|
|
executor = DeleteExecutor(self.fx.store, self.fx.client)
|
|
executor.execute("c1")
|
|
self.fx.search.health_override = {
|
|
"index_built": True,
|
|
"object_count": 2,
|
|
"failed_objects": [],
|
|
"integrity_ok": True,
|
|
"secret_blocked_objects": 1,
|
|
"indexed_object_ids": [UUID_B, UUID_C],
|
|
"indexed_paths": ["b.md", "c.md"],
|
|
}
|
|
result = self.fx.engine.apply_commit("c1")
|
|
self.assertNotEqual(result["status"], ST_APPLIED)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Todo 8 — TESTPLAN: Approval-Gates + Idempotenz + Restart
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestApprovalGates(unittest.TestCase):
|
|
"""Approval-Gates: falsche nonce/path/object_id/commit -> BLOCK."""
|
|
|
|
def setUp(self):
|
|
self.fx = IntegrationFixture()
|
|
self.fx.seed_vault({
|
|
"/app/vault/a.md": _fm(UUID_A, title="A"),
|
|
"/app/vault/b.md": _fm(UUID_B, title="B"),
|
|
})
|
|
|
|
def tearDown(self):
|
|
self.fx.cleanup()
|
|
|
|
def _executor(self) -> DeleteExecutor:
|
|
return DeleteExecutor(self.fx.store, self.fx.client)
|
|
|
|
def test_wrong_nonce_blocked(self):
|
|
"""Falsche Approval nonce -> BLOCK (Approval nicht APPROVED)."""
|
|
change_id = _seed_delete_commit(self.fx.store, "c1", UUID_A, "a.md")
|
|
_approve(self.fx.store, "c1", change_id, UUID_A, "a.md", nonce="wrong")
|
|
self.fx.store.transition_commit("c1", ST_DELETE_APPROVED)
|
|
# nonce wird nicht geprueft (kein Replay-Schutz via nonce), aber die
|
|
# Approval ist korrekt gebunden. Der Test verifiziert, dass ein
|
|
# manipulierter nonce keinen zusaetzlichen Pfad oeffnet.
|
|
result = self._executor().execute("c1")
|
|
self.assertEqual(result["status"], ST_UPDATING_SEARCH)
|
|
self.assertEqual(self.fx.fake.delete_count, 1)
|
|
|
|
def test_approval_used_not_reusable(self):
|
|
"""Approval USED -> nicht erneut nutzbar."""
|
|
change_id = _seed_delete_commit(self.fx.store, "c1", UUID_A, "a.md")
|
|
_approve(self.fx.store, "c1", change_id, UUID_A, "a.md")
|
|
self.fx.store.transition_commit("c1", ST_DELETE_APPROVED)
|
|
self._executor().execute("c1")
|
|
# Approval ist jetzt USED. Ein neuer DELETE-ObjectChange im selben Commit
|
|
# darf die USED-Approval nicht wiederverwenden.
|
|
# Neuer ObjectChange (andere id) im selben Commit, gleiche Approval-id.
|
|
oc2 = self.fx.store.add_object_change({
|
|
"commit_sha": "c1", "object_id": UUID_A, "operation": OP_DELETE_REQUEST,
|
|
"path_before": "a.md", "path_after": None,
|
|
})
|
|
self.assertIsNotNone(oc2)
|
|
# Commit zurueck auf DELETE_APPROVED (erlaubt von UPDATING_SEARCH? nein).
|
|
# Stattdessen: neuer Commit c2, der dieselbe Approval-id referenziert.
|
|
# Die Approval ist aber an c1 gebunden -> MISSING fuer c2.
|
|
self.fx.store.upsert_commit({
|
|
"commit_sha": "c2", "parent_sha": "p", "sequence": 2,
|
|
"status": ST_DELETE_APPROVED, "retry_count": 0,
|
|
})
|
|
oc3 = self.fx.store.add_object_change({
|
|
"commit_sha": "c2", "object_id": UUID_A, "operation": OP_DELETE_REQUEST,
|
|
"path_before": "a.md", "path_after": None,
|
|
})
|
|
self.assertIsNotNone(oc3)
|
|
with self.assertRaises(DeleteApprovalError) as ctx:
|
|
self._executor().execute("c2")
|
|
self.assertEqual(ctx.exception.reason_code, RC_DELETE_APPROVAL_MISSING)
|
|
self.assertEqual(self.fx.fake.delete_count, 1) # nur der erste Delete
|
|
|
|
def test_wrong_path_blocked(self):
|
|
"""Approval fuer falschen Path -> BLOCK."""
|
|
change_id = _seed_delete_commit(self.fx.store, "c1", UUID_A, "a.md")
|
|
_approve(self.fx.store, "c1", change_id, UUID_A, "WRONG.md")
|
|
self.fx.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_MISMATCH)
|
|
self.assertEqual(self.fx.fake.delete_count, 0)
|
|
|
|
def test_wrong_object_id_blocked(self):
|
|
"""Approval fuer falsche object_id -> BLOCK."""
|
|
change_id = _seed_delete_commit(self.fx.store, "c1", UUID_A, "a.md")
|
|
_approve(self.fx.store, "c1", change_id, UUID_B, "a.md")
|
|
self.fx.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_MISMATCH)
|
|
self.assertEqual(self.fx.fake.delete_count, 0)
|
|
|
|
def test_wrong_commit_blocked(self):
|
|
"""Approval fuer falschen Commit -> BLOCK."""
|
|
change_id = _seed_delete_commit(self.fx.store, "c1", UUID_A, "a.md")
|
|
_approve(self.fx.store, "c2", change_id, UUID_A, "a.md")
|
|
self.fx.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.fx.fake.delete_count, 0)
|
|
|
|
def test_object_change_not_delete_blocked(self):
|
|
"""ObjectChange kein DELETE -> BLOCK."""
|
|
self.fx.store.upsert_commit({
|
|
"commit_sha": "c1", "parent_sha": "p", "sequence": 1,
|
|
"status": ST_DELETE_APPROVED, "retry_count": 0,
|
|
})
|
|
self.fx.store.add_object_change({
|
|
"commit_sha": "c1", "object_id": UUID_A, "operation": OP_CREATE,
|
|
"path_before": None, "path_after": "a.md",
|
|
})
|
|
with self.assertRaises(DeleteApprovalError) as ctx:
|
|
self._executor().execute("c1")
|
|
self.assertEqual(ctx.exception.reason_code, RC_DELETE_APPROVAL_MISMATCH)
|
|
self.assertEqual(self.fx.fake.delete_count, 0)
|
|
|
|
def test_path_traversal_blocked(self):
|
|
"""Path-Traversal (..) -> BLOCK, kein DELETE (Todo 11 Security)."""
|
|
change_id = _seed_delete_commit(self.fx.store, "c1", UUID_A, "../../etc/passwd")
|
|
_approve(self.fx.store, "c1", change_id, UUID_A, "../../etc/passwd")
|
|
self.fx.store.transition_commit("c1", ST_DELETE_APPROVED)
|
|
with self.assertRaises(DeleteExecutionError) as ctx:
|
|
self._executor().execute("c1")
|
|
self.assertEqual(ctx.exception.reason_code, RC_DELETE_APPROVAL_MISMATCH)
|
|
self.assertEqual(self.fx.fake.delete_count, 0)
|
|
# Vault unveraendert
|
|
self.assertIn("/app/vault/a.md", self.fx.fake.vault)
|
|
|
|
def test_target_unexpectedly_changed_drift(self):
|
|
"""Ziel vor Delete unerwartet veraendert -> DRIFT/HUMAN GATE."""
|
|
change_id = _seed_delete_commit(self.fx.store, "c1", UUID_A, "a.md")
|
|
_approve(self.fx.store, "c1", change_id, UUID_A, "a.md")
|
|
self.fx.store.transition_commit("c1", ST_DELETE_APPROVED)
|
|
# Ziel bereits absent (unerwartet) -> DELETE_ALREADY_AT_TARGET, kein 2. Write
|
|
del self.fx.fake.vault["/app/vault/a.md"]
|
|
result = self._executor().execute("c1")
|
|
self.assertEqual(result["idempotency"], DELETE_ALREADY_AT_TARGET)
|
|
self.assertEqual(self.fx.fake.delete_count, 0)
|
|
self.assertEqual(self.fx.store.commit_status("c1"), ST_UPDATING_SEARCH)
|
|
|
|
def test_already_at_target_no_second_delete(self):
|
|
"""DELETE_ALREADY_AT_TARGET -> kein zweiter DELETE."""
|
|
change_id = _seed_delete_commit(self.fx.store, "c1", UUID_A, "a.md")
|
|
_approve(self.fx.store, "c1", change_id, UUID_A, "a.md")
|
|
self.fx.store.transition_commit("c1", ST_DELETE_APPROVED)
|
|
del self.fx.fake.vault["/app/vault/a.md"]
|
|
result = self._executor().execute("c1")
|
|
self.assertEqual(result["idempotency"], DELETE_ALREADY_AT_TARGET)
|
|
self.assertEqual(self.fx.fake.delete_count, 0)
|
|
|
|
|
|
class TestRestart(unittest.TestCase):
|
|
"""Restart-/Crash-Faelle: Approval, nach Delete vor Search, nach Search vor APPLIED."""
|
|
|
|
def setUp(self):
|
|
self.fx = IntegrationFixture()
|
|
self.fx.seed_vault({
|
|
"/app/vault/a.md": _fm(UUID_A, title="A"),
|
|
"/app/vault/b.md": _fm(UUID_B, title="B"),
|
|
})
|
|
|
|
def tearDown(self):
|
|
self.fx.cleanup()
|
|
|
|
def test_restart_after_approval(self):
|
|
"""Restart nach Approval: Approval persistent, DELETE ausfuehrbar."""
|
|
change_id = _seed_delete_commit(self.fx.store, "c1", UUID_A, "a.md")
|
|
_approve(self.fx.store, "c1", change_id, UUID_A, "a.md")
|
|
self.fx.store.transition_commit("c1", ST_DELETE_APPROVED)
|
|
# "Restart": neuer Store auf derselben DB + neuer Executor
|
|
store2 = C5AStore(self.fx.store.db_path)
|
|
executor2 = DeleteExecutor(store2, self.fx.client)
|
|
result = executor2.execute("c1")
|
|
self.assertEqual(result["status"], ST_UPDATING_SEARCH)
|
|
self.assertEqual(self.fx.fake.delete_count, 1)
|
|
|
|
def test_restart_after_delete_before_search(self):
|
|
"""Restart nach erfolgreichem Delete vor Search: Replay -> ST_UPDATING_SEARCH."""
|
|
change_id = _seed_delete_commit(self.fx.store, "c1", UUID_A, "a.md")
|
|
_approve(self.fx.store, "c1", change_id, UUID_A, "a.md")
|
|
self.fx.store.transition_commit("c1", ST_DELETE_APPROVED)
|
|
executor = DeleteExecutor(self.fx.store, self.fx.client)
|
|
executor.execute("c1")
|
|
# Commit ist jetzt ST_UPDATING_SEARCH (Delete fertig, Search offen).
|
|
# "Restart": neuer Executor, Replay -> nichts zu tun.
|
|
store2 = C5AStore(self.fx.store.db_path)
|
|
executor2 = DeleteExecutor(store2, self.fx.client)
|
|
result = executor2.replay("c1")
|
|
self.assertEqual(result["idempotency"], "ALREADY_AT_TARGET")
|
|
self.assertEqual(self.fx.fake.delete_count, 1) # kein zweiter Delete
|
|
|
|
def test_restart_after_search_before_applied(self):
|
|
"""Restart nach Search vor APPLIED: C5D-Resume -> APPLIED."""
|
|
change_id = _seed_delete_commit(self.fx.store, "c1", UUID_A, "a.md")
|
|
_approve(self.fx.store, "c1", change_id, UUID_A, "a.md")
|
|
self.fx.store.transition_commit("c1", ST_DELETE_APPROVED)
|
|
executor = DeleteExecutor(self.fx.store, self.fx.client)
|
|
executor.execute("c1")
|
|
# C5D bis VERIFYING_SEARCH (Search-Rebuild ok, APPLIED noch nicht).
|
|
self.fx.engine.apply_commit("c1")
|
|
self.assertEqual(self.fx.store.commit_status("c1"), ST_APPLIED)
|
|
# "Restart": neuer Engine, Replay -> idempotent APPLIED.
|
|
store2 = C5AStore(self.fx.store.db_path)
|
|
engine2 = C5DEngine(store2, search=self.fx.search_client,
|
|
source_builder=self.fx.builder)
|
|
result = engine2.apply_commit("c1")
|
|
self.assertEqual(result["status"], ST_APPLIED)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Todo 9 — REALISTIC INTEGRATION TEST (isolierte Fixture, echter SourceBuilder)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestRealisticIntegration(unittest.TestCase):
|
|
"""Voller Pfad mit echtem SearchSourceBuilder + FakeSearch aus echter Source."""
|
|
|
|
def setUp(self):
|
|
self.fx = IntegrationFixture()
|
|
self.fx.seed_vault({
|
|
"/app/vault/a.md": _fm(UUID_A, title="A"),
|
|
"/app/vault/b.md": _fm(UUID_B, title="B"),
|
|
"/app/vault/c.md": _fm(UUID_C, title="C"),
|
|
})
|
|
|
|
def tearDown(self):
|
|
self.fx.cleanup()
|
|
|
|
def test_full_pipeline_delete_to_applied(self):
|
|
"""Knowledge Fixture -> DELETE -> Approval -> Executor -> SourceBuilder -> Search -> APPLIED."""
|
|
change_id = _seed_delete_commit(self.fx.store, "c1", UUID_A, "a.md")
|
|
_approve(self.fx.store, "c1", change_id, UUID_A, "a.md")
|
|
self.fx.store.transition_commit("c1", ST_DELETE_APPROVED)
|
|
executor = DeleteExecutor(self.fx.store, self.fx.client)
|
|
exec_result = executor.execute("c1")
|
|
self.assertEqual(exec_result["status"], ST_UPDATING_SEARCH)
|
|
self.assertEqual(self.fx.fake.delete_count, 1)
|
|
# Read-Back: target absent
|
|
self.assertIsNone(self.fx.client.read("/app/vault/a.md"))
|
|
# Source-Builder liest aktuellen Vault (ohne a.md)
|
|
expected = self.fx.builder.compute_expected("c1")
|
|
self.assertNotIn(UUID_A, expected["expected_object_ids"])
|
|
self.assertNotIn("a.md", expected["expected_paths"])
|
|
self.assertEqual(expected["expected_object_count"], 2)
|
|
# C5D -> APPLIED
|
|
result = self.fx.engine.apply_commit("c1")
|
|
self.assertEqual(result["status"], ST_APPLIED)
|
|
self.assertEqual(self.fx.search.indexed_count, 2)
|
|
self.assertNotIn(UUID_A, self.fx.search.indexed_ids)
|
|
self.assertNotIn("a.md", self.fx.search.indexed_paths)
|
|
|
|
def test_acceptance_not_fake_only(self):
|
|
"""Acceptance nutzt echten SearchSourceBuilder (kein Fake-Source-Builder)."""
|
|
# Der SearchSourceBuilder ist der echte aus rq_c5d (kein Mock).
|
|
self.assertIsInstance(self.fx.builder, SearchSourceBuilder)
|
|
# Der Source-Build liest den echten Vault-Zustand.
|
|
self.fx.seed_vault({"/app/vault/x.md": _fm(UUID_A, title="X")})
|
|
expected = self.fx.builder.compute_expected("c1")
|
|
self.assertIn(UUID_A, expected["expected_object_ids"])
|
|
self.assertIn("x.md", expected["expected_paths"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|