1133 lines
50 KiB
Python
1133 lines
50 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Red Queen — C5C Testsuite (Tolaria Propagation & Drift Verification).
|
|
|
|
Testet gegen ein kontrolliertes lokales Git-Fixture-Repo + Fake-Tolaria-API.
|
|
Jeder Test nutzt eine frische temp-DB (tempfile.mkdtemp), nie die Produkt-DB.
|
|
KEINE produktiven Writes. KEINE produktiven Knowledge-Dokumente.
|
|
|
|
Abgedeckte Faelle (§19):
|
|
CREATE success / already target / path collision
|
|
CONTENT_UPDATE success / already target / unexpected drift
|
|
METADATA_UPDATE / STATE_UPDATE / TAGS_UPDATE
|
|
source/canonical ordered success / canonical blocked if source missing / dangling derived_from
|
|
rename/move safe case / rename/move ambiguity
|
|
DELETE_REQUEST -> Human Gate
|
|
SUPERSEDE safe case
|
|
legacy object -> Human Gate
|
|
secret detected -> no write
|
|
Tolaria unavailable -> retry
|
|
Read-back mismatch -> fail closed
|
|
multi-object commit success
|
|
partial commit failure + replay
|
|
idempotent replay
|
|
no Forgejo write
|
|
no Search rebuild
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest import mock
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
# C5C 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_HUMAN_REVIEW_REQUIRED, ST_RETRY_PENDING, ST_DEAD,
|
|
ST_APPLIED, IDEM_ALREADY_AT_TARGET, IDEM_ALREADY_APPLIED, IDEM_RETRY_SAFE,
|
|
RC_UNEXPECTED_TOLARIA_DRIFT, RC_UNKNOWN_OBJECT_ID, RC_SECRET_DETECTED,
|
|
RC_DANGLING_DERIVED_FROM, RC_AMBIGUOUS_DELETE, RC_TOLARIA_UNAVAILABLE,
|
|
OP_CREATE, OP_CONTENT_UPDATE, OP_METADATA_UPDATE, OP_STATE_UPDATE,
|
|
OP_TAGS_UPDATE, OP_SOURCE_CANONICAL_RELATION_UPDATE, OP_RENAME, OP_MOVE,
|
|
OP_DELETE_REQUEST, OP_SUPERSEDE,
|
|
)
|
|
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, TolariaWriteError, ReadBackMismatchError,
|
|
UnexpectedDriftError,
|
|
assert_no_search_calls, assert_no_master_write,
|
|
)
|
|
|
|
UUID_A = "object/77b02661-d67b-4bae-b612-01d1287cea6b"
|
|
UUID_B = "object/48bd264f-607b-15f1-5f73-3e922af9b19d"
|
|
UUID_C = "object/deea2ed6-cd6f-b006-2e4e-949abea60904"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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="c5c_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)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class FakeTolaria:
|
|
"""In-Memory Fake der Tolaria-Vault-API (read/write/verify)."""
|
|
|
|
def __init__(self):
|
|
self.vault: Dict[str, str] = {} # vault_path -> content
|
|
self.write_count = 0
|
|
self.fail_next_write = False
|
|
self.fail_next_read = False
|
|
self.unavailable = False
|
|
|
|
def read(self, vault_path: str) -> Optional[str]:
|
|
if self.unavailable:
|
|
raise TolariaUnavailableError("Tolaria down", RC_TOLARIA_UNAVAILABLE)
|
|
if self.fail_next_read:
|
|
self.fail_next_read = False
|
|
raise TolariaUnavailableError("read timeout", 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)
|
|
if self.fail_next_write:
|
|
self.fail_next_write = False
|
|
raise TolariaUnavailableError("write timeout", 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)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Test-Helfer: C5A-Store + Commit mit ObjectChanges
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _make_store() -> C5AStore:
|
|
db = os.path.join(tempfile.mkdtemp(prefix="c5c_db_"), "c5a.db")
|
|
return C5AStore(db)
|
|
|
|
|
|
def _seed_commit(store: C5AStore, sha: str, parent: Optional[str],
|
|
objs: List[Dict[str, Any]]) -> None:
|
|
"""Legt einen Commit + ObjectChanges im Store an (Status READY)."""
|
|
store.upsert_commit({
|
|
"commit_sha": sha, "parent_sha": parent, "sequence": 1,
|
|
"status": ST_READY, "retry_count": 0,
|
|
})
|
|
for oc in objs:
|
|
oc["commit_sha"] = sha
|
|
store.add_object_change(oc)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestPreWriteDriftCheck(unittest.TestCase):
|
|
"""§5 Pre-Write Drift Check (harte Invariante)."""
|
|
|
|
def setUp(self):
|
|
self.fake = FakeTolaria()
|
|
self.client = FakeTolariaClient(self.fake)
|
|
|
|
def test_write_allowed_when_current_is_before(self):
|
|
before = _fm(UUID_A, state="current") + "body v1"
|
|
target = _fm(UUID_A, state="current") + "body v2"
|
|
self.fake.vault["/app/vault/x.md"] = before
|
|
drift, cur = pre_write_drift_check(self.client, "/app/vault/x.md", before, target)
|
|
self.assertEqual(drift, DRIFT_WRITE_ALLOWED)
|
|
self.assertEqual(cur, before)
|
|
|
|
def test_already_at_target_no_write(self):
|
|
target = _fm(UUID_A, state="current") + "body v2"
|
|
self.fake.vault["/app/vault/x.md"] = target
|
|
drift, cur = pre_write_drift_check(self.client, "/app/vault/x.md", None, target)
|
|
self.assertEqual(drift, DRIFT_ALREADY_AT_TARGET)
|
|
|
|
def test_unexpected_drift_fail_closed(self):
|
|
before = _fm(UUID_A, state="current") + "body v1"
|
|
target = _fm(UUID_A, state="current") + "body v2"
|
|
# Tolaria hat einen DRITTEN Zustand (weder before noch target)
|
|
self.fake.vault["/app/vault/x.md"] = _fm(UUID_A, state="current") + "body v3"
|
|
drift, cur = pre_write_drift_check(self.client, "/app/vault/x.md", before, target)
|
|
self.assertEqual(drift, DRIFT_UNEXPECTED)
|
|
|
|
def test_create_write_allowed_when_absent(self):
|
|
target = _fm(UUID_A, state="current") + "body"
|
|
drift, cur = pre_write_drift_check(self.client, "/app/vault/x.md", None, target)
|
|
self.assertEqual(drift, DRIFT_WRITE_ALLOWED)
|
|
self.assertIsNone(cur)
|
|
|
|
|
|
class TestCreate(unittest.TestCase):
|
|
"""§6 CREATE."""
|
|
|
|
def setUp(self):
|
|
self.fake = FakeTolaria()
|
|
self.client = FakeTolariaClient(self.fake)
|
|
self.repo = FixtureRepo()
|
|
self.store = _make_store()
|
|
self.reader = GitReader(self.repo.dir)
|
|
self.prop = C5CPropagator(self.store, self.reader, self.client)
|
|
|
|
def tearDown(self):
|
|
self.repo.cleanup()
|
|
self.store.close()
|
|
|
|
def _create_commit(self, content: str) -> str:
|
|
self.repo.write("modul-09.md", content)
|
|
return self.repo.commit("create")
|
|
|
|
def test_create_success(self):
|
|
content = _fm(UUID_A, type="arch", role="module", representation="source", state="current")
|
|
sha = self._create_commit(content)
|
|
_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(parse_frontmatter(content)[1]),
|
|
}])
|
|
res = self.prop.propagate_commit(sha)
|
|
self.assertEqual(res["status"], ST_UPDATING_SEARCH)
|
|
self.assertEqual(self.fake.vault["/app/vault/modul-09.md"], content)
|
|
self.assertEqual(self.fake.write_count, 1)
|
|
|
|
def test_create_already_target(self):
|
|
content = _fm(UUID_A, type="arch", role="module", representation="source", state="current")
|
|
sha = self._create_commit(content)
|
|
# Tolaria hat bereits exakt den Ziel-Inhalt
|
|
self.fake.vault["/app/vault/modul-09.md"] = content
|
|
_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(parse_frontmatter(content)[1]),
|
|
}])
|
|
res = self.prop.propagate_commit(sha)
|
|
self.assertEqual(res["status"], ST_UPDATING_SEARCH)
|
|
self.assertEqual(self.fake.write_count, 0) # kein Write
|
|
|
|
def test_create_path_collision(self):
|
|
# Pfad ist bereits mit ANDEREM Inhalt belegt (weder before noch target)
|
|
content = _fm(UUID_A, type="arch", role="module", representation="source", state="current")
|
|
sha = self._create_commit(content)
|
|
self.fake.vault["/app/vault/modul-09.md"] = _fm(UUID_B) + "anderer inhalt"
|
|
_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(parse_frontmatter(content)[1]),
|
|
}])
|
|
res = self.prop.propagate_commit(sha)
|
|
self.assertEqual(res["status"], ST_HUMAN_REVIEW_REQUIRED)
|
|
self.assertEqual(self.fake.write_count, 0) # kein Write bei Drift
|
|
|
|
|
|
class TestContentUpdate(unittest.TestCase):
|
|
"""§7 CONTENT_UPDATE."""
|
|
|
|
def setUp(self):
|
|
self.fake = FakeTolaria()
|
|
self.client = FakeTolariaClient(self.fake)
|
|
self.repo = FixtureRepo()
|
|
self.store = _make_store()
|
|
self.reader = GitReader(self.repo.dir)
|
|
self.prop = C5CPropagator(self.store, self.reader, self.client)
|
|
|
|
def tearDown(self):
|
|
self.repo.cleanup()
|
|
self.store.close()
|
|
|
|
def _content_update_commit(self, before: str, after: str) -> tuple:
|
|
self.repo.write("modul-09.md", before)
|
|
parent = self.repo.commit("v1")
|
|
self.repo.write("modul-09.md", after)
|
|
sha = self.repo.commit("v2")
|
|
return sha, parent
|
|
|
|
def test_content_update_success(self):
|
|
before = _fm(UUID_A, state="current") + "body v1"
|
|
after = _fm(UUID_A, state="current") + "body v2"
|
|
sha, parent = self._content_update_commit(before, after)
|
|
self.fake.vault["/app/vault/modul-09.md"] = before # Tolaria = BEFORE
|
|
_seed_commit(self.store, sha, parent, [{
|
|
"object_id": UUID_A, "operation": OP_CONTENT_UPDATE,
|
|
"path_before": "modul-09.md", "path_after": "modul-09.md",
|
|
"content_hash_before": content_hash(parse_frontmatter(before)[1]),
|
|
"content_hash_after": content_hash(parse_frontmatter(after)[1]),
|
|
}])
|
|
res = self.prop.propagate_commit(sha)
|
|
self.assertEqual(res["status"], ST_UPDATING_SEARCH)
|
|
self.assertEqual(self.fake.vault["/app/vault/modul-09.md"], after)
|
|
self.assertEqual(self.fake.write_count, 1)
|
|
|
|
def test_content_update_already_target(self):
|
|
before = _fm(UUID_A, state="current") + "body v1"
|
|
after = _fm(UUID_A, state="current") + "body v2"
|
|
sha, parent = self._content_update_commit(before, after)
|
|
self.fake.vault["/app/vault/modul-09.md"] = after # bereits am Ziel
|
|
_seed_commit(self.store, sha, parent, [{
|
|
"object_id": UUID_A, "operation": OP_CONTENT_UPDATE,
|
|
"path_before": "modul-09.md", "path_after": "modul-09.md",
|
|
"content_hash_before": content_hash(parse_frontmatter(before)[1]),
|
|
"content_hash_after": content_hash(parse_frontmatter(after)[1]),
|
|
}])
|
|
res = self.prop.propagate_commit(sha)
|
|
self.assertEqual(res["status"], ST_UPDATING_SEARCH)
|
|
self.assertEqual(self.fake.write_count, 0)
|
|
|
|
def test_content_update_unexpected_drift(self):
|
|
before = _fm(UUID_A, state="current") + "body v1"
|
|
after = _fm(UUID_A, state="current") + "body v2"
|
|
sha, parent = self._content_update_commit(before, after)
|
|
# Tolaria hat einen DRITTEN Zustand
|
|
self.fake.vault["/app/vault/modul-09.md"] = _fm(UUID_A, state="current") + "body v3"
|
|
_seed_commit(self.store, sha, parent, [{
|
|
"object_id": UUID_A, "operation": OP_CONTENT_UPDATE,
|
|
"path_before": "modul-09.md", "path_after": "modul-09.md",
|
|
"content_hash_before": content_hash(parse_frontmatter(before)[1]),
|
|
"content_hash_after": content_hash(parse_frontmatter(after)[1]),
|
|
}])
|
|
res = self.prop.propagate_commit(sha)
|
|
self.assertEqual(res["status"], ST_HUMAN_REVIEW_REQUIRED)
|
|
self.assertEqual(self.fake.write_count, 0) # kein Ueberschreiben
|
|
|
|
|
|
class TestMetadataStateTags(unittest.TestCase):
|
|
"""§8 METADATA / STATE / TAGS."""
|
|
|
|
def setUp(self):
|
|
self.fake = FakeTolaria()
|
|
self.client = FakeTolariaClient(self.fake)
|
|
self.repo = FixtureRepo()
|
|
self.store = _make_store()
|
|
self.reader = GitReader(self.repo.dir)
|
|
self.prop = C5CPropagator(self.store, self.reader, self.client)
|
|
|
|
def tearDown(self):
|
|
self.repo.cleanup()
|
|
self.store.close()
|
|
|
|
def _update_commit(self, before: str, after: str) -> tuple:
|
|
self.repo.write("modul-09.md", before)
|
|
parent = self.repo.commit("v1")
|
|
self.repo.write("modul-09.md", after)
|
|
sha = self.repo.commit("v2")
|
|
return sha, parent
|
|
|
|
def test_metadata_update(self):
|
|
before = _fm(UUID_A, type="arch", role="module", representation="source", state="current") + "body"
|
|
after = _fm(UUID_A, type="arch", role="module", representation="source", state="current", summary="neu") + "body"
|
|
sha, parent = self._update_commit(before, after)
|
|
self.fake.vault["/app/vault/modul-09.md"] = before
|
|
_seed_commit(self.store, sha, parent, [{
|
|
"object_id": UUID_A, "operation": OP_METADATA_UPDATE,
|
|
"path_before": "modul-09.md", "path_after": "modul-09.md",
|
|
"metadata_hash_before": metadata_hash(parse_frontmatter(before)[0]),
|
|
"metadata_hash_after": metadata_hash(parse_frontmatter(after)[0]),
|
|
}])
|
|
res = self.prop.propagate_commit(sha)
|
|
self.assertEqual(res["status"], ST_UPDATING_SEARCH)
|
|
self.assertEqual(self.fake.vault["/app/vault/modul-09.md"], after)
|
|
|
|
def test_state_update(self):
|
|
before = _fm(UUID_A, state="current") + "body"
|
|
after = _fm(UUID_A, state="historical") + "body"
|
|
sha, parent = self._update_commit(before, after)
|
|
self.fake.vault["/app/vault/modul-09.md"] = before
|
|
_seed_commit(self.store, sha, parent, [{
|
|
"object_id": UUID_A, "operation": OP_STATE_UPDATE,
|
|
"path_before": "modul-09.md", "path_after": "modul-09.md",
|
|
"metadata_hash_before": metadata_hash(parse_frontmatter(before)[0]),
|
|
"metadata_hash_after": metadata_hash(parse_frontmatter(after)[0]),
|
|
}])
|
|
res = self.prop.propagate_commit(sha)
|
|
self.assertEqual(res["status"], ST_UPDATING_SEARCH)
|
|
self.assertEqual(self.fake.vault["/app/vault/modul-09.md"], after)
|
|
|
|
def test_tags_update(self):
|
|
before = _fm(UUID_A, tags=["a"]) + "body"
|
|
after = _fm(UUID_A, tags=["a", "b"]) + "body"
|
|
sha, parent = self._update_commit(before, after)
|
|
self.fake.vault["/app/vault/modul-09.md"] = before
|
|
_seed_commit(self.store, sha, parent, [{
|
|
"object_id": UUID_A, "operation": OP_TAGS_UPDATE,
|
|
"path_before": "modul-09.md", "path_after": "modul-09.md",
|
|
"metadata_hash_before": metadata_hash(parse_frontmatter(before)[0]),
|
|
"metadata_hash_after": metadata_hash(parse_frontmatter(after)[0]),
|
|
}])
|
|
res = self.prop.propagate_commit(sha)
|
|
self.assertEqual(res["status"], ST_UPDATING_SEARCH)
|
|
self.assertEqual(self.fake.vault["/app/vault/modul-09.md"], after)
|
|
|
|
|
|
class TestSourceCanonical(unittest.TestCase):
|
|
"""§9 Source/Canonical Relation."""
|
|
|
|
def setUp(self):
|
|
self.fake = FakeTolaria()
|
|
self.client = FakeTolariaClient(self.fake)
|
|
self.repo = FixtureRepo()
|
|
self.store = _make_store()
|
|
self.reader = GitReader(self.repo.dir)
|
|
self.prop = C5CPropagator(self.store, self.reader, self.client)
|
|
|
|
def tearDown(self):
|
|
self.repo.cleanup()
|
|
self.store.close()
|
|
|
|
def test_source_canonical_ordered_success(self):
|
|
# Source (Root) zuerst, dann Canonical (derived_from=Source)
|
|
source = _fm(UUID_A, type="arch", role="module", representation="source", state="current") + "body"
|
|
canonical = _fm(UUID_B, type="arch", role="module", representation="canonical",
|
|
state="current", derived_from=UUID_A) + "body"
|
|
self.repo.write("modul-09.md", source)
|
|
self.repo.write("notes/trading/system-docs/modul-09.md", canonical)
|
|
sha = self.repo.commit("pair")
|
|
# Beide Objekte im selben Commit
|
|
_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(parse_frontmatter(source)[1])},
|
|
{"object_id": UUID_B, "operation": OP_CREATE, "path_before": None,
|
|
"path_after": "notes/trading/system-docs/modul-09.md",
|
|
"content_hash_after": content_hash(parse_frontmatter(canonical)[1])},
|
|
])
|
|
res = self.prop.propagate_commit(sha)
|
|
self.assertEqual(res["status"], ST_UPDATING_SEARCH)
|
|
self.assertEqual(self.fake.vault["/app/vault/modul-09.md"], source)
|
|
self.assertEqual(self.fake.vault["/app/vault/notes/trading/system-docs/modul-09.md"], canonical)
|
|
|
|
def test_canonical_blocked_if_source_missing(self):
|
|
# Canonical referenziert Source, die NICHT existiert -> dangling
|
|
canonical = _fm(UUID_B, type="arch", role="module", representation="canonical",
|
|
state="current", derived_from=UUID_A) + "body"
|
|
self.repo.write("notes/trading/system-docs/modul-09.md", canonical)
|
|
sha = self.repo.commit("canonical-only")
|
|
_seed_commit(self.store, sha, None, [{
|
|
"object_id": UUID_B, "operation": OP_CREATE, "path_before": None,
|
|
"path_after": "notes/trading/system-docs/modul-09.md",
|
|
"content_hash_after": content_hash(parse_frontmatter(canonical)[1]),
|
|
}])
|
|
res = self.prop.propagate_commit(sha)
|
|
self.assertEqual(res["status"], ST_HUMAN_REVIEW_REQUIRED)
|
|
self.assertEqual(self.fake.write_count, 0) # kein Write bei dangling
|
|
|
|
def test_dangling_derived_from(self):
|
|
# derived_from zeigt auf nicht-existierende ID
|
|
content = _fm(UUID_B, type="arch", role="module", representation="canonical",
|
|
state="current", derived_from="object/00000000-0000-0000-0000-000000000000") + "body"
|
|
self.repo.write("modul-09.md", content)
|
|
sha = self.repo.commit("dangling")
|
|
_seed_commit(self.store, sha, None, [{
|
|
"object_id": UUID_B, "operation": OP_CREATE, "path_before": None,
|
|
"path_after": "modul-09.md",
|
|
"content_hash_after": content_hash(parse_frontmatter(content)[1]),
|
|
}])
|
|
res = self.prop.propagate_commit(sha)
|
|
self.assertEqual(res["status"], ST_HUMAN_REVIEW_REQUIRED)
|
|
self.assertEqual(self.fake.write_count, 0)
|
|
|
|
|
|
class TestRenameMove(unittest.TestCase):
|
|
"""§10 RENAME / MOVE -> Human Gate (sichere v1-Policy)."""
|
|
|
|
def setUp(self):
|
|
self.fake = FakeTolaria()
|
|
self.client = FakeTolariaClient(self.fake)
|
|
self.repo = FixtureRepo()
|
|
self.store = _make_store()
|
|
self.reader = GitReader(self.repo.dir)
|
|
self.prop = C5CPropagator(self.store, self.reader, self.client)
|
|
|
|
def tearDown(self):
|
|
self.repo.cleanup()
|
|
self.store.close()
|
|
|
|
def test_rename_safe_case_human_gate(self):
|
|
content = _fm(UUID_A, state="current") + "body"
|
|
self.repo.write("modul-09.md", content)
|
|
parent = self.repo.commit("v1")
|
|
self.repo.write("modul-09-new.md", content)
|
|
self.repo.write("modul-09.md", "") # alt entfernen
|
|
sha = self.repo.commit("rename")
|
|
_seed_commit(self.store, sha, parent, [{
|
|
"object_id": UUID_A, "operation": OP_RENAME,
|
|
"path_before": "modul-09.md", "path_after": "modul-09-new.md",
|
|
}])
|
|
res = self.prop.propagate_commit(sha)
|
|
self.assertEqual(res["status"], ST_HUMAN_REVIEW_REQUIRED)
|
|
self.assertEqual(self.fake.write_count, 0) # kein automatischer Rename
|
|
|
|
def test_move_ambiguity_human_gate(self):
|
|
content = _fm(UUID_A, state="current") + "body"
|
|
self.repo.write("modul-09.md", content)
|
|
parent = self.repo.commit("v1")
|
|
self.repo.write("sub/modul-09.md", content)
|
|
self.repo.write("modul-09.md", "")
|
|
sha = self.repo.commit("move")
|
|
_seed_commit(self.store, sha, parent, [{
|
|
"object_id": UUID_A, "operation": OP_MOVE,
|
|
"path_before": "modul-09.md", "path_after": "sub/modul-09.md",
|
|
}])
|
|
res = self.prop.propagate_commit(sha)
|
|
self.assertEqual(res["status"], ST_HUMAN_REVIEW_REQUIRED)
|
|
self.assertEqual(self.fake.write_count, 0)
|
|
|
|
|
|
class TestDeleteSupersedeLegacy(unittest.TestCase):
|
|
"""§11 DELETE, §12 SUPERSEDE, §13 LEGACY."""
|
|
|
|
def setUp(self):
|
|
self.fake = FakeTolaria()
|
|
self.client = FakeTolariaClient(self.fake)
|
|
self.repo = FixtureRepo()
|
|
self.store = _make_store()
|
|
self.reader = GitReader(self.repo.dir)
|
|
self.prop = C5CPropagator(self.store, self.reader, self.client)
|
|
|
|
def tearDown(self):
|
|
self.repo.cleanup()
|
|
self.store.close()
|
|
|
|
def test_delete_request_human_gate(self):
|
|
content = _fm(UUID_A, state="current") + "body"
|
|
self.repo.write("modul-09.md", content)
|
|
parent = self.repo.commit("v1")
|
|
self.repo.write("modul-09.md", "")
|
|
sha = self.repo.commit("delete")
|
|
_seed_commit(self.store, sha, parent, [{
|
|
"object_id": UUID_A, "operation": OP_DELETE_REQUEST,
|
|
"path_before": "modul-09.md", "path_after": None,
|
|
}])
|
|
res = self.prop.propagate_commit(sha)
|
|
self.assertEqual(res["status"], ST_HUMAN_REVIEW_REQUIRED)
|
|
self.assertEqual(self.fake.write_count, 0) # kein Hard Delete
|
|
|
|
def test_supersede_safe_case(self):
|
|
before = _fm(UUID_A, state="current") + "body"
|
|
after = _fm(UUID_A, state="superseded") + "body"
|
|
self.repo.write("modul-09.md", before)
|
|
parent = self.repo.commit("v1")
|
|
self.repo.write("modul-09.md", after)
|
|
sha = self.repo.commit("supersede")
|
|
self.fake.vault["/app/vault/modul-09.md"] = before
|
|
_seed_commit(self.store, sha, parent, [{
|
|
"object_id": UUID_A, "operation": OP_SUPERSEDE,
|
|
"path_before": "modul-09.md", "path_after": "modul-09.md",
|
|
"metadata_hash_before": metadata_hash(parse_frontmatter(before)[0]),
|
|
"metadata_hash_after": metadata_hash(parse_frontmatter(after)[0]),
|
|
}])
|
|
res = self.prop.propagate_commit(sha)
|
|
self.assertEqual(res["status"], ST_UPDATING_SEARCH)
|
|
self.assertEqual(self.fake.vault["/app/vault/modul-09.md"], after)
|
|
|
|
def test_legacy_object_human_gate(self):
|
|
# Legacy-Objekt (keine gueltige object_id) -> Human Gate
|
|
content = "---\nknowledge_schema: 1\nid: legacy\n---\nbody"
|
|
self.repo.write("README.md", content)
|
|
sha = self.repo.commit("legacy")
|
|
_seed_commit(self.store, sha, None, [{
|
|
"object_id": None, "operation": OP_METADATA_UPDATE,
|
|
"path_before": None, "path_after": "README.md",
|
|
"reason_code": "UNKNOWN_OBJECT_ID",
|
|
}])
|
|
res = self.prop.propagate_commit(sha)
|
|
self.assertEqual(res["status"], ST_HUMAN_REVIEW_REQUIRED)
|
|
self.assertEqual(self.fake.write_count, 0)
|
|
|
|
|
|
class TestSecretSafety(unittest.TestCase):
|
|
"""Secret detected -> no write."""
|
|
|
|
def setUp(self):
|
|
self.fake = FakeTolaria()
|
|
self.client = FakeTolariaClient(self.fake)
|
|
self.repo = FixtureRepo()
|
|
self.store = _make_store()
|
|
self.reader = GitReader(self.repo.dir)
|
|
self.prop = C5CPropagator(self.store, self.reader, self.client)
|
|
|
|
def tearDown(self):
|
|
self.repo.cleanup()
|
|
self.store.close()
|
|
|
|
def test_secret_detected_no_write(self):
|
|
content = _fm(UUID_A, state="current") + "body\nkey: sk-1234567890abcdefghijklmnopqrstuvwxyz"
|
|
self.repo.write("modul-09.md", content)
|
|
sha = self.repo.commit("secret")
|
|
_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(parse_frontmatter(content)[1]),
|
|
}])
|
|
res = self.prop.propagate_commit(sha)
|
|
self.assertEqual(res["status"], ST_HUMAN_REVIEW_REQUIRED)
|
|
self.assertEqual(self.fake.write_count, 0)
|
|
|
|
|
|
class TestRetry(unittest.TestCase):
|
|
"""§17 Retry: nur technische Fehler retrybar."""
|
|
|
|
def setUp(self):
|
|
self.fake = FakeTolaria()
|
|
self.client = FakeTolariaClient(self.fake)
|
|
self.repo = FixtureRepo()
|
|
self.store = _make_store()
|
|
self.reader = GitReader(self.repo.dir)
|
|
self.prop = C5CPropagator(self.store, self.reader, self.client)
|
|
|
|
def tearDown(self):
|
|
self.repo.cleanup()
|
|
self.store.close()
|
|
|
|
def test_tolaria_unavailable_retry(self):
|
|
content = _fm(UUID_A, state="current") + "body"
|
|
self.repo.write("modul-09.md", content)
|
|
sha = self.repo.commit("create")
|
|
_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(parse_frontmatter(content)[1]),
|
|
}])
|
|
# Tolaria down -> retrybar
|
|
self.fake.unavailable = True
|
|
res = self.prop.propagate_commit(sha)
|
|
self.assertEqual(res["status"], ST_RETRY_PENDING)
|
|
self.assertEqual(res["results"][0]["retry_count"], 1) # _handle_failure hat increment_retry aufgerufen
|
|
|
|
def test_retry_replay_after_recovery(self):
|
|
# Commit in RETRY_PENDING -> Tolaria wieder up -> Replay moeglich
|
|
content = _fm(UUID_A, state="current") + "body"
|
|
self.repo.write("modul-09.md", content)
|
|
sha = self.repo.commit("create")
|
|
_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(parse_frontmatter(content)[1]),
|
|
}])
|
|
# Erster Lauf: Tolaria down -> RETRY_PENDING
|
|
self.fake.unavailable = True
|
|
res1 = self.prop.propagate_commit(sha)
|
|
self.assertEqual(res1["status"], ST_RETRY_PENDING)
|
|
# Zweiter Lauf: Tolaria up -> Replay muss PROPAGATING_TOLARIA erreichen
|
|
self.fake.unavailable = False
|
|
res2 = self.prop.propagate_commit(sha)
|
|
self.assertEqual(res2["status"], ST_UPDATING_SEARCH) # vollstaendig propagiert
|
|
self.assertEqual(self.fake.write_count, 1) # genau ein Write
|
|
|
|
def test_max_retries_reaches_dead(self):
|
|
# max_retries=1 -> nach 1 Fehlschlag DEAD
|
|
content = _fm(UUID_A, state="current") + "body"
|
|
self.repo.write("modul-09.md", content)
|
|
sha = self.repo.commit("create")
|
|
_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(parse_frontmatter(content)[1]),
|
|
}])
|
|
self.fake.unavailable = True
|
|
prop = C5CPropagator(self.store, self.reader, self.client, max_retries=1)
|
|
res = prop.propagate_commit(sha)
|
|
self.assertEqual(res["status"], ST_DEAD)
|
|
self.assertEqual(self.store.commit_status(sha), ST_DEAD)
|
|
|
|
def test_drift_not_retryable(self):
|
|
# Drift ist NICHT retrybar -> direkt Human Gate
|
|
before = _fm(UUID_A, state="current") + "body v1"
|
|
after = _fm(UUID_A, state="current") + "body v2"
|
|
self.repo.write("modul-09.md", before)
|
|
parent = self.repo.commit("v1")
|
|
self.repo.write("modul-09.md", after)
|
|
sha = self.repo.commit("v2")
|
|
self.fake.vault["/app/vault/modul-09.md"] = _fm(UUID_A, state="current") + "body v3"
|
|
_seed_commit(self.store, sha, parent, [{
|
|
"object_id": UUID_A, "operation": OP_CONTENT_UPDATE,
|
|
"path_before": "modul-09.md", "path_after": "modul-09.md",
|
|
"content_hash_before": content_hash(parse_frontmatter(before)[1]),
|
|
"content_hash_after": content_hash(parse_frontmatter(after)[1]),
|
|
}])
|
|
res = self.prop.propagate_commit(sha)
|
|
self.assertEqual(res["status"], ST_HUMAN_REVIEW_REQUIRED)
|
|
self.assertEqual(self.store.increment_retry(sha), 1) # kein Retry-Increment durch C5C
|
|
|
|
|
|
class TestReadBackMismatch(unittest.TestCase):
|
|
"""Read-back mismatch -> fail closed."""
|
|
|
|
def setUp(self):
|
|
self.fake = FakeTolaria()
|
|
self.client = FakeTolariaClient(self.fake)
|
|
self.repo = FixtureRepo()
|
|
self.store = _make_store()
|
|
self.reader = GitReader(self.repo.dir)
|
|
self.prop = C5CPropagator(self.store, self.reader, self.client)
|
|
|
|
def tearDown(self):
|
|
self.repo.cleanup()
|
|
self.store.close()
|
|
|
|
def test_readback_mismatch_fail_closed(self):
|
|
content = _fm(UUID_A, state="current") + "body"
|
|
self.repo.write("modul-09.md", content)
|
|
sha = self.repo.commit("create")
|
|
_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(parse_frontmatter(content)[1]),
|
|
}])
|
|
# Fake schreibt, aber Read-Back liefert anderen Inhalt (Simulation)
|
|
self.fake.vault["/app/vault/modul-09.md"] = _fm(UUID_A, state="current") + "ANDERER body"
|
|
res = self.prop.propagate_commit(sha)
|
|
self.assertEqual(res["status"], ST_HUMAN_REVIEW_REQUIRED)
|
|
|
|
|
|
class TestCommitAtomicity(unittest.TestCase):
|
|
"""§15 Commit-Atomicity + Replay."""
|
|
|
|
def setUp(self):
|
|
self.fake = FakeTolaria()
|
|
self.client = FakeTolariaClient(self.fake)
|
|
self.repo = FixtureRepo()
|
|
self.store = _make_store()
|
|
self.reader = GitReader(self.repo.dir)
|
|
self.prop = C5CPropagator(self.store, self.reader, self.client)
|
|
|
|
def tearDown(self):
|
|
self.repo.cleanup()
|
|
self.store.close()
|
|
|
|
def test_multi_object_commit_success(self):
|
|
c1 = _fm(UUID_A, state="current") + "body A"
|
|
c2 = _fm(UUID_B, state="current") + "body B"
|
|
self.repo.write("modul-09.md", c1)
|
|
self.repo.write("modul-10.md", c2)
|
|
sha = self.repo.commit("multi")
|
|
_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(parse_frontmatter(c1)[1])},
|
|
{"object_id": UUID_B, "operation": OP_CREATE, "path_before": None,
|
|
"path_after": "modul-10.md",
|
|
"content_hash_after": content_hash(parse_frontmatter(c2)[1])},
|
|
])
|
|
res = self.prop.propagate_commit(sha)
|
|
self.assertEqual(res["status"], ST_UPDATING_SEARCH)
|
|
self.assertEqual(self.fake.write_count, 2)
|
|
|
|
def test_partial_commit_failure(self):
|
|
# Objekt 1 ok, Objekt 2 drift -> Commit NICHT als complete markieren
|
|
c1 = _fm(UUID_A, state="current") + "body A"
|
|
c2 = _fm(UUID_B, state="current") + "body B"
|
|
self.repo.write("modul-09.md", c1)
|
|
self.repo.write("modul-10.md", c2)
|
|
sha = self.repo.commit("multi")
|
|
# Objekt 2 hat Drift in Tolaria
|
|
self.fake.vault["/app/vault/modul-10.md"] = _fm(UUID_B, state="current") + "DRIFT"
|
|
_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(parse_frontmatter(c1)[1])},
|
|
{"object_id": UUID_B, "operation": OP_CREATE, "path_before": None,
|
|
"path_after": "modul-10.md",
|
|
"content_hash_after": content_hash(parse_frontmatter(c2)[1])},
|
|
])
|
|
res = self.prop.propagate_commit(sha)
|
|
self.assertEqual(res["status"], ST_HUMAN_REVIEW_REQUIRED)
|
|
# Commit NICHT als Tolaria-complete (nicht UPDATING_SEARCH)
|
|
self.assertNotEqual(res["status"], ST_UPDATING_SEARCH)
|
|
|
|
def test_idempotent_replay(self):
|
|
# Nach erfolgreichem Write: Replay erkennt ALREADY_AT_TARGET, kein Doppel-Write
|
|
content = _fm(UUID_A, state="current") + "body"
|
|
self.repo.write("modul-09.md", content)
|
|
sha = self.repo.commit("create")
|
|
_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(parse_frontmatter(content)[1]),
|
|
}])
|
|
# Erster Lauf: schreibt
|
|
res1 = self.prop.propagate_commit(sha)
|
|
self.assertEqual(res1["status"], ST_UPDATING_SEARCH)
|
|
writes_after_first = self.fake.write_count
|
|
# Zweiter Lauf (Replay): Tolaria hat bereits Ziel -> kein Write
|
|
res2 = self.prop.propagate_commit(sha)
|
|
self.assertEqual(res2["status"], ST_UPDATING_SEARCH)
|
|
self.assertEqual(self.fake.write_count, writes_after_first) # kein Doppel-Write
|
|
|
|
|
|
class TestGuarantees(unittest.TestCase):
|
|
"""No-Search / No-Master-Write Guarantee (statisch)."""
|
|
|
|
def test_no_search_calls(self):
|
|
res = assert_no_search_calls()
|
|
self.assertTrue(res["no_search_calls"], res)
|
|
|
|
def test_no_master_write(self):
|
|
res = assert_no_master_write()
|
|
self.assertTrue(res["no_master_write"], res)
|
|
|
|
|
|
class TestDryRun(unittest.TestCase):
|
|
"""§20 Live-Dry-Run (read-only, kein Write)."""
|
|
|
|
def setUp(self):
|
|
self.fake = FakeTolaria()
|
|
self.client = FakeTolariaClient(self.fake)
|
|
self.repo = FixtureRepo()
|
|
self.store = _make_store()
|
|
self.reader = GitReader(self.repo.dir)
|
|
self.dry = C5CDryRun(self.store, self.reader, self.client)
|
|
|
|
def tearDown(self):
|
|
self.repo.cleanup()
|
|
self.store.close()
|
|
|
|
def test_dry_run_plan(self):
|
|
content = _fm(UUID_A, state="current") + "body"
|
|
self.repo.write("modul-09.md", content)
|
|
sha = self.repo.commit("create")
|
|
_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(parse_frontmatter(content)[1]),
|
|
}])
|
|
plan = self.dry.plan_commit(sha)
|
|
self.assertEqual(plan["plan"][0]["plan"], PLAN_WOULD_WRITE)
|
|
self.assertEqual(self.fake.write_count, 0) # KEIN Write im Dry-Run
|
|
|
|
def test_dry_run_already_at_target(self):
|
|
content = _fm(UUID_A, state="current") + "body"
|
|
self.repo.write("modul-09.md", content)
|
|
sha = self.repo.commit("create")
|
|
self.fake.vault["/app/vault/modul-09.md"] = content
|
|
_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(parse_frontmatter(content)[1]),
|
|
}])
|
|
plan = self.dry.plan_commit(sha)
|
|
self.assertEqual(plan["plan"][0]["plan"], PLAN_ALREADY_AT_TARGET)
|
|
self.assertEqual(self.fake.write_count, 0)
|
|
|
|
def test_dry_run_drift(self):
|
|
content = _fm(UUID_A, state="current") + "body"
|
|
self.repo.write("modul-09.md", content)
|
|
sha = self.repo.commit("create")
|
|
self.fake.vault["/app/vault/modul-09.md"] = _fm(UUID_A, state="current") + "DRIFT"
|
|
_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(parse_frontmatter(content)[1]),
|
|
}])
|
|
plan = self.dry.plan_commit(sha)
|
|
self.assertEqual(plan["plan"][0]["plan"], PLAN_DRIFT)
|
|
self.assertEqual(self.fake.write_count, 0)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# C5C HOTFIX: Not-Found-Read-Semantik (regression)
|
|
# Testet die ECHTE TolariaClient.read() gegen nachgestelltes HTTP-Verhalten.
|
|
# Kein Fake-Client, kein Netzwerk: _post wird gemockt.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestTolariaReadNotFoundSemantic(unittest.TestCase):
|
|
"""C5C-Hotfix: fehlendes Objekt -> None; alle anderen Fehler fail-closed."""
|
|
|
|
def setUp(self):
|
|
self.client = TolariaClient(base_url="http://fake")
|
|
|
|
def _post(self, status: int, body: str, reason_code=None):
|
|
"""Liefert einen Callable, der ein TolariaWriteError/JSON wirft/liefert."""
|
|
def fake_post(endpoint, payload):
|
|
if status == 200:
|
|
return json.loads(body)
|
|
raise TolariaWriteError(
|
|
f"Tolaria HTTP {status} auf {endpoint}: {body}",
|
|
reason_code or "INVALID_SCHEMA",
|
|
http_code=status,
|
|
)
|
|
return fake_post
|
|
|
|
# A) EXISTING OBJECT -> HTTP 200 -> content zurück
|
|
def test_a_existing_returns_content(self):
|
|
content = "hello vault"
|
|
with mock.patch.object(self.client, "_post",
|
|
side_effect=self._post(200, json.dumps({"content": content}))):
|
|
self.assertEqual(self.client.read("/app/vault/x.md"), content)
|
|
|
|
# B) MISSING OBJECT -> HTTP 400 + "Invalid or missing path" -> None
|
|
def test_b_missing_object_returns_none(self):
|
|
with mock.patch.object(self.client, "_post",
|
|
side_effect=self._post(400, json.dumps({"error": "Invalid or missing path"}))):
|
|
self.assertIsNone(self.client.read("/app/vault/c5f-controlled-canary.md"))
|
|
|
|
# C) OTHER HTTP 400 -> weiterhin fail-closed (Exception)
|
|
def test_c_other_400_fails_closed(self):
|
|
with mock.patch.object(self.client, "_post",
|
|
side_effect=self._post(400, json.dumps({"error": "Malformed request"}))):
|
|
with self.assertRaises(TolariaWriteError):
|
|
self.client.read("/app/vault/x.md")
|
|
|
|
# D) HTTP 401/403 -> weiterhin Auth-Fehler
|
|
def test_d_auth_fails_closed(self):
|
|
for status in (401, 403):
|
|
with mock.patch.object(self.client, "_post",
|
|
side_effect=self._post(status, json.dumps({"error": "unauthorized"}))):
|
|
with self.assertRaises(TolariaWriteError):
|
|
self.client.read("/app/vault/x.md")
|
|
|
|
# E) HTTP 500 -> weiterhin technischer Fehler (retrybar, Unavailable)
|
|
def test_e_500_is_unavailable(self):
|
|
def fake_500(endpoint, payload):
|
|
raise TolariaUnavailableError("Tolaria HTTP 500", "TOLARIA_UNAVAILABLE")
|
|
with mock.patch.object(self.client, "_post", side_effect=fake_500):
|
|
with self.assertRaises(TolariaUnavailableError):
|
|
self.client.read("/app/vault/x.md")
|
|
|
|
# F) CREATE pre_write_drift_check: before=None current=None -> WRITE_ALLOWED
|
|
def test_f_create_no_target_no_current_write_allowed(self):
|
|
# read() liefert None (missing) -> drift check => WRITE_ALLOWED
|
|
with mock.patch.object(self.client, "_post",
|
|
side_effect=self._post(400, json.dumps({"error": "Invalid or missing path"}))):
|
|
drift, cur = pre_write_drift_check(
|
|
self.client, "/app/vault/c5f-controlled-canary.md", None,
|
|
_fm(UUID_A, state="current") + "body")
|
|
self.assertEqual(drift, DRIFT_WRITE_ALLOWED)
|
|
self.assertIsNone(cur)
|
|
|
|
# G) existierendes unerwartetes Target -> weiterhin Drift/Human Gate
|
|
def test_g_unexpected_target_drift(self):
|
|
target = _fm(UUID_A, state="current") + "body v2"
|
|
# Tolaria CURRENT weder None (missing) noch BEFORE -> UNEXPECTED_DRIFT
|
|
with mock.patch.object(self.client, "_post",
|
|
side_effect=self._post(200, json.dumps({"content": _fm(UUID_A, state="current") + "DRIFT"}))):
|
|
drift, cur = pre_write_drift_check(
|
|
self.client, "/app/vault/x.md", _fm(UUID_A, state="current") + "body v1", target)
|
|
self.assertEqual(drift, DRIFT_UNEXPECTED)
|
|
|
|
|
|
class TestTolariaWriteNullResponseSemantic(unittest.TestCase):
|
|
"""C5C-Hotfix: /save HTTP 200 + JSON null -> erfolgreicher Transport;
|
|
Read-Back bleibt zwingend; 4xx/5xx weiterhin fail-closed."""
|
|
|
|
def setUp(self):
|
|
self.client = TolariaClient(base_url="http://fake")
|
|
|
|
def _post(self, status: int, body: str, reason_code=None):
|
|
"""Liefert einen Callable, der ein TolariaWriteError/JSON wirft/liefert."""
|
|
def fake_post(endpoint, payload):
|
|
if status == 200:
|
|
return json.loads(body)
|
|
raise TolariaWriteError(
|
|
f"Tolaria HTTP {status} auf {endpoint}: {body}",
|
|
reason_code or "INVALID_SCHEMA",
|
|
http_code=status,
|
|
)
|
|
return fake_post
|
|
|
|
def _content(self, body="body"):
|
|
return _fm(UUID_A, state="current") + body
|
|
|
|
# A) /save HTTP 200 + JSON null -> write() kein Crash, Save-Step erfolgreich
|
|
def test_a_save_200_null_no_crash(self):
|
|
def fake_post(endpoint, payload):
|
|
self.assertEqual(endpoint, "save")
|
|
return None # HTTP 200 + JSON null
|
|
with mock.patch.object(self.client, "_post", side_effect=fake_post):
|
|
resp = self.client.write("/app/vault/x.md", self._content())
|
|
# write() liefert ein Dict ({}), kein Crash
|
|
self.assertIsInstance(resp, dict)
|
|
|
|
# B) /save HTTP 200 + {} -> weiterhin erfolgreich
|
|
def test_b_save_200_empty_ok(self):
|
|
with mock.patch.object(self.client, "_post", side_effect=lambda ep, pl: {}):
|
|
resp = self.client.write("/app/vault/x.md", self._content())
|
|
self.assertIsInstance(resp, dict)
|
|
|
|
# C) /save HTTP 200 + normaler Success-Payload -> erfolgreich
|
|
def test_c_save_200_payload_ok(self):
|
|
with mock.patch.object(self.client, "_post", side_effect=lambda ep, pl: {"ok": True}):
|
|
resp = self.client.write("/app/vault/x.md", self._content())
|
|
self.assertEqual(resp, {"ok": True})
|
|
|
|
# D) /save 400 -> weiterhin FAIL CLOSED
|
|
def test_d_save_400_fail_closed(self):
|
|
with mock.patch.object(self.client, "_post", side_effect=self._post(400, json.dumps({"error": "Invalid or missing path"}))):
|
|
with self.assertRaises(TolariaWriteError):
|
|
self.client.write("/app/vault/x.md", self._content())
|
|
|
|
# E) /save 401/403 -> Auth-Failure (fail-closed)
|
|
def test_e_save_401_403_fail_closed(self):
|
|
for status in (401, 403):
|
|
with mock.patch.object(self.client, "_post", side_effect=self._post(status, json.dumps({"error": "unauthorized"}))):
|
|
with self.assertRaises(TolariaWriteError):
|
|
self.client.write("/app/vault/x.md", self._content())
|
|
|
|
# F) /save 500 -> technischer Fehler / Retry Contract (Unavailable)
|
|
def test_f_save_500_unavailable(self):
|
|
def fake_500(endpoint, payload):
|
|
raise TolariaUnavailableError("Tolaria HTTP 500", "TOLARIA_UNAVAILABLE")
|
|
with mock.patch.object(self.client, "_post", side_effect=fake_500):
|
|
with self.assertRaises(TolariaUnavailableError):
|
|
self.client.write("/app/vault/x.md", self._content())
|
|
|
|
# G) erfolgreicher Save + Read-Back Match -> C5C darf weiter
|
|
def test_g_save_then_readback_match(self):
|
|
content = self._content()
|
|
calls = {"n": 0}
|
|
|
|
def fake_post(endpoint, payload):
|
|
calls["n"] += 1
|
|
if endpoint == "save":
|
|
return None # HTTP 200 + JSON null
|
|
if endpoint == "content":
|
|
return {"content": content} # Read-Back liefert exakt dasselbe
|
|
raise AssertionError(f"unerwarteter endpoint {endpoint}")
|
|
|
|
with mock.patch.object(self.client, "_post", side_effect=fake_post):
|
|
self.client.write("/app/vault/x.md", content)
|
|
v = self.client.verify("/app/vault/x.md", content)
|
|
self.assertTrue(v["ok"])
|
|
self.assertEqual(calls["n"], 2) # save + read-back
|
|
|
|
# H) erfolgreicher Save + Read-Back Mismatch -> FAIL CLOSED / Drift -> KEIN Search-Step
|
|
def test_h_save_then_readback_mismatch_fail_closed(self):
|
|
content = self._content("TARGET")
|
|
drifted = self._content("DRIFTED")
|
|
|
|
def fake_post(endpoint, payload):
|
|
if endpoint == "save":
|
|
return None
|
|
if endpoint == "content":
|
|
return {"content": drifted}
|
|
raise AssertionError(f"unerwarteter endpoint {endpoint}")
|
|
|
|
with mock.patch.object(self.client, "_post", side_effect=fake_post):
|
|
self.client.write("/app/vault/x.md", content)
|
|
v = self.client.verify("/app/vault/x.md", content)
|
|
self.assertFalse(v["ok"]) # Mismatch -> fail-closed, kein falscher Erfolg
|
|
self.assertTrue(v["content_hash_match"] is False or v["field_mismatches"])
|
|
|
|
# I) erfolgreicher Save + Read-Back unavailable -> kein falscher Erfolg
|
|
def test_i_save_then_readback_unavailable(self):
|
|
content = self._content()
|
|
n = {"n": 0}
|
|
|
|
def fake_post(endpoint, payload):
|
|
n["n"] += 1
|
|
if endpoint == "save":
|
|
return None
|
|
raise TolariaUnavailableError("Tolaria down", "TOLARIA_UNAVAILABLE")
|
|
|
|
with mock.patch.object(self.client, "_post", side_effect=fake_post):
|
|
self.client.write("/app/vault/x.md", content)
|
|
with self.assertRaises(TolariaUnavailableError):
|
|
self.client.verify("/app/vault/x.md", content)
|
|
|
|
# J) bestehende read()-Not-Found-Regression bleibt PASS
|
|
def test_j_read_not_found_regression_kept(self):
|
|
# missing -> None (Hotfix 1 unverändert)
|
|
with mock.patch.object(self.client, "_post",
|
|
side_effect=self._post(400, json.dumps({"error": "Invalid or missing path"}))):
|
|
self.assertIsNone(self.client.read("/app/vault/c5f-controlled-canary.md"))
|
|
# anderer 400 -> fail-closed
|
|
with mock.patch.object(self.client, "_post",
|
|
side_effect=self._post(400, json.dumps({"error": "Malformed"}))):
|
|
with self.assertRaises(TolariaWriteError):
|
|
self.client.read("/app/vault/x.md")
|
|
|
|
# K) C5A-E-Regression wird separat über die volle Suite in w4 geprüft;
|
|
# hier zusätzlich: read() akzeptiert KEIN None-Swallowing (andere Endpoints unverändert)
|
|
def test_k_read_other_endpoint_none_still_fails_closed(self):
|
|
# read() mit None-Response (nur bei content-Endpoint + Not-Found ist None ok,
|
|
# aber ein pauschales None -> muss fail-closed bleiben). content-Endpoint liefert
|
|
# bei HTTP-200 normalerweise ein dict; ein None hier ist unerwartet -> read()
|
|
# würde sonst `if "error" in resp` crashen. read() darf None nur aus dem
|
|
# 400+Not-Found-Fall zurückgeben, NICHT pauschal aus einer None-Response.
|
|
def fake_null(endpoint, payload):
|
|
return None # unerwartete None-Response auf content
|
|
with mock.patch.object(self.client, "_post", side_effect=fake_null):
|
|
with self.assertRaises(TypeError):
|
|
self.client.read("/app/vault/x.md")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|