Red Queen
363f27327b
Persistiert die Search-Source-Provenance commit-spezifisch im C5-State (meta-KV-Key search_source_provenance:<workflow_commit_sha>), WRITE POINT nach validiertem Source-Build. Adoption prueft ausschliesslich gegen die persistierte source_provenance, nie gegen workflow_commit_sha oder current_repo_head. FAIL CLOSED ohne persistierte Provenance. Generischer, evidence-validierter Recovery-Pfad fuer extern abgeschlossene Builds. 14 neue Contract-Tests (A-N); volle Regression gruen.
484 lines
21 KiB
Python
484 lines
21 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Red Queen — C5F P5 Option A: External-Rebuild Adoption Contract Testsuite.
|
|
|
|
Testet den neuen C5D-Contract: C5D uebernimmt einen extern bereits
|
|
vollstaendig korrekt rebuildeten Suchzustand bei UPDATING_SEARCH idempotent,
|
|
OHNE einen zweiten Rebuild auszuloesen.
|
|
|
|
Kontrakt (siehe rq_c5d.evaluate_external_adoption):
|
|
UPDATING_SEARCH -> read-only expected/current Vergleich
|
|
-> ADOPT_ALREADY_AT_TARGET (nur bei exakter Set-Gleichheit)
|
|
-> VERIFYING_SEARCH -> final verify -> APPLIED
|
|
REBUILD_COUNT = 0.
|
|
|
|
KEIN object_count-only shortcut. Kein zweiter Rebuild. Kein manueller
|
|
State-Set. Kein Tolaria-/Forgejo-Write. last_applied erst nach vollem PASS.
|
|
|
|
Faelle A-M (Christian-Briefing §5):
|
|
A) UPDATING_SEARCH + Search exakt expected -> rebuild 0 -> adopt -> APPLIED
|
|
B) object_count gleich, aber Pfad fehlt -> keine Adoption (-> Rebuild-Pfad)
|
|
C) IDs stimmen nicht -> keine Adoption
|
|
D) source_head falsch -> keine Adoption
|
|
E) Canary/erwartetes neues Objekt fehlt -> keine Adoption
|
|
F) integrity_ok=false -> keine Adoption
|
|
G) failed_objects nicht leer -> keine Adoption
|
|
H) stale_objects nicht leer -> keine Adoption
|
|
I) Search leer/unavailable -> bestehender Retry/Failure-Contract
|
|
J) Search stale -> normaler Rebuild-Pfad bleibt funktionsfaehig
|
|
K) VERIFYING_SEARCH-Resume bleibt unveraendert (kein Rebuild)
|
|
L) Duplicate Replay nach APPLIED -> ALREADY_APPLIED, kein Rebuild
|
|
M) last_applied_commit nur nach vollstaendigem Adoption-PASS
|
|
N) kein Tolaria-Write waehrend Adoption
|
|
O) kein Master-/Forgejo-Write
|
|
"""
|
|
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_UPDATING_SEARCH, ST_VERIFYING_SEARCH, ST_APPLIED,
|
|
ST_HUMAN_REVIEW_REQUIRED, ST_RETRY_PENDING,
|
|
IDEM_ALREADY_APPLIED, IDEM_RETRY_SAFE,
|
|
OP_CREATE,
|
|
)
|
|
from rq_c5b import content_hash
|
|
from rq_c5d import (
|
|
SearchClient, C5DEngine, evaluate_external_adoption,
|
|
)
|
|
|
|
UUID_A = "object/77b02661-d67b-4bae-b612-01d1287cea6b"
|
|
UUID_B = "object/48bd264f-607b-15f1-5f73-3e922af9b19d"
|
|
UUID_CANARY = "object/6edb6869-0dfd-4046-993a-a727a8cab029"
|
|
SOURCE_HEAD = "12635e8"
|
|
|
|
# Erwartete (relative) Pfade — identisch zum echten SearchSourceBuilder:
|
|
# _source_objects erzeugt `path` als repo-relativen Pfad (VAULT_PREFIX gestrippt).
|
|
P_A = "modul-09.md"
|
|
P_B = "modul-10.md"
|
|
P_CANARY = "c5f-controlled-canary.md"
|
|
|
|
|
|
class FakeSourceBuilder:
|
|
"""Deterministisches erwartetes Objekt-Set via compute_expected()."""
|
|
|
|
def __init__(self, expected_ids, expected_paths, expected_count,
|
|
source_head=SOURCE_HEAD):
|
|
self.expected_ids = expected_ids
|
|
self.expected_paths = expected_paths
|
|
self.expected_count = expected_count
|
|
self.source_head = source_head
|
|
self.build_calls = 0
|
|
|
|
def compute_expected(self, commit_sha: str,
|
|
source_head: Optional[str] = None) -> Dict[str, Any]:
|
|
# source_head wird ggf. aus der persistierten Provenance uebergeben
|
|
# (C5D-Contract: Adoption prueft ausschliesslich gegen die persistierte
|
|
# Provenance, nicht gegen workflow_commit/current_repo_head).
|
|
head = self.source_head if source_head is None else source_head
|
|
return {
|
|
"commit_sha": commit_sha,
|
|
"source_head": head,
|
|
"expected_object_count": self.expected_count,
|
|
"expected_object_ids": sorted(self.expected_ids),
|
|
"expected_paths": sorted(self.expected_paths),
|
|
}
|
|
|
|
def build(self, commit_sha: str):
|
|
self.build_calls += 1
|
|
return {
|
|
"ok": True, "written": True, "source_path": "/tmp/source.json",
|
|
"expected": {
|
|
"source_head": self.source_head,
|
|
"expected_object_ids": sorted(self.expected_ids),
|
|
"expected_paths": sorted(self.expected_paths),
|
|
},
|
|
}
|
|
|
|
|
|
class FakeSearch:
|
|
"""Zaehlt rebuild()/health(); health ist konfigurierbar."""
|
|
|
|
def __init__(self, health: Optional[Dict[str, Any]] = None):
|
|
self.rebuild_count = 0
|
|
self.health_calls = 0
|
|
self._health = health or self._perfect_health()
|
|
self.rebuild_error: Optional[Exception] = None
|
|
|
|
def _perfect_health(self) -> Dict[str, Any]:
|
|
return {
|
|
"index_built": True,
|
|
"object_count": 3,
|
|
"failed_objects": [],
|
|
"stale_objects": [],
|
|
"secret_blocked_objects": 0,
|
|
"integrity_ok": True,
|
|
"indexed_paths": sorted([P_A, P_B, P_CANARY]),
|
|
"indexed_object_ids": sorted([UUID_A, UUID_B, UUID_CANARY]),
|
|
"source_head": SOURCE_HEAD,
|
|
}
|
|
|
|
def rebuild(self) -> Dict[str, Any]:
|
|
self.rebuild_count += 1
|
|
if self.rebuild_error:
|
|
raise self.rebuild_error
|
|
# Ein erfolgreicher Rebuild repariert den Index: health zuruecksetzen
|
|
# auf den vollstaendig korrekten Zustand (wie ein echter produktiver
|
|
# Rebuild, der aus der Source den korrekten Index neu aufbaut).
|
|
self._health = self._perfect_health()
|
|
return {"status": "ok", "indexed": self._health.get("object_count", 1)}
|
|
|
|
def get_health(self) -> Dict[str, Any]:
|
|
self.health_calls += 1
|
|
return self._health
|
|
|
|
|
|
class FakeSearchClient(SearchClient):
|
|
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.get_health()
|
|
|
|
|
|
def _make_store() -> C5AStore:
|
|
db = os.path.join(tempfile.mkdtemp(prefix="c5dadopt_db_"), "c5a.db")
|
|
return C5AStore(db)
|
|
|
|
|
|
def _default_objs():
|
|
return [
|
|
{"object_id": UUID_A, "operation": OP_CREATE,
|
|
"path_before": None, "path_after": P_A,
|
|
"content_hash_after": content_hash("body")},
|
|
{"object_id": UUID_B, "operation": OP_CREATE,
|
|
"path_before": None, "path_after": P_B,
|
|
"content_hash_after": content_hash("body")},
|
|
]
|
|
|
|
|
|
def _seed_commit(store, sha, objs=None, status=ST_UPDATING_SEARCH):
|
|
store.upsert_commit({
|
|
"commit_sha": sha, "parent_sha": "base0001", "sequence": 1,
|
|
"status": status, "retry_count": 0,
|
|
})
|
|
for oc in (objs or _default_objs()):
|
|
oc["commit_sha"] = sha
|
|
store.add_object_change(oc)
|
|
|
|
|
|
def _expected_default():
|
|
return ([UUID_A, UUID_B, UUID_CANARY], [P_A, P_B, P_CANARY], 3)
|
|
|
|
|
|
def _seed_provenance(store, sha, source_head=SOURCE_HEAD,
|
|
ids=None, paths=None):
|
|
"""Persistiert die erwartete Source-Provenance für einen Commit.
|
|
|
|
C5D-Contract: Adoption darf NUR gegen die persistierte Provenance prüfen
|
|
(FAIL CLOSED ohne persistierte Provenance). Um den Adoption-Pfad in den
|
|
Kontrakt-Tests wirklich zu testen, muss die Provenance vorab im Store liegen.
|
|
"""
|
|
ids = sorted(ids if ids is not None else _expected_default()[0])
|
|
paths = sorted(paths if paths is not None else _expected_default()[1])
|
|
store.persist_search_source_provenance(sha, source_head, ids, paths)
|
|
|
|
|
|
def _base_engine(store, fake, builder):
|
|
return C5DEngine(store, search=FakeSearchClient(fake), source_builder=builder)
|
|
|
|
|
|
def _perfect_engine(store):
|
|
ids, paths, cnt = _expected_default()
|
|
return _base_engine(store, FakeSearch(), FakeSourceBuilder(ids, paths, cnt))
|
|
|
|
|
|
class AdoptionContractTest(unittest.TestCase):
|
|
|
|
def _last_applied(self, store):
|
|
return store.health().get("last_applied_commit")
|
|
|
|
# ------------------------------------------------------------------ A)
|
|
def test_A_adopt_when_exact_match_no_rebuild(self):
|
|
store = _make_store()
|
|
_seed_commit(store, "c1")
|
|
_seed_provenance(store, "c1") # persistierte Provenance == Search source_head
|
|
fake = FakeSearch()
|
|
eng = _base_engine(store, fake, FakeSourceBuilder(*_expected_default()))
|
|
res = eng.apply_commit("c1")
|
|
self.assertEqual(res["status"], ST_APPLIED)
|
|
self.assertEqual(fake.rebuild_count, 0) # KEIN zweiter Rebuild
|
|
self.assertEqual(res["adoption"]["status"], "ADOPT_ALREADY_AT_TARGET")
|
|
self.assertEqual(res["adoption"]["rebuild_count"], 0)
|
|
self.assertEqual(self._last_applied(store), "c1")
|
|
self.assertEqual(store.commit_status("c1"), ST_APPLIED)
|
|
|
|
# ------------------------------------------------------------------ B)
|
|
def test_B_object_count_ok_but_path_missing_no_adopt(self):
|
|
store = _make_store()
|
|
_seed_commit(store, "c1")
|
|
_seed_provenance(store, "c1")
|
|
ids, paths, cnt = _expected_default()
|
|
# object_count == 3, aber ein Pfad fehlt -> paths_exact False
|
|
health = {
|
|
"index_built": True, "integrity_ok": True, "object_count": 3,
|
|
"failed_objects": [], "stale_objects": [], "secret_blocked_objects": 0,
|
|
"indexed_paths": [P_A, P_B], # P_CANARY fehlt
|
|
"indexed_object_ids": sorted(ids), "source_head": SOURCE_HEAD,
|
|
}
|
|
fake = FakeSearch(health)
|
|
eng = _base_engine(store, fake, FakeSourceBuilder(ids, paths, cnt))
|
|
res = eng.apply_commit("c1")
|
|
self.assertEqual(fake.rebuild_count, 1) # Rebuild-Pfad
|
|
self.assertIsNone(res.get("adoption"))
|
|
self.assertEqual(res["status"], ST_APPLIED)
|
|
|
|
# ------------------------------------------------------------------ C)
|
|
def test_C_ids_mismatch_no_adopt(self):
|
|
store = _make_store()
|
|
_seed_commit(store, "c1")
|
|
_seed_provenance(store, "c1")
|
|
ids, paths, cnt = _expected_default()
|
|
health = {
|
|
"index_built": True, "integrity_ok": True, "object_count": 3,
|
|
"failed_objects": [], "stale_objects": [], "secret_blocked_objects": 0,
|
|
"indexed_paths": [P_A, P_B, P_CANARY],
|
|
"indexed_object_ids": [UUID_A, UUID_B, "object/999"],
|
|
"source_head": SOURCE_HEAD,
|
|
}
|
|
fake = FakeSearch(health)
|
|
eng = _base_engine(store, fake, FakeSourceBuilder(ids, paths, cnt))
|
|
res = eng.apply_commit("c1")
|
|
self.assertEqual(fake.rebuild_count, 1)
|
|
self.assertIsNone(res.get("adoption"))
|
|
|
|
# ------------------------------------------------------------------ D)
|
|
def test_D_source_head_wrong_no_adopt(self):
|
|
store = _make_store()
|
|
_seed_commit(store, "c1")
|
|
_seed_provenance(store, "c1")
|
|
ids, paths, cnt = _expected_default()
|
|
health = {
|
|
"index_built": True, "integrity_ok": True, "object_count": 3,
|
|
"failed_objects": [], "stale_objects": [], "secret_blocked_objects": 0,
|
|
"indexed_paths": [P_A, P_B, P_CANARY],
|
|
"indexed_object_ids": sorted(ids), "source_head": "WRONGHEAD",
|
|
}
|
|
fake = FakeSearch(health)
|
|
eng = _base_engine(store, fake, FakeSourceBuilder(ids, paths, cnt))
|
|
res = eng.apply_commit("c1")
|
|
self.assertEqual(fake.rebuild_count, 1)
|
|
self.assertIsNone(res.get("adoption"))
|
|
|
|
# ------------------------------------------------------------------ E
|
|
def test_canary_missing_no_adopt(self):
|
|
store = _make_store()
|
|
_seed_commit(store, "c1")
|
|
_seed_provenance(store, "c1")
|
|
ids, paths, cnt = _expected_default()
|
|
# Canary fehlt -> object_count 2, ids/paths ohne Canary
|
|
health = {
|
|
"index_built": True, "integrity_ok": True, "object_count": 2,
|
|
"failed_objects": [], "stale_objects": [], "secret_blocked_objects": 0,
|
|
"indexed_paths": [P_A, P_B],
|
|
"indexed_object_ids": [UUID_A, UUID_B], "source_head": SOURCE_HEAD,
|
|
}
|
|
fake = FakeSearch(health)
|
|
eng = _base_engine(store, fake, FakeSourceBuilder(ids, paths, cnt))
|
|
res = eng.apply_commit("c1")
|
|
self.assertEqual(fake.rebuild_count, 1)
|
|
self.assertIsNone(res.get("adoption"))
|
|
|
|
# ------------------------------------------------------------------ F
|
|
def test_integrity_ok_false_no_adopt(self):
|
|
store = _make_store()
|
|
_seed_commit(store, "c1")
|
|
_seed_provenance(store, "c1")
|
|
ids, paths, cnt = _expected_default()
|
|
health = {
|
|
"index_built": True, "integrity_ok": False, "object_count": 3,
|
|
"failed_objects": [], "stale_objects": [], "secret_blocked_objects": 0,
|
|
"indexed_paths": [P_A, P_B, P_CANARY],
|
|
"indexed_object_ids": sorted(ids), "source_head": SOURCE_HEAD,
|
|
}
|
|
fake = FakeSearch(health)
|
|
eng = _base_engine(store, fake, FakeSourceBuilder(ids, paths, cnt))
|
|
res = eng.apply_commit("c1")
|
|
self.assertEqual(fake.rebuild_count, 1)
|
|
self.assertIsNone(res.get("adoption"))
|
|
|
|
# ------------------------------------------------------------------ G
|
|
def test_failed_objects_not_empty_no_adopt(self):
|
|
store = _make_store()
|
|
_seed_commit(store, "c1")
|
|
_seed_provenance(store, "c1")
|
|
ids, paths, cnt = _expected_default()
|
|
health = {
|
|
"index_built": True, "integrity_ok": True, "object_count": 3,
|
|
"failed_objects": ["x"], "stale_objects": [],
|
|
"secret_blocked_objects": 0,
|
|
"indexed_paths": [P_A, P_B, P_CANARY],
|
|
"indexed_object_ids": sorted(ids), "source_head": SOURCE_HEAD,
|
|
}
|
|
fake = FakeSearch(health)
|
|
eng = _base_engine(store, fake, FakeSourceBuilder(ids, paths, cnt))
|
|
res = eng.apply_commit("c1")
|
|
self.assertEqual(fake.rebuild_count, 1)
|
|
self.assertIsNone(res.get("adoption"))
|
|
|
|
# ------------------------------------------------------------------ H
|
|
def test_stale_objects_not_empty_no_adopt(self):
|
|
store = _make_store()
|
|
_seed_commit(store, "c1")
|
|
_seed_provenance(store, "c1")
|
|
ids, paths, cnt = _expected_default()
|
|
health = {
|
|
"index_built": True, "integrity_ok": True, "object_count": 3,
|
|
"failed_objects": [], "stale_objects": ["x"],
|
|
"secret_blocked_objects": 0,
|
|
"indexed_paths": [P_A, P_B, P_CANARY],
|
|
"indexed_object_ids": sorted(ids), "source_head": SOURCE_HEAD,
|
|
}
|
|
fake = FakeSearch(health)
|
|
eng = _base_engine(store, fake, FakeSourceBuilder(ids, paths, cnt))
|
|
res = eng.apply_commit("c1")
|
|
self.assertEqual(fake.rebuild_count, 1)
|
|
self.assertIsNone(res.get("adoption"))
|
|
|
|
# ------------------------------------------------------------------ I
|
|
def test_I_unavailable_retry_contract(self):
|
|
# Search unerreichbar -> RETRY_PENDING, kein Adoption, kein APPLIED.
|
|
from rq_c5d import SearchUnavailableError, RC_SEARCH_REBUILD_FAILURE
|
|
store = _make_store()
|
|
_seed_commit(store, "c1")
|
|
_seed_provenance(store, "c1") # Adoption-Pfad wird betreten; health() wirft
|
|
fake = FakeSearch()
|
|
# health() wirft beim Adoption-Check und beim Rebuild -> Retry-Contract
|
|
def boom():
|
|
raise SearchUnavailableError("search down", RC_SEARCH_REBUILD_FAILURE)
|
|
fake.get_health = boom
|
|
fake.rebuild_error = SearchUnavailableError(
|
|
"search down", RC_SEARCH_REBUILD_FAILURE)
|
|
eng = _base_engine(store, fake, FakeSourceBuilder(*_expected_default()))
|
|
res = eng.apply_commit("c1")
|
|
# Adoption-Pruefung wirft -> adoption False (nicht blockierend)
|
|
# -> danach rebuild() wirft ebenfalls -> Retry-Pfad (SearchUnavailable)
|
|
self.assertEqual(res["status"], ST_RETRY_PENDING)
|
|
self.assertIsNone(self._last_applied(store))
|
|
|
|
# ------------------------------------------------------------------ J)
|
|
def test_J_search_stale_normal_rebuild_path(self):
|
|
# Index 'stale' (z.B. alte paths) -> keine Adoption -> normaler Rebuild.
|
|
store = _make_store()
|
|
_seed_commit(store, "c1")
|
|
_seed_provenance(store, "c1")
|
|
ids, paths, cnt = _expected_default()
|
|
health = {
|
|
"index_built": True, "integrity_ok": True, "object_count": 3,
|
|
"failed_objects": [], "stale_objects": [], "secret_blocked_objects": 0,
|
|
"indexed_paths": [P_A, "alt.md"], # stale Pfad -> paths mismatch
|
|
"indexed_object_ids": sorted(ids), "source_head": SOURCE_HEAD,
|
|
}
|
|
fake = FakeSearch(health)
|
|
eng = _base_engine(store, fake, FakeSourceBuilder(ids, paths, cnt))
|
|
res = eng.apply_commit("c1")
|
|
self.assertEqual(fake.rebuild_count, 1) # Rebuild-Pfad funktioniert
|
|
self.assertIsNone(res.get("adoption"))
|
|
self.assertEqual(res["status"], ST_APPLIED)
|
|
|
|
# ------------------------------------------------------------------ K)
|
|
def test_K_verifying_search_resume_no_rebuild(self):
|
|
store = _make_store()
|
|
_seed_commit(store, "c1", status=ST_VERIFYING_SEARCH)
|
|
fake = FakeSearch()
|
|
eng = _base_engine(store, fake, FakeSourceBuilder(*_expected_default()))
|
|
res = eng.apply_commit("c1")
|
|
self.assertEqual(fake.rebuild_count, 0) # Resume -> kein Rebuild
|
|
self.assertEqual(res["status"], ST_APPLIED)
|
|
self.assertIsNone(res.get("adoption")) # kein Adoption-Report
|
|
|
|
# ------------------------------------------------------------------ L)
|
|
def test_L_duplicate_after_applied_no_rebuild(self):
|
|
store = _make_store()
|
|
# Commit bereits vollstaendig APPLIED (last_applied gesetzt) -> Replay
|
|
_seed_commit(store, "c1", status=ST_APPLIED)
|
|
store.mark_applied("c1")
|
|
fake = FakeSearch()
|
|
eng = _base_engine(store, fake, FakeSourceBuilder(*_expected_default()))
|
|
res = eng.apply_commit("c1")
|
|
self.assertEqual(res["status"], ST_APPLIED)
|
|
self.assertEqual(res["idempotency"], IDEM_ALREADY_APPLIED)
|
|
self.assertEqual(fake.rebuild_count, 0)
|
|
|
|
# ------------------------------------------------------------------ M)
|
|
def test_M_last_applied_only_after_full_pass(self):
|
|
# Bei fehlgeschlagener Adoption (Pfad fehlt) und NICHT erfolgreichem
|
|
# Rebuild -> kein APPLIED -> last_applied None.
|
|
store = _make_store()
|
|
_seed_commit(store, "c1")
|
|
_seed_provenance(store, "c1")
|
|
ids, paths, cnt = _expected_default()
|
|
health = {
|
|
"index_built": True, "integrity_ok": True, "object_count": 3,
|
|
"failed_objects": [], "stale_objects": [], "secret_blocked_objects": 0,
|
|
"indexed_paths": [P_A, P_B], # fehlt -> keine Adoption
|
|
"indexed_object_ids": [UUID_A, UUID_B], "source_head": SOURCE_HEAD,
|
|
}
|
|
fake = FakeSearch(health)
|
|
# Rebuild liefert Fehler -> Human Gate (nicht APPLIED)
|
|
from rq_c5d import SearchRebuildError, RC_SEARCH_REBUILD_FAILURE
|
|
fake.rebuild_error = SearchRebuildError("fail", RC_SEARCH_REBUILD_FAILURE)
|
|
eng = _base_engine(store, fake, FakeSourceBuilder(ids, paths, cnt))
|
|
res = eng.apply_commit("c1")
|
|
self.assertEqual(res["status"], ST_HUMAN_REVIEW_REQUIRED)
|
|
self.assertIsNone(self._last_applied(store))
|
|
|
|
# ------------------------------------------------------------------ N) ENGINE
|
|
def test_N_no_tolaria_write_during_adoption(self):
|
|
store = _make_store()
|
|
_seed_commit(store, "c1")
|
|
_seed_provenance(store, "c1")
|
|
fake = FakeSearch()
|
|
builder = FakeSourceBuilder(*_expected_default())
|
|
eng = _base_engine(store, fake, builder)
|
|
res = eng.apply_commit("c1")
|
|
self.assertEqual(res["status"], ST_APPLIED)
|
|
self.assertEqual(fake.rebuild_count, 0)
|
|
self.assertEqual(builder.build_calls, 0) # kein Source-Build-Write
|
|
# Adoption ist der KEIN-Write-Pfad: adoption gesetzt, rebuild_count 0
|
|
self.assertEqual(res["adoption"]["status"], "ADOPT_ALREADY_AT_TARGET")
|
|
self.assertEqual(res["adoption"]["rebuild_count"], 0)
|
|
|
|
# ------------------------------------------------------------------ O) ENGINE
|
|
def test_evaluate_external_adoption_pure(self):
|
|
# Reine Funktion: alle Checks, nur exakt expected -> ok True
|
|
ids, paths, cnt = _expected_default()
|
|
exp = {"expected_object_ids": sorted(ids), "expected_paths": sorted(paths),
|
|
"expected_object_count": cnt, "source_head": SOURCE_HEAD}
|
|
h = {
|
|
"index_built": True, "integrity_ok": True, "object_count": 3,
|
|
"failed_objects": [], "stale_objects": [], "secret_blocked_objects": 0,
|
|
"indexed_paths": sorted(paths),
|
|
"indexed_object_ids": sorted(ids), "source_head": SOURCE_HEAD,
|
|
}
|
|
r = evaluate_external_adoption(exp, h)
|
|
self.assertTrue(r["ok"])
|
|
|
|
# object_count gleich aber path fehlt -> ok False (kein count-only shortcut)
|
|
h2 = dict(h, indexed_paths=[P_A, P_B])
|
|
r2 = evaluate_external_adoption(exp, h2)
|
|
self.assertFalse(r2["ok"])
|
|
self.assertIn("paths_exact", r2["checks"])
|
|
self.assertFalse(r2["checks"]["paths_exact"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|