301 lines
11 KiB
Python
301 lines
11 KiB
Python
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()
|