from __future__ import annotations import copy import json import unittest from pathlib import Path from jsonschema import Draft202012Validator from docforge.errors import DocForgeError from docforge.graph_projection import build_graph_projection_package from docforge.projection_contract import ( GraphViewPlanV1, ManualRenderPlanV1, ProjectionPackageV1, ProjectionReceiptV1, ) from docforge_renderers.graph import PortableGraphHtmlRenderer ROOT = Path(__file__).resolve().parents[1] SCHEMAS = ROOT / "schemas" class ProjectionSchemaTests(unittest.TestCase): @staticmethod def schema(name: str) -> dict[str, object]: return json.loads((SCHEMAS / name).read_text(encoding="utf-8")) @classmethod def validator(cls, name: str) -> Draft202012Validator: return Draft202012Validator(cls.schema(name)) @staticmethod def project_identity() -> dict[str, object]: return { "project_id": "schema-fixture", "project_root_fingerprint": "a" * 16, "adapter": "generic", "revision": "fixture-revision", "source_hash": "b" * 64, } @classmethod def manual_plan(cls) -> ManualRenderPlanV1: return ManualRenderPlanV1.create( { "project": cls.project_identity(), "view": { "view_id": "manual", "title": "Schema Manual", "families": ["guide"], "renderer": "generic_html", }, "changeset_hash": None, "pages": [ { "node_id": "guide.schema", "title": "Projection schema", "family": "guide", "authority": "authoritative", "status": "approved", "tags": ["schema"], "summary": "Defines the projection schema fixture.", "content": "The schema fixture is deterministic.", "content_hash": "c" * 64, "components": ["manual.commonmark@1"], "breadcrumbs": [], "cross_references": [], "backlinks": [], } ], "navigation": [ { "node_id": "guide.schema", "title": "Projection schema", } ], "search_documents": [ { "node_id": "guide.schema", "title": "Projection schema", "summary": "Defines the projection schema fixture.", "family": "guide", "status": "approved", "tags": ["schema"], } ], "diagnostics": {"orphans": ["guide.schema"], "cycles": []}, } ) @classmethod def graph_plan(cls) -> GraphViewPlanV1: return GraphViewPlanV1.create( { "project": cls.project_identity(), "view": { "view_id": "portable", "title": "Portable graph", "initial_mode": "nodes", "scope": { "kind": "exact_root", "root_node_id": "guide.schema", "depth": 2, }, "filters": { "families": [], "relations": [], "authorities": [], "statuses": [], "tags": [], }, "detail_fields": [ "node_id", "title", "family", "authority", "status", "tags", "summary", "content_hash", ], }, "bounds": { "depth": 2, "max_nodes": 100, "max_edges": 400, "max_work": 100000, }, "policy": { "visibility": "selected_graph_only", "source_paths": "excluded", "source_bodies": "excluded", "database_queries": "forbidden", "executable_content": "forbidden", "logic": "forbidden", "logic_requested": False, }, "graph": { "root_node_id": "guide.schema", "nodes": [], "edges": [], "logic_projections": [], }, "omissions": [], "diagnostics": { "selection": "exact_root", "returned_nodes": 0, "returned_edges": 0, "examined_work_units": 0, "truncated": False, "ordering": "node_id;source_id,relation,target_id", }, } ) @classmethod def package( cls, plan: ManualRenderPlanV1 | GraphViewPlanV1 | None = None, ) -> ProjectionPackageV1: selected = plan or cls.manual_plan() kind = "manual" if isinstance(selected, ManualRenderPlanV1) else "graph" return ProjectionPackageV1.create( kind=kind, plan=selected, renderer={ "renderer_id": "generic_html", "renderer_version": "1", }, components=[{"component_id": "projection.document@1"}], assets=[ { "asset_id": "projection.template", "media_type": "text/plain; charset=utf-8", "sha256": "d" * 64, "text": "fixture", } ], output_policy={ "artifact_ids": ["projection.html"], "max_total_bytes": 1000000, }, ) @classmethod def receipt(cls) -> ProjectionReceiptV1: package = cls.package() return ProjectionReceiptV1.create( kind="manual", package_id=package.package_id, plan_id=package.document["plan_id"], # type: ignore[arg-type] renderer={ "renderer_id": "generic_html", "renderer_version": "1", }, artifacts=[ { "artifact_id": "manual.html", "media_type": "text/html; charset=utf-8", "sha256": "e" * 64, "bytes": 123, } ], diagnostics={"warnings": []}, timing={"elapsed_ns": 123456}, peak_memory_bytes=None, ) def test_schemas_are_valid_and_accept_current_documents(self) -> None: documents = { "manual-render-plan.schema.json": self.manual_plan().as_dict(), "graph-view-plan.schema.json": self.graph_plan().as_dict(), "projection-package.schema.json": self.package().as_dict(), "projection-receipt.schema.json": self.receipt().as_dict(), } for name, document in documents.items(): with self.subTest(schema=name): schema = self.schema(name) Draft202012Validator.check_schema(schema) Draft202012Validator(schema).validate(document) graph_package = self.package(self.graph_plan()).as_dict() self.validator("projection-package.schema.json").validate(graph_package) typed_graph_package = build_graph_projection_package( self.graph_plan(), renderer_id=PortableGraphHtmlRenderer.renderer_id, renderer_version=PortableGraphHtmlRenderer.renderer_version, max_output_bytes=1_000_000, ) graph_receipt = PortableGraphHtmlRenderer().render(typed_graph_package).receipt.as_dict() self.validator("projection-receipt.schema.json").validate(graph_receipt) def test_unknown_fields_are_rejected_at_contract_boundaries(self) -> None: cases = ( ( "manual-render-plan.schema.json", self.manual_plan().as_dict(), ), ( "graph-view-plan.schema.json", self.graph_plan().as_dict(), ), ( "projection-package.schema.json", self.package().as_dict(), ), ( "projection-receipt.schema.json", self.receipt().as_dict(), ), ) for name, document in cases: with self.subTest(schema=name): document["unexpected"] = True self.assertFalse(self.validator(name).is_valid(document)) manual = self.manual_plan().as_dict() manual["pages"][0]["unexpected"] = True # type: ignore[index] self.assertFalse(self.validator("manual-render-plan.schema.json").is_valid(manual)) package = self.package().as_dict() package["assets"][0]["path"] = "/tmp/escape" # type: ignore[index] self.assertFalse(self.validator("projection-package.schema.json").is_valid(package)) receipt = self.receipt().as_dict() receipt["artifacts"][0]["content"] = "not receipt evidence" # type: ignore[index] self.assertFalse(self.validator("projection-receipt.schema.json").is_valid(receipt)) def test_obviously_malformed_identities_and_structures_are_rejected(self) -> None: manual = self.manual_plan().as_dict() manual["plan_id"] = "not-a-sha256" self.assertFalse(self.validator("manual-render-plan.schema.json").is_valid(manual)) graph = self.graph_plan().as_dict() graph["project"]["project_root_fingerprint"] = "wrong" # type: ignore[index] self.assertFalse(self.validator("graph-view-plan.schema.json").is_valid(graph)) graph = self.graph_plan().as_dict() graph["bounds"] = -1 self.assertFalse(self.validator("graph-view-plan.schema.json").is_valid(graph)) package = self.package().as_dict() package["assets"][0].pop("sha256") # type: ignore[index] self.assertFalse(self.validator("projection-package.schema.json").is_valid(package)) package = self.package().as_dict() package["output_policy"]["max_total_bytes"] = 0 # type: ignore[index] self.assertFalse(self.validator("projection-package.schema.json").is_valid(package)) package = self.package().as_dict() package["kind"] = "graph" self.assertFalse(self.validator("projection-package.schema.json").is_valid(package)) receipt = self.receipt().as_dict() receipt["artifacts"][0]["artifact_id"] = "../manual.html" # type: ignore[index] self.assertFalse(self.validator("projection-receipt.schema.json").is_valid(receipt)) receipt = self.receipt().as_dict() receipt["artifacts"][0]["bytes"] = -1 # type: ignore[index] self.assertFalse(self.validator("projection-receipt.schema.json").is_valid(receipt)) receipt = self.receipt().as_dict() receipt["peak_memory_bytes"] = True self.assertFalse(self.validator("projection-receipt.schema.json").is_valid(receipt)) def test_assets_and_artifacts_enforce_fixed_collection_bounds(self) -> None: package = self.package().as_dict() package["assets"] = [copy.deepcopy(package["assets"][0]) for _ in range(33)] # type: ignore[index] self.assertFalse(self.validator("projection-package.schema.json").is_valid(package)) receipt = self.receipt().as_dict() receipt["artifacts"] = [ copy.deepcopy(receipt["artifacts"][0]) for _ in range(33) # type: ignore[index] ] self.assertFalse(self.validator("projection-receipt.schema.json").is_valid(receipt)) def test_canonical_identity_equality_remains_a_runtime_check(self) -> None: document = self.manual_plan().as_dict() document["plan_id"] = "f" * 64 self.validator("manual-render-plan.schema.json").validate(document) with self.assertRaises(DocForgeError) as raised: ManualRenderPlanV1.from_dict(document) self.assertEqual("invalid_projection", raised.exception.code) if __name__ == "__main__": unittest.main()