test_mcp_hitl.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# tests/tests_mcp/test_mcp_hitl.py
 
"""
Purpose
-------
Integration tests for the Phase 5B HITL (Human-in-the-Loop) pipeline.
 
Responsibilities
----------------
- Verify write-tool first-call (confirm=False) returns confirmation
  string; second-call (confirm=True) executes via mock sandbox.
- Verify oct_format double-guard (apply + confirm) and dry-run path.
- Verify audit records for both HITL calls.
- Verify policy denial and validation failure short-circuit before HITL.
- Verify rate-limit state preserved across calls.
- Verify T-06 privilege escalation and T-06b strict-profile denial.
 
Diagnostics
-----------
Domain: MCP-TESTS
Levels:
    L2 — test lifecycle
    L3 — assertion details
    L4 — deep tracing
 
Contracts
---------
- Tests use mock sandbox and ``tmp_path``; no real subprocess is
  spawned.
"""
 
from __future__ import annotations
 
import json
from pathlib import Path
from unittest.mock import MagicMock, patch
 
import pytest
from pydantic import ValidationError
 
from oct.mcp.audit import McpAuditLogger
from oct.mcp.config import McpConfig
from oct.mcp.executor import ToolExecutor
from oct.mcp.metrics import McpMetrics, PROMETHEUS_AVAILABLE
from oct.mcp.policy import McpPolicy
from oct.mcp.redactor import Redactor
from oct.mcp.safety import HITL_SENTINEL, SafetyGate
from oct.mcp.sandbox import SandboxExecutor, SandboxResult
from oct.mcp.server import RateLimiter, ServerState
from oct.mcp.tools import _run_tool_write
 
 
# -------------------------------------------------------------------
# Fixtures / helpers
# -------------------------------------------------------------------
 
def _isolated_metrics() -> McpMetrics:
    """McpMetrics on a private CollectorRegistry — avoids
    ``Duplicated timeseries`` collisions across tests."""
    if not PROMETHEUS_AVAILABLE:
        return McpMetrics()
    from prometheus_client import CollectorRegistry
    return McpMetrics(registry=CollectorRegistry())
 
 
def _make_write_state(
    tmp_path: Path,
    sandbox_output: str = "Done.",
    sandbox_exit_code: int = 0,
    profile: str = "default",
    safe_mode: bool = False,
    rate_limit: int = 30,
    session_id: str = "hitl-test-session",
) -> tuple[ServerState, McpAuditLogger]:
    """Build a ServerState with SafetyGate for HITL integration tests."""
    (tmp_path / "pyproject.toml").write_text(
        '[project]\nname="test"\nversion="0.1"\n'
    )
    mock_sandbox = MagicMock(spec=SandboxExecutor)
    mock_sandbox.run.return_value = SandboxResult(
        exit_code=sandbox_exit_code,
        stdout=sandbox_output,
        stderr="",
    )
    audit_log = tmp_path / "audit.log"
    audit_logger = McpAuditLogger(audit_log_path=audit_log)
    config = McpConfig()
    config.safe_mode = safe_mode
    state = ServerState(
        project_root=tmp_path,
        config=config,
        policy=McpPolicy(profile=profile, safe_mode=safe_mode),
        executor=ToolExecutor(sandbox=mock_sandbox),
        redactor=Redactor(),
        audit_logger=audit_logger,
        rate_limiter=RateLimiter(limit_per_minute=rate_limit),
        safety=SafetyGate(config),
        metrics=_isolated_metrics(),
        session_id=session_id,
    )
    return state, audit_logger
 
 
def _last_audit_record(logger: McpAuditLogger) -> dict:
    """Read the last line from the audit log as a dict."""
    assert logger.current_path is not None
    lines = logger.current_path.read_text(encoding="utf-8").strip().splitlines()
    return json.loads(lines[-1])
 
 
# -------------------------------------------------------------------
# First call (confirm=False) → requires_confirmation
# -------------------------------------------------------------------
 
class TestHITLFirstCallRequiresConfirmation:
    @pytest.mark.parametrize("tool_name,extra_args", [
        ("oct_clean", {}),
        ("oct_scaffold", {"name": "myproject"}),
        ("oct_docs", {}),
        ("oct_export_source", {}),
        ("oct_install_hooks", {}),
        ("oct_git_hooks_install", {}),
        ("oct_git_init", {}),
        ("oct_git_changelog", {}),
    ])
    def test_first_call_returns_confirmation_string(
        self, tool_name, extra_args, tmp_path: Path
    ):
        state, _ = _make_write_state(tmp_path)
        import oct.mcp.server as _server_module
        _server_module._server_state = state
        args = {"confirm": False, **extra_args}
        result = _run_tool_write(tool_name, args)
        assert result.startswith(HITL_SENTINEL), (
            f"{tool_name}: expected HITL_SENTINEL, got: {result!r}"
        )
 
    def test_first_call_writes_requires_confirmation_audit(self, tmp_path: Path):
        state, logger = _make_write_state(tmp_path)
        import oct.mcp.server as _server_module
        _server_module._server_state = state
        _run_tool_write("oct_clean", {"confirm": False})
        record = _last_audit_record(logger)
        assert record["decision"] == "requires_confirmation"
 
    def test_first_call_sandbox_not_called(self, tmp_path: Path):
        state, _ = _make_write_state(tmp_path)
        import oct.mcp.server as _server_module
        _server_module._server_state = state
        _run_tool_write("oct_clean", {"confirm": False})
        # The sandbox mock should NOT have been called.
        state.executor._sandbox.run.assert_not_called()
 
    def test_format_first_call_requires_confirmation(self, tmp_path: Path):
        state, _ = _make_write_state(tmp_path)
        import oct.mcp.server as _server_module
        _server_module._server_state = state
        result = _run_tool_write("oct_format", {"confirm": False, "apply": False})
        assert result.startswith(HITL_SENTINEL)
 
    def test_git_commit_first_call_requires_confirmation(self, tmp_path: Path):
        state, _ = _make_write_state(tmp_path)
        import oct.mcp.server as _server_module
        _server_module._server_state = state
        result = _run_tool_write("oct_git_commit", {"confirm": False, "message": "feat: test"})
        assert result.startswith(HITL_SENTINEL)
 
 
# -------------------------------------------------------------------
# Second call (confirm=True) → executes
# -------------------------------------------------------------------
 
class TestHITLSecondCallExecutes:
    def test_second_call_executes_and_returns_output(self, tmp_path: Path):
        state, _ = _make_write_state(tmp_path, sandbox_output="Files cleaned.")
        import oct.mcp.server as _server_module
        _server_module._server_state = state
        result = _run_tool_write("oct_clean", {"confirm": True, "dry_run": True})
        assert result == "Files cleaned."
 
    def test_second_call_writes_allowed_audit(self, tmp_path: Path):
        state, logger = _make_write_state(tmp_path, sandbox_output="Done.")
        import oct.mcp.server as _server_module
        _server_module._server_state = state
        _run_tool_write("oct_clean", {"confirm": True, "dry_run": True})
        record = _last_audit_record(logger)
        assert record["decision"] == "allowed"
 
    def test_second_call_sandbox_called(self, tmp_path: Path):
        state, _ = _make_write_state(tmp_path)
        import oct.mcp.server as _server_module
        _server_module._server_state = state
        _run_tool_write("oct_clean", {"confirm": True})
        state.executor._sandbox.run.assert_called_once()
 
    def test_git_commit_second_call_executes(self, tmp_path: Path):
        state, _ = _make_write_state(tmp_path, sandbox_output='{"exit_code": 0}')
        import oct.mcp.server as _server_module
        _server_module._server_state = state
        result = _run_tool_write("oct_git_commit", {"confirm": True, "message": "feat: test"})
        assert isinstance(result, str)
        state.executor._sandbox.run.assert_called_once()
 
 
# -------------------------------------------------------------------
# oct_format double guard
# -------------------------------------------------------------------
 
class TestFormatDoubleGuardE2E:
    def test_format_confirm_true_apply_false_dry_run_in_argv(self, tmp_path: Path):
        """E2E: apply=False, confirm=True → executor sees --dry-run."""
        state, _ = _make_write_state(tmp_path, sandbox_output="[DRY RUN] no changes")
        import oct.mcp.server as _server_module
        _server_module._server_state = state
        _run_tool_write("oct_format", {"confirm": True, "apply": False})
        # Extract the cmd that was passed to the sandbox mock.
        call_args = state.executor._sandbox.run.call_args
        cmd = call_args[1]["cmd"] if "cmd" in call_args[1] else call_args[0][0]
        assert "--dry-run" in cmd
        assert "--fix" not in cmd
 
    def test_format_confirm_true_apply_true_fix_in_argv(self, tmp_path: Path):
        """E2E: apply=True, confirm=True → executor sees --fix."""
        state, _ = _make_write_state(tmp_path, sandbox_output="Fixed 3 files.")
        import oct.mcp.server as _server_module
        _server_module._server_state = state
        _run_tool_write("oct_format", {"confirm": True, "apply": True})
        call_args = state.executor._sandbox.run.call_args
        cmd = call_args[1]["cmd"] if "cmd" in call_args[1] else call_args[0][0]
        assert "--fix" in cmd
        assert "--dry-run" not in cmd
 
    def test_format_confirm_false_no_sandbox_call(self, tmp_path: Path):
        """E2E: confirm=False → sandbox never called."""
        state, _ = _make_write_state(tmp_path)
        import oct.mcp.server as _server_module
        _server_module._server_state = state
        _run_tool_write("oct_format", {"confirm": False, "apply": True})
        state.executor._sandbox.run.assert_not_called()
 
 
# -------------------------------------------------------------------
# Both calls write audit records
# -------------------------------------------------------------------
 
class TestBothCallsWriteAuditRecords:
    def test_first_call_audit_then_second_call_audit(self, tmp_path: Path):
        state, logger = _make_write_state(tmp_path, sandbox_output="Done.")
        import oct.mcp.server as _server_module
        _server_module._server_state = state
 
        # First call (confirm=False)
        _run_tool_write("oct_clean", {"confirm": False})
        assert logger.current_path is not None
        lines = logger.current_path.read_text(encoding="utf-8").strip().splitlines()
        assert len(lines) == 1
        r1 = json.loads(lines[0])
        assert r1["decision"] == "requires_confirmation"
 
        # Second call (confirm=True)
        _run_tool_write("oct_clean", {"confirm": True})
        lines = logger.current_path.read_text(encoding="utf-8").strip().splitlines()
        assert len(lines) == 2
        r2 = json.loads(lines[1])
        assert r2["decision"] == "allowed"
 
 
# -------------------------------------------------------------------
# Policy denial short-circuits before HITL gate
# -------------------------------------------------------------------
 
class TestPolicyShortCircuit:
    def test_strict_profile_denies_before_hitl(self, tmp_path: Path):
        state, logger = _make_write_state(tmp_path, profile="strict")
        import oct.mcp.server as _server_module
        _server_module._server_state = state
        result = _run_tool_write("oct_format", {"confirm": True, "apply": True})
        assert "[oct-mcp error]" in result
        record = _last_audit_record(logger)
        assert record["decision"] == "denied"
        assert record["policy_rule"] == "strict_no_destructive"
        state.executor._sandbox.run.assert_not_called()
 
    def test_safe_mode_denies_before_hitl(self, tmp_path: Path):
        state, logger = _make_write_state(tmp_path, safe_mode=True)
        import oct.mcp.server as _server_module
        _server_module._server_state = state
        result = _run_tool_write("oct_clean", {"confirm": True})
        assert "[oct-mcp error]" in result
        record = _last_audit_record(logger)
        assert record["decision"] == "denied"
        state.executor._sandbox.run.assert_not_called()
 
 
# -------------------------------------------------------------------
# Validation failure short-circuits before HITL gate
# -------------------------------------------------------------------
 
class TestValidationShortCircuit:
    def test_validation_failure_returns_error(self, tmp_path: Path):
        state, logger = _make_write_state(tmp_path)
        import oct.mcp.server as _server_module
        _server_module._server_state = state
        # no_verify is not a valid field for oct_git_commit
        result = _run_tool_write("oct_git_commit", {"no_verify": True, "confirm": True})
        assert "[oct-mcp error]" in result.lower() or "validation" in result.lower()
 
    def test_validation_failure_writes_audit(self, tmp_path: Path):
        state, logger = _make_write_state(tmp_path)
        import oct.mcp.server as _server_module
        _server_module._server_state = state
        _run_tool_write("oct_git_commit", {"no_verify": True, "confirm": True})
        record = _last_audit_record(logger)
        assert record["decision"] == "denied"
 
    def test_missing_required_field_returns_error(self, tmp_path: Path):
        state, _ = _make_write_state(tmp_path)
        import oct.mcp.server as _server_module
        _server_module._server_state = state
        # oct_new requires 'path'
        result = _run_tool_write("oct_new", {"confirm": True})
        assert "[oct-mcp error]" in result.lower() or "validation" in result.lower()
 
 
# -------------------------------------------------------------------
# Rate limit preserved across HITL round trips
# -------------------------------------------------------------------
 
class TestRateLimitAcrossHITL:
    def test_rate_limit_counts_both_calls(self, tmp_path: Path):
        """Rate limiter counts both the first (confirmation) and second (execute) calls."""
        state, _ = _make_write_state(tmp_path, rate_limit=2)
        import oct.mcp.server as _server_module
        _server_module._server_state = state
 
        # Call 1: confirmation
        r1 = _run_tool_write("oct_clean", {"confirm": False})
        assert r1.startswith(HITL_SENTINEL)
 
        # Call 2: execute
        r2 = _run_tool_write("oct_clean", {"confirm": True})
        assert not r2.startswith(HITL_SENTINEL)  # executed
 
        # Call 3: rate limited
        r3 = _run_tool_write("oct_clean", {"confirm": True})
        assert "rate limit" in r3.lower() or "[oct-mcp error]" in r3.lower()
 
 
# -------------------------------------------------------------------
# T-06: privilege escalation attempt
# -------------------------------------------------------------------
 
class TestPrivilegeEscalation:
    def test_t06_no_verify_field_rejected_at_validation(self, tmp_path: Path):
        """T-06: no_verify in git commit → ValidationError → error response."""
        state, _ = _make_write_state(tmp_path)
        import oct.mcp.server as _server_module
        _server_module._server_state = state
        result = _run_tool_write(
            "oct_git_commit",
            {"message": "feat: test", "confirm": True, "no_verify": True},
        )
        # Must be an error, not executed
        assert "[oct-mcp error]" in result or "validation" in result.lower()
        state.executor._sandbox.run.assert_not_called()
 
    def test_t06b_strict_profile_blocks_write(self, tmp_path: Path):
        """T-06b: write tools in strict profile → denied."""
        state, _ = _make_write_state(tmp_path, profile="strict")
        import oct.mcp.server as _server_module
        _server_module._server_state = state
        for tool_name, extra in [
            ("oct_format", {"apply": True}),
            ("oct_clean", {}),
            ("oct_git_commit", {"message": "feat: test"}),
        ]:
            result = _run_tool_write(tool_name, {"confirm": True, **extra})
            assert "[oct-mcp error]" in result, f"{tool_name} should be denied in strict"
            state.executor._sandbox.run.assert_not_called()