DocForge2/tests/test_policy.py

from __future__ import annotations

import json
import shutil
import tempfile
import unittest
from pathlib import Path
from unittest import mock

from jsonschema import Draft202012Validator

from docforge.application import GenericCanonicalApplier
from docforge.errors import DocForgeError
from docforge.index import ProjectIndex
from docforge.mcp_server import (
    ALL_TOOLS,
    APPLICATION_TOOLS,
    READ_TOOLS,
    DocForgeService,
)
from docforge.policy import (
    POLICY_PRECEDENCE,
    capability_mode,
    compose_effective_policy,
)
from docforge.project import Project

ROOT = Path(__file__).resolve().parents[1]
FIXTURES = ROOT / "tests" / "fixtures"
POLICY_SCHEMA = json.loads((ROOT / "schemas" / "policy.schema.json").read_text(encoding="utf-8"))


class EffectivePolicyTests(unittest.TestCase):
    def copy_fixture(self, destination: Path) -> Path:
        root = destination / "alpha"
        shutil.copytree(FIXTURES / "alpha", root)
        return root

    def test_policy_schema_and_legacy_adapter_projection_are_exact(self) -> None:
        standard = compose_effective_policy(
            selected_mode="proposal",
            capability_source="factory_default",
            no_ast=False,
            diagnostics=False,
            render_configured=True,
            application_enabled=False,
        )
        preserve = compose_effective_policy(
            selected_mode="read",
            capability_source="explicit",
            no_ast=True,
            diagnostics=True,
            render_configured=False,
            application_enabled=False,
        )
        validator = Draft202012Validator(POLICY_SCHEMA)
        validator.validate(standard.as_dict())
        validator.validate(preserve.as_dict())
        self.assertEqual(list(POLICY_PRECEDENCE), preserve.as_dict()["precedence"])
        self.assertEqual(
            {
                "mode": "standard",
                "ast_analysis": "allowed",
                "logic_projection": "allowed",
                "incremental_extraction": "allowed",
                "adapter_rewrite": "not_requested",
            },
            standard.adapter_policy(),
        )
        self.assertEqual("preserve-no-ast", preserve.adapter_policy()["mode"])
        self.assertEqual(["docforge_get_logic"], preserve.adapter_policy()["blocked_tools"])
        self.assertEqual("off", preserve.as_dict()["logic_indexing"])
        self.assertEqual("enabled", preserve.as_dict()["profiling"])

    def test_invalid_or_unavailable_capability_fails_closed(self) -> None:
        with self.assertRaises(DocForgeError) as invalid:
            capability_mode("admin", default="read")
        self.assertEqual("invalid_capability_mode", invalid.exception.code)
        with self.assertRaises(DocForgeError) as unavailable:
            compose_effective_policy(
                selected_mode="application",
                capability_source="explicit",
                no_ast=False,
                diagnostics=False,
                render_configured=True,
                application_enabled=False,
            )
        self.assertEqual("capability_unavailable", unavailable.exception.code)

    def test_modes_preserve_default_surfaces_and_narrow_authority(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = self.copy_fixture(Path(directory))
            project = Project.open(root)

            read = DocForgeService(
                project,
                proposal_writer="alpha-editor",
                capability_mode_name="read",
            )
            proposal = DocForgeService(
                project,
                proposal_writer="alpha-editor",
                capability_mode_name="proposal",
            )
            application = DocForgeService(
                project,
                proposal_writer="alpha-editor",
                canonical_applier_id="alpha-editor",
                canonical_applier=GenericCanonicalApplier(project),
                capability_mode_name="application",
            )
            operator = DocForgeService(
                project,
                proposal_writer="alpha-editor",
                capability_mode_name="operator",
            )

            self.assertEqual(READ_TOOLS, read.tool_surface)
            self.assertFalse(read.changesets.access()["enabled"])
            self.assertEqual(ALL_TOOLS, proposal.tool_surface)
            self.assertTrue(proposal.changesets.access()["enabled"])
            self.assertFalse(proposal.application.enabled)
            self.assertEqual((*ALL_TOOLS, *APPLICATION_TOOLS), application.tool_surface)
            self.assertTrue(application.application.enabled)
            self.assertEqual(ALL_TOOLS, operator.tool_surface)
            self.assertTrue(operator.capabilities()["operator"]["enabled"])

    def test_bootstrap_reuses_synchronized_identity_and_filters_workflow(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = self.copy_fixture(Path(directory))
            project = Project.open(root)
            ProjectIndex(project).build()
            read = DocForgeService(project, capability_mode_name="read")
            with mock.patch.object(
                project,
                "load",
                side_effect=AssertionError("bootstrap must not reload after synchronization"),
            ):
                result = read.bootstrap()

            self.assertEqual("ok", result["status"])
            self.assertEqual(1, result["session_contract"]["schema_version"])
            self.assertEqual("read", result["effective_policy"]["capability_mode"])
            self.assertEqual(
                result["effective_policy"], result["session_contract"]["effective_policy"]
            )
            self.assertNotIn("docforge_register_changes", result["recommended_workflow"])
            self.assertNotIn("docforge_apply_changeset", result["recommended_workflow"])
            self.assertEqual(
                "docforge_get_task_context",
                result["session_contract"]["recommended_first_operation"]["tool"],
            )

            proposal = DocForgeService(
                project,
                proposal_writer="alpha-editor",
                capability_mode_name="proposal",
            ).bootstrap()
            self.assertIn("docforge_register_changes", proposal["recommended_workflow"])
            self.assertIn("docforge_get_changeset_diff", proposal["recommended_workflow"])
            self.assertNotIn("docforge_apply_changeset", proposal["recommended_workflow"])