from __future__ import annotations import json import tempfile import unittest from dataclasses import replace from pathlib import Path from typing import Any, cast from docforge.errors import DocForgeError from docforge.graph_projection import ( GraphViewRequestV1, build_graph_view_plan, ) from docforge.models import Edge, Limits, Node, ProjectDescriptor, ProjectSnapshot from docforge.projection_contract import GraphViewPlanV1 def _document(plan: GraphViewPlanV1) -> dict[str, Any]: return cast(dict[str, Any], plan.as_dict()) def _node( node_id: str, *, title: str | None = None, family: str = "code", authority: str = "derived", status: str = "active", tags: tuple[str, ...] = (), ) -> Node: return Node( node_id=node_id, title=title or node_id, family=family, authority=authority, status=status, tags=tags, summary=f"Summary for {node_id}", content=f"SECRET SOURCE BODY {node_id}", source_path=f"/private/source/{node_id}.py", source_anchor=f"line-{len(node_id)}", content_hash=(node_id.encode("utf-8").hex() + "0" * 64)[:64], ) def _snapshot( root: Path, nodes: tuple[Node, ...], edges: tuple[Edge, ...], ) -> ProjectSnapshot: descriptor = ProjectDescriptor( schema_version=1, project_id="graph-project", title="Graph project", adapter="generic", root=root, descriptor_path=root / ".docforge" / "project.toml", descriptor_hash="d" * 64, content_roots=(root / "docs",), authority_files=(), cache_root=root / ".docforge" / "cache", index_path=root / ".docforge" / "cache" / "index.sqlite3", changeset_root=root / ".docforge" / "changesets", proposal_writers=(), render=None, allowed_relations=tuple(sorted({edge.relation for edge in edges})), profiles=(), limits=Limits(), ) return ProjectSnapshot( descriptor=descriptor, nodes=nodes, edges=edges, revision="revision-1", source_hash="a" * 64, ) class GraphProjectionTests(unittest.TestCase): def test_exact_root_plan_is_deterministic_sorted_and_path_free(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) nodes = ( _node("c", family="docs", tags=("python",)), _node("a", tags=("python", "callable")), _node("b", tags=("python",)), _node("unrelated"), ) edges = ( Edge("b", "calls", "c"), Edge("a", "calls", "b"), Edge("unrelated", "calls", "c"), ) request = GraphViewRequestV1( view_id="architecture", title="Architecture", root_node_id="a", depth=2, max_nodes=10, max_edges=10, max_work=100, ) first = build_graph_view_plan(_snapshot(root, nodes, edges), request, True) second = build_graph_view_plan( _snapshot(root, tuple(reversed(nodes)), tuple(reversed(edges))), request, True, ) self.assertEqual(first.as_dict(), second.as_dict()) GraphViewPlanV1.from_dict(first.as_dict()) document = _document(first) self.assertEqual( ["a", "b", "c"], [node["node_id"] for node in document["graph"]["nodes"]] ) self.assertEqual( [ {"source_id": "a", "relation": "calls", "target_id": "b"}, {"source_id": "b", "relation": "calls", "target_id": "c"}, ], document["graph"]["edges"], ) encoded = json.dumps(document, sort_keys=True) self.assertNotIn(str(root), encoded) self.assertNotIn("SECRET SOURCE BODY", encoded) self.assertNotIn("/private/source/", encoded) self.assertNotIn("line-1", encoded) self.assertEqual("excluded", document["policy"]["source_paths"]) self.assertEqual("excluded", document["policy"]["source_bodies"]) self.assertEqual("allowed", document["policy"]["logic"]) self.assertEqual("exact_root", document["view"]["scope"]["kind"]) def test_lexical_scope_uses_metadata_only_and_closed_filters(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) nodes = ( _node( "api.handler", title="Request handler", family="code", authority="derived", tags=("python", "route"), ), _node( "api.test", title="Handler proof", family="test", authority="approved_plan", tags=("python", "test"), ), replace( _node("hidden.body", family="code", tags=("python",)), content="request handler appears only in the forbidden source body", ), ) edges = ( Edge("api.handler", "tested_by", "api.test"), Edge("hidden.body", "relates_to", "api.handler"), ) request = GraphViewRequestV1( view_id="routes", title="Routes", query="request handler", families=("code",), authorities=("derived",), tags=("python", "route"), relations=("tested_by",), max_nodes=10, max_edges=10, max_work=100, ) plan = build_graph_view_plan(_snapshot(root, nodes, edges), request, False) document = _document(plan) self.assertEqual( ["api.handler"], [node["node_id"] for node in document["graph"]["nodes"]], ) self.assertEqual([], document["graph"]["edges"]) self.assertEqual( { "families": ["code"], "relations": ["tested_by"], "authorities": ["derived"], "statuses": [], "tags": ["python", "route"], }, document["view"]["filters"], ) self.assertEqual("lexical", document["view"]["scope"]["kind"]) def test_result_and_work_limits_emit_explicit_omissions(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) nodes = tuple(_node(value) for value in ("a", "b", "c", "d")) edges = ( Edge("a", "calls", "b"), Edge("a", "calls", "c"), Edge("a", "calls", "d"), Edge("b", "calls", "c"), ) result_limited = build_graph_view_plan( _snapshot(root, nodes, edges), GraphViewRequestV1( view_id="limited", title="Limited", root_node_id="a", max_nodes=2, max_edges=0, max_work=100, ), False, ) result_limited = _document(result_limited) self.assertEqual( ["a", "b"], [node["node_id"] for node in result_limited["graph"]["nodes"]] ) self.assertEqual([], result_limited["graph"]["edges"]) self.assertEqual( ["edge_result_limit", "node_result_limit"], [item["code"] for item in result_limited["omissions"]], ) work_limited = build_graph_view_plan( _snapshot(root, nodes, edges), GraphViewRequestV1( view_id="work", title="Work", root_node_id="a", max_nodes=10, max_edges=10, max_work=1, ), False, ) work_limited = _document(work_limited) self.assertIn( "work_limit", [item["code"] for item in work_limited["omissions"]], ) self.assertEqual( 1, work_limited["diagnostics"]["examined_work_units"], ) self.assertTrue(work_limited["diagnostics"]["truncated"]) def test_no_ast_policy_excludes_requested_logic(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) snapshot = _snapshot(root, (_node("a"),), ()) request = GraphViewRequestV1( view_id="logic", title="Logic", root_node_id="a", initial_mode="logic", include_logic=True, ) blocked = _document(build_graph_view_plan(snapshot, request, False)) self.assertEqual("forbidden", blocked["policy"]["logic"]) self.assertEqual([], blocked["graph"]["logic_projections"]) self.assertIn( "logic_forbidden", [item["code"] for item in blocked["omissions"]], ) allowed = _document(build_graph_view_plan(snapshot, request, True)) self.assertEqual("allowed", allowed["policy"]["logic"]) self.assertNotIn( "logic_forbidden", [item["code"] for item in allowed["omissions"]], ) def test_edge_and_node_filters_constrain_exact_root_bfs(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) nodes = ( _node("root", family="code", status="active"), _node("code-child", family="code", status="active"), _node("doc-child", family="docs", status="active"), _node("old-child", family="code", status="historical"), ) edges = ( Edge("root", "calls", "code-child"), Edge("root", "documents", "doc-child"), Edge("root", "calls", "old-child"), ) plan = build_graph_view_plan( _snapshot(root, nodes, edges), GraphViewRequestV1( view_id="filtered", title="Filtered", root_node_id="root", families=("code",), statuses=("active",), relations=("calls",), max_work=100, ), False, ) plan = _document(plan) self.assertEqual( ["code-child", "root"], [node["node_id"] for node in plan["graph"]["nodes"]] ) self.assertEqual( [{"source_id": "root", "relation": "calls", "target_id": "code-child"}], plan["graph"]["edges"], ) def test_invalid_requests_and_graphs_fail_closed(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) snapshot = _snapshot(root, (_node("a"),), ()) invalid_requests = ( GraphViewRequestV1(view_id="v", title="V"), GraphViewRequestV1( view_id="v", title="V", root_node_id="a", query="a", ), GraphViewRequestV1( view_id="v", title="V", root_node_id="a", max_nodes=0, ), GraphViewRequestV1( view_id="v", title="V", query="***", ), GraphViewRequestV1( view_id="v", title="V", root_node_id="a", families=("code", "code"), ), ) for request in invalid_requests: with self.subTest(request=request), self.assertRaises(DocForgeError) as error: build_graph_view_plan(snapshot, request, False) self.assertEqual("invalid_graph_view_request", error.exception.code) with self.assertRaises(DocForgeError) as missing: build_graph_view_plan( snapshot, GraphViewRequestV1( view_id="v", title="V", root_node_id="missing", ), False, ) self.assertEqual("missing_node", missing.exception.code) duplicate = _snapshot(root, (_node("a"), _node("a")), ()) with self.assertRaises(DocForgeError) as invalid: build_graph_view_plan( duplicate, GraphViewRequestV1( view_id="v", title="V", root_node_id="a", ), False, ) self.assertEqual("invalid_projection", invalid.exception.code) def test_plan_identity_changes_with_generation_request_and_policy(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) snapshot = _snapshot(root, (_node("a"),), ()) request = GraphViewRequestV1( view_id="v", title="V", root_node_id="a", ) base = build_graph_view_plan(snapshot, request, False) self.assertEqual( base.plan_id, build_graph_view_plan(snapshot, request, False).plan_id, ) self.assertNotEqual( base.plan_id, build_graph_view_plan( replace(snapshot, source_hash="b" * 64), request, False, ).plan_id, ) self.assertNotEqual( base.plan_id, build_graph_view_plan( snapshot, replace(request, title="Other"), False, ).plan_id, ) self.assertNotEqual( base.plan_id, build_graph_view_plan(snapshot, request, True).plan_id, ) if __name__ == "__main__": unittest.main()