test_mcp_policy.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# tests/tests_mcp/test_mcp_policy.py
 
"""
Purpose
-------
Regression tests for :mod:`oct.mcp.policy` — ``TOOL_MANIFEST``,
``ToolSpec``, ``McpPolicy``, and ``PolicyDecision``.
 
Responsibilities
----------------
- Verify all profile behaviours (default, strict, airgapped).
- Verify safe_mode enforcement and unknown tool rejection.
- Verify manifest completeness for registered tools.
 
Diagnostics
-----------
Domain: MCP-TESTS
Levels:
    L2 — test lifecycle
    L3 — assertion details
    L4 — deep tracing
 
Contracts
---------
- Tests are in-memory policy checks; no subprocess or filesystem writes.
"""
 
from __future__ import annotations
 
import pytest
 
from oct.mcp.policy import (
    TOOL_MANIFEST,
    McpPolicy,
    PolicyDecision,
    ToolSpec,
)
 
 
# -------------------------------------------------------------------
# TOOL_MANIFEST
# -------------------------------------------------------------------
 
_PHASE_5A_TOOLS = {
    "oct_lint", "oct_health", "oct_skeleton", "oct_deps",
    "oct_test", "oct_typecheck", "oct_diag",
    "oct_git_status", "oct_git_check",
}
 
_PHASE_5B_TOOLS = {
    "oct_format", "oct_docs", "oct_export_source", "oct_new",
    "oct_scaffold", "oct_clean", "oct_install_hooks",
    "oct_git_commit", "oct_git_hooks_install", "oct_git_init",
    "oct_git_changelog",
}
 
 
class TestToolManifest:
    def test_twenty_tools_present(self):
        assert len(TOOL_MANIFEST) == 20
 
    def test_all_phase_5a_tools_present(self):
        assert _PHASE_5A_TOOLS.issubset(set(TOOL_MANIFEST.keys()))
 
    def test_all_phase_5b_tools_present(self):
        assert _PHASE_5B_TOOLS.issubset(set(TOOL_MANIFEST.keys()))
 
    def test_phase_5a_tools_non_destructive(self):
        for name in _PHASE_5A_TOOLS:
            spec = TOOL_MANIFEST[name]
            assert spec.destructive is False, f"{name} should not be destructive"
 
    def test_phase_5b_tools_destructive(self):
        for name in _PHASE_5B_TOOLS:
            spec = TOOL_MANIFEST[name]
            assert spec.destructive is True, f"{name} should be destructive"
 
    def test_phase_5a_tools_auto_approve(self):
        for name in _PHASE_5A_TOOLS:
            spec = TOOL_MANIFEST[name]
            assert spec.auto_approve is True, f"{name} should be auto_approve"
 
    def test_phase_5b_tools_not_auto_approve(self):
        for name in _PHASE_5B_TOOLS:
            spec = TOOL_MANIFEST[name]
            assert spec.auto_approve is False, f"{name} should not be auto_approve"
 
    def test_phase_5a_tools_have_read_permission(self):
        for name in _PHASE_5A_TOOLS:
            spec = TOOL_MANIFEST[name]
            assert "read" in spec.permissions, f"{name} must have read permission"
 
    def test_phase_5b_tools_have_write_permission(self):
        for name in _PHASE_5B_TOOLS:
            spec = TOOL_MANIFEST[name]
            assert "write" in spec.permissions, f"{name} must have write permission"
 
    def test_tool_spec_has_cli(self):
        for name, spec in TOOL_MANIFEST.items():
            assert spec.cli, f"{name} must have a cli subcommand"
 
 
# -------------------------------------------------------------------
# ToolSpec
# -------------------------------------------------------------------
 
class TestToolSpec:
    def test_permissions_coerced_to_frozenset(self):
        spec = ToolSpec(cli="lint", permissions={"read"})
        assert isinstance(spec.permissions, frozenset)
 
    def test_frozen(self):
        spec = ToolSpec(cli="lint", permissions=frozenset({"read"}))
        with pytest.raises((AttributeError, TypeError)):
            spec.cli = "other"  # frozen=True should prevent this
 
 
# -------------------------------------------------------------------
# McpPolicy — constructor
# -------------------------------------------------------------------
 
class TestMcpPolicyConstructor:
    def test_default_profile(self):
        policy = McpPolicy()
        assert policy.profile == "default"
 
    def test_strict_profile(self):
        policy = McpPolicy(profile="strict")
        assert policy.profile == "strict"
 
    def test_airgapped_profile(self):
        policy = McpPolicy(profile="airgapped")
        assert policy.profile == "airgapped"
 
    def test_invalid_profile_raises(self):
        with pytest.raises(ValueError, match="Invalid MCP profile"):
            McpPolicy(profile="superadmin")
 
 
# -------------------------------------------------------------------
# McpPolicy — default profile
# -------------------------------------------------------------------
 
class TestMcpPolicyDefault:
    def test_all_manifest_tools_allowed(self):
        policy = McpPolicy(profile="default")
        for tool in TOOL_MANIFEST:
            decision = policy.check(tool)
            assert decision.allowed, f"{tool} should be allowed in default profile"
 
    def test_unknown_tool_denied(self):
        policy = McpPolicy(profile="default")
        decision = policy.check("evil_tool")
        assert decision.allowed is False
        assert decision.decision == "denied"
        assert decision.policy_rule == "unknown_tool"
 
    def test_decision_has_reason(self):
        policy = McpPolicy(profile="default")
        decision = policy.check("oct_lint")
        assert decision.reason != ""
 
 
# -------------------------------------------------------------------
# McpPolicy — safe mode
# -------------------------------------------------------------------
 
class TestMcpPolicySafeMode:
    def test_safe_mode_denies_all_tools(self):
        policy = McpPolicy(safe_mode=True)
        for tool in TOOL_MANIFEST:
            decision = policy.check(tool)
            assert decision.allowed is False
            assert decision.policy_rule == "safe_mode"
 
    def test_safe_mode_reason_mentions_safe_mode(self):
        policy = McpPolicy(safe_mode=True)
        decision = policy.check("oct_lint")
        assert "safe mode" in decision.reason.lower()
 
 
# -------------------------------------------------------------------
# McpPolicy — airgapped profile
# -------------------------------------------------------------------
 
class TestMcpPolicyAirgapped:
    def test_airgapped_denies_all_tools(self):
        policy = McpPolicy(profile="airgapped")
        for tool in TOOL_MANIFEST:
            decision = policy.check(tool)
            assert decision.allowed is False
            assert decision.policy_rule == "airgapped_profile"
 
    def test_airgapped_unknown_tool_also_denied(self):
        policy = McpPolicy(profile="airgapped")
        decision = policy.check("unknown_tool")
        assert decision.allowed is False
 
    def test_airgapped_reason_mentions_profile(self):
        policy = McpPolicy(profile="airgapped")
        decision = policy.check("oct_lint")
        assert "airgapped" in decision.reason.lower()
 
 
# -------------------------------------------------------------------
# McpPolicy — strict profile
# -------------------------------------------------------------------
 
class TestMcpPolicyStrict:
    def test_strict_allows_non_destructive_tools(self):
        policy = McpPolicy(profile="strict")
        for tool in TOOL_MANIFEST:
            spec = TOOL_MANIFEST[tool]
            decision = policy.check(tool)
            if not spec.destructive:
                assert decision.allowed, f"{tool} should be allowed in strict (non-destructive)"
 
    def test_strict_denies_all_write_tools(self):
        """Phase 5B write tools (destructive=True) are all denied in strict profile."""
        policy = McpPolicy(profile="strict")
        for tool in _PHASE_5B_TOOLS:
            decision = policy.check(tool)
            assert decision.allowed is False, f"{tool} should be denied in strict"
            assert "destructive" in decision.policy_rule
 
 
# -------------------------------------------------------------------
# PolicyDecision
# -------------------------------------------------------------------
 
class TestPolicyDecision:
    def test_allowed_true_means_proceed(self):
        d = PolicyDecision(allowed=True, decision="allowed", policy_rule="x", reason="ok")
        assert d.allowed is True
 
    def test_allowed_false_means_blocked(self):
        d = PolicyDecision(allowed=False, decision="denied", policy_rule="safe_mode", reason="blocked")
        assert d.allowed is False