from __future__ import annotations import shutil import tempfile import unittest from pathlib import Path from unittest import mock from docforge.application import CanonicalApplicationService from docforge.errors import DocForgeError from docforge.graph_rendering import GraphRenderService from docforge.index import ProjectIndex from docforge.mcp_server import DocForgeService from docforge.project import Project from docforge.rendering import RenderService from docforge.viewer_manager import ViewerManagerClient 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 include_logic = false """ class ProjectionPolicyIntegrationTests(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) def project(self, *, graph: bool = False) -> Project: if graph: descriptor = self.root / ".docforge/project.toml" descriptor.write_text( descriptor.read_text(encoding="utf-8") + GRAPH_CONFIG, encoding="utf-8", ) return Project.open(self.root) def test_disabled_manual_blocks_work_before_load_but_status_remains_receipt_only(self) -> None: project = self.project() service = RenderService(project, manual_policy="disabled") with mock.patch.object( project, "load", side_effect=AssertionError("disabled manual policy must fail before loading"), ): status = service.status("manual") self.assertEqual("stale", status["state"]) for operation in ( lambda: service.render("manual"), lambda: service.deep_status("manual"), lambda: service.preview("proposal", "manual"), ): with self.subTest(operation=operation), self.assertRaises(DocForgeError) as raised: operation() self.assertEqual( "projection_policy_forbids_operation", raised.exception.code, ) def test_disabled_graph_blocks_plan_and_render_but_status_remains_available(self) -> None: project = self.project(graph=True) service = GraphRenderService(project, portable_graph_policy="disabled") with mock.patch.object( project, "load", side_effect=AssertionError("disabled graph policy must fail before loading"), ): status = service.status("architecture") self.assertEqual("stale", status["state"]) for operation in ( lambda: service.plan("architecture"), lambda: service.render("architecture"), ): with self.subTest(operation=operation), self.assertRaises(DocForgeError) as raised: operation() self.assertEqual( "projection_policy_forbids_operation", raised.exception.code, ) def test_live_viewer_disabled_blocks_start_before_index_work_but_allows_lifecycle(self) -> None: project = self.project() index = ProjectIndex(project) client = ViewerManagerClient(index, live_viewer_policy="disabled") with ( mock.patch.object( index, "check", side_effect=AssertionError("disabled viewer must fail before index work"), ), self.assertRaises(DocForgeError) as raised, ): client.start(node_id="guide.workflow") self.assertEqual("projection_policy_forbids_operation", raised.exception.code) with mock.patch.object( client, "_lifecycle_request", side_effect=( {"status": "ok", "state": "not_running"}, {"status": "ok", "state": "stopped"}, ), ): self.assertEqual("not_running", client.status()["state"]) self.assertEqual("stopped", client.stop()["state"]) def test_application_auto_renders_and_explicit_or_disabled_skip_cleanly(self) -> None: project = self.project() for mode, expected_action, expected_calls in ( ("auto", "rendered", 1), ("explicit", "skipped_explicit", 0), ("disabled", "skipped_disabled", 0), ): with self.subTest(mode=mode): service = CanonicalApplicationService( project, applier_id="alpha-editor", applier=mock.Mock(), manual_policy=mode, # type: ignore[arg-type] ) with ( mock.patch.object( service.changesets, "apply", return_value={"status": "ok", "applied": True}, ), mock.patch.object( service.index, "build", return_value={"status": "ok"}, ), mock.patch.object( service.index, "check", return_value={"status": "ok"}, ), mock.patch.object( service.rendering, "render", return_value={"status": "ok", "state": "current"}, ) as rendered, ): result = service.apply("policy-application", "a" * 64) refresh = result["derived_refresh"] assert isinstance(refresh, dict) self.assertEqual( {"mode": mode, "action": expected_action}, refresh["render_policy"], ) self.assertEqual(expected_calls, rendered.call_count) self.assertEqual("ok", refresh["status"]) def test_mcp_exposes_v2_without_changing_v1_policy_or_legacy_render_projection(self) -> None: project = self.project(graph=True) baseline = DocForgeService(project, capability_mode_name="read") narrowed = DocForgeService( project, capability_mode_name="read", manual_projection_policy="disabled", portable_graph_policy="disabled", live_viewer_policy="disabled", ) self.assertEqual(baseline.policy.as_dict(), narrowed.policy.as_dict()) result = narrowed.bootstrap() self.assertEqual(narrowed.policy.as_dict(), result["effective_policy"]) self.assertEqual( narrowed.projection_policy.as_dict(), result["projection_policy"], ) self.assertEqual( narrowed.projection_policy.policy_hash, result["projection_policy_hash"], ) session = result["session_contract"] assert isinstance(session, dict) self.assertEqual( { "manual": narrowed.policy.manual_render, "graph": narrowed.policy.graph_render, "live_viewer": narrowed.policy.live_viewer, }, session["render_policies"], ) self.assertEqual( "stale", narrowed.graph_render_status("architecture")["state"], ) blocked_plan = narrowed.graph_plan("architecture") self.assertEqual( "projection_policy_forbids_operation", blocked_plan["error"]["code"], ) def test_direct_service_policy_values_are_runtime_validated(self) -> None: project = self.project(graph=True) factories = ( lambda: RenderService(project, manual_policy="bogus"), # type: ignore[arg-type] lambda: GraphRenderService( project, portable_graph_policy="bogus", # type: ignore[arg-type] ), lambda: ViewerManagerClient( ProjectIndex(project), live_viewer_policy="bogus", # type: ignore[arg-type] ), lambda: CanonicalApplicationService( project, applier_id=None, applier=None, manual_policy="bogus", # type: ignore[arg-type] ), ) for factory in factories: with self.subTest(factory=factory), self.assertRaises(DocForgeError) as raised: factory() self.assertEqual("invalid_projection_policy", raised.exception.code) if __name__ == "__main__": unittest.main()