405 lines
15 KiB
Python
405 lines
15 KiB
Python
|
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()
|