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()