From 2b98059b44f4d46b4d4cce776f163e893c647c76 Mon Sep 17 00:00:00 2001 From: Andraxion Date: Wed, 29 Jul 2026 16:41:42 -0400 Subject: [PATCH] Prove derived artifact recovery against exact oracles --- Makefile | 1 + tests/test_milestone5_recovery.py | 301 ++++++++++++++++++++++++++++++ 2 files changed, 302 insertions(+) create mode 100644 tests/test_milestone5_recovery.py diff --git a/Makefile b/Makefile index 351347d..9719ddd 100644 --- a/Makefile +++ b/Makefile @@ -119,6 +119,7 @@ concurrency-m5: recovery-m5: PYTHONPYCACHEPREFIX=$(PYTHONPYCACHEPREFIX) $(PYTHON) -m pytest -q \ -p no:cacheprovider --basetemp=$(PYTEST_BASETEMP)-recovery \ + tests/test_milestone5_recovery.py \ tests/test_core.py::DocForgeCoreTests::test_missing_or_corrupt_generation_falls_back_and_repairs \ tests/test_core.py::DocForgeCoreTests::test_stale_source_fails_closed_and_failed_rebuild_preserves_index \ tests/test_incremental_cache.py \ diff --git a/tests/test_milestone5_recovery.py b/tests/test_milestone5_recovery.py new file mode 100644 index 0000000..ebd1773 --- /dev/null +++ b/tests/test_milestone5_recovery.py @@ -0,0 +1,301 @@ +from __future__ import annotations + +import hashlib +import json +import shutil +import tempfile +import unittest +from dataclasses import dataclass +from pathlib import Path +from typing import cast + +from docforge.generation_diff import generation_diff_path +from docforge.graph_rendering import GraphRenderService +from docforge.index import ProjectIndex +from docforge.pagination import canonical_hash +from docforge.project import Project +from docforge.rendering import RenderService + +ROOT = Path(__file__).resolve().parents[1] +FIXTURES = ROOT / "tests" / "fixtures" + +GRAPH_CONFIG = """ + +[graph_render] +output_root = ".docforge/portable-graph" + +[[graph_render.views]] +id = "architecture" +renderer = "portable_graph_html" +output = "architecture.html" +title = "Alpha architecture" +root = "guide.workflow" +initial_mode = "nodes" +depth = 2 +max_nodes = 20 +max_edges = 40 +max_work = 1000 +families = ["guide", "proof"] +relations = ["depends_on", "proves"] +authorities = [] +statuses = [] +tags = [] +include_logic = false +""" + +INDEX_IDENTITY_FIELDS = ( + "project_id", + "project_root_fingerprint", + "revision", + "source_hash", + "node_hash", + "node_count", + "edge_hash", + "edge_count", + "logic_hash", + "logic_projection_count", + "logic_node_count", + "logic_edge_count", + "index_schema_version", + "adapter", + "status", +) + +RENDER_SEMANTIC_FIELDS = ( + "schema_version", + "project_id", + "project_root_fingerprint", + "adapter", + "revision", + "source_hash", + "view_id", + "view_config_hash", + "renderer", + "renderer_version", + "render_identity", + "template_hash", + "output_hash", + "output_bytes", +) + + +@dataclass(frozen=True) +class RecoveryOracle: + canonical_bytes: dict[str, bytes] + canonical_collection_hash: str + full_snapshot_hash: str + index_identity: dict[str, object] + + +class Milestone5RecoveryTests(unittest.TestCase): + def setUp(self) -> None: + self.temporary = tempfile.TemporaryDirectory() + self.addCleanup(self.temporary.cleanup) + self.root = Path(self.temporary.name) / "alpha" + shutil.copytree(FIXTURES / "alpha", self.root) + shutil.rmtree(self.root / ".docforge/cache", ignore_errors=True) + shutil.rmtree(self.root / ".docforge/rendered", ignore_errors=True) + shutil.rmtree(self.root / ".docforge/portable-graph", ignore_errors=True) + descriptor = self.root / ".docforge/project.toml" + descriptor.write_text( + descriptor.read_text(encoding="utf-8") + GRAPH_CONFIG, + encoding="utf-8", + ) + + project = Project.open(self.root) + ProjectIndex(project).build() + RenderService(project).render("manual") + GraphRenderService(project).render("architecture") + self.oracle = self._capture_oracle() + + @staticmethod + def _sha256(raw: bytes) -> str: + return hashlib.sha256(raw).hexdigest() + + def _canonical_bytes(self, project: Project) -> dict[str, bytes]: + descriptor = project.descriptor + paths = ( + descriptor.descriptor_path, + *descriptor.authority_files, + *project.canonical_source_paths(), + descriptor.render.template_root / "manual.html", # type: ignore[union-attr] + ) + return { + path.relative_to(self.root).as_posix(): path.read_bytes() + for path in sorted(set(paths), key=lambda item: item.relative_to(self.root).as_posix()) + } + + def _capture_oracle(self) -> RecoveryOracle: + project = Project.open(self.root) + canonical_bytes = self._canonical_bytes(project) + snapshot = project.load() + snapshot_document = { + "revision": snapshot.revision, + "source_hash": snapshot.source_hash, + "nodes": [node.as_dict() for node in snapshot.nodes], + "edges": [edge.as_dict() for edge in snapshot.edges], + } + checked = ProjectIndex(project).check() + index_identity = {field: checked[field] for field in INDEX_IDENTITY_FIELDS} + canonical_collection_hash = canonical_hash( + { + path: { + "bytes": len(raw), + "sha256": self._sha256(raw), + } + for path, raw in canonical_bytes.items() + } + ) + return RecoveryOracle( + canonical_bytes=canonical_bytes, + canonical_collection_hash=canonical_collection_hash, + full_snapshot_hash=canonical_hash(snapshot_document), + index_identity=index_identity, + ) + + def _assert_oracle_preserved(self) -> None: + repaired = self._capture_oracle() + self.assertEqual(self.oracle.canonical_bytes, repaired.canonical_bytes) + self.assertEqual( + self.oracle.canonical_collection_hash, + repaired.canonical_collection_hash, + ) + self.assertEqual(self.oracle.full_snapshot_hash, repaired.full_snapshot_hash) + self.assertEqual(self.oracle.index_identity, repaired.index_identity) + + @staticmethod + def _projection_semantics(value: object) -> dict[str, object]: + receipt = cast(dict[str, object], value) + return { + key: item + for key, item in receipt.items() + if key not in {"peak_memory_bytes", "receipt_id", "timing"} + } + + @classmethod + def _render_semantics(cls, receipt: dict[str, object]) -> dict[str, object]: + return { + **{field: receipt[field] for field in RENDER_SEMANTIC_FIELDS}, + "projection_receipt": cls._projection_semantics(receipt["projection_receipt"]), + } + + @classmethod + def _graph_manifest_semantics(cls, manifest: dict[str, object]) -> dict[str, object]: + return { + key: (cls._projection_semantics(value) if key == "receipt" else value) + for key, value in manifest.items() + if key not in {"publication_id", "store", "output"} + } + + def test_corrupt_index_attestation_is_rebuilt_from_the_exact_index_oracle(self) -> None: + project = Project.open(self.root) + index = ProjectIndex(project) + attestation = index.attestation_path + exact_attestation = attestation.read_bytes() + + attestation.write_bytes(b"{corrupt-attestation") + self.assertFalse(index._attestation_matches()) + + synchronized = ProjectIndex(Project.open(self.root)).synchronize() + + self.assertEqual("current", synchronized["synchronization"]["action"]) + self.assertEqual(exact_attestation, attestation.read_bytes()) + self.assertTrue(ProjectIndex(Project.open(self.root))._attestation_matches()) + self._assert_oracle_preserved() + + def test_corrupt_render_receipt_is_rerendered_to_the_exact_semantic_oracle(self) -> None: + receipt_path = self.root / ".docforge/cache/render-receipts/manual.json" + output_path = self.root / ".docforge/rendered/manual.html" + exact_output = output_path.read_bytes() + oracle_receipt = cast( + dict[str, object], + json.loads(receipt_path.read_text(encoding="utf-8")), + ) + + receipt_path.write_bytes(b"{corrupt-render-receipt") + service = RenderService(Project.open(self.root)) + broken = service.status("manual") + self.assertEqual("unverified", broken["outputs"][0]["state"]) + self.assertEqual("receipt_corrupt", broken["outputs"][0]["reason"]) + + repaired = service.render("manual") + repaired_receipt = cast( + dict[str, object], + json.loads(receipt_path.read_text(encoding="utf-8")), + ) + + self.assertEqual("current", repaired["state"]) + self.assertEqual(exact_output, output_path.read_bytes()) + self.assertEqual( + self._render_semantics(oracle_receipt), + self._render_semantics(repaired_receipt), + ) + self.assertEqual("current", service.status("manual")["state"]) + self.assertEqual("current", service.deep_status("manual")["state"]) + self._assert_oracle_preserved() + + def test_corrupt_generation_diff_is_rebuilt_to_the_exact_current_graph_oracle(self) -> None: + project = Project.open(self.root) + receipt_path = generation_diff_path(project.descriptor) + receipt_path.write_bytes(b"{corrupt-generation-diff") + broken = ProjectIndex(project).generation_diff() + self.assertEqual("unverified", broken["receipt_state"]) + self.assertEqual("corrupt_receipt", broken["receipt_reason"]) + + ProjectIndex(Project.open(self.root)).build() + index = ProjectIndex(Project.open(self.root)) + repaired = index.generation_diff() + receipt = cast(dict[str, object], repaired["generation_diff"]) + + self.assertEqual("current", repaired["receipt_state"]) + self.assertEqual("current", repaired["staleness"]) + self.assertEqual("baseline", receipt["kind"]) + self.assertEqual("no_meaningful_transition", receipt["reason"]) + self.assertEqual( + { + field: self.oracle.index_identity[field] + for field in ( + "revision", + "source_hash", + "node_count", + "node_hash", + "edge_count", + "edge_hash", + "index_schema_version", + ) + }, + receipt["to_generation"], + ) + self._assert_oracle_preserved() + + def test_corrupt_portable_graph_manifest_is_rerendered_to_exact_bytes(self) -> None: + manifest_path = ( + self.root / ".docforge/cache/projection-publications/graph/architecture.json" + ) + output_path = self.root / ".docforge/portable-graph/architecture.html" + oracle_manifest = cast( + dict[str, object], + json.loads(manifest_path.read_text(encoding="utf-8")), + ) + exact_output = output_path.read_bytes() + + manifest_path.write_bytes(b"{corrupt-portable-graph-manifest") + service = GraphRenderService(Project.open(self.root)) + broken = service.status("architecture") + self.assertEqual("missing", broken["outputs"][0]["state"]) + self.assertEqual("manifest_missing", broken["outputs"][0]["reason"]) + + repaired = service.render("architecture") + repaired_manifest = cast( + dict[str, object], + json.loads(manifest_path.read_text(encoding="utf-8")), + ) + + self.assertEqual("current", repaired["state"]) + self.assertEqual(exact_output, output_path.read_bytes()) + self.assertEqual( + self._graph_manifest_semantics(oracle_manifest), + self._graph_manifest_semantics(repaired_manifest), + ) + self.assertEqual("current", service.status("architecture")["state"]) + self._assert_oracle_preserved()