test_mcp_metrics.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# tests/tests_mcp/test_mcp_metrics.py
 
"""
Purpose
-------
Phase 5C tests for :class:`oct.mcp.metrics.McpMetrics` (Prometheus
instrumentation).
 
Responsibilities
----------------
- Verify no-op behaviour when ``prometheus_client`` is not installed.
- Verify ``record_call`` increments counters and records histogram.
- Verify ``record_rate_limit_hit`` and ``record_validation_failure``
  counters.
- Verify ``start_server`` launches a background daemon thread.
- Verify end-to-end metric recording through ``_run_tool``.
 
Diagnostics
-----------
Domain: MCP-TESTS
Levels:
    L2 — test lifecycle
    L3 — assertion details
    L4 — deep tracing
 
Contracts
---------
- Tests mock ``prometheus_client`` imports; no real metrics server is
  started.
"""
 
from __future__ import annotations
 
import json
from pathlib import Path
from unittest.mock import MagicMock, patch, call
 
import pytest
 
from oct.mcp.metrics import McpMetrics, PROMETHEUS_AVAILABLE
 
 
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
 
def _is_prometheus_installed() -> bool:
    try:
        import prometheus_client  # noqa: F401
        return True
    except ImportError:
        return False
 
 
def _isolated_metrics() -> McpMetrics:
    """McpMetrics on a private CollectorRegistry — avoids
    ``Duplicated timeseries`` collisions across tests when
    ``prometheus_client`` is installed."""
    if not PROMETHEUS_AVAILABLE:
        return McpMetrics()
    from prometheus_client import CollectorRegistry
    return McpMetrics(registry=CollectorRegistry())
 
 
# ---------------------------------------------------------------------------
# Module-level constant
# ---------------------------------------------------------------------------
 
class TestPrometheusAvailableFlag:
    def test_prometheus_available_is_bool(self):
        assert isinstance(PROMETHEUS_AVAILABLE, bool)
 
    def test_prometheus_available_matches_import(self):
        assert PROMETHEUS_AVAILABLE == _is_prometheus_installed()
 
 
# ---------------------------------------------------------------------------
# No-op behavior when prometheus_client is absent
# ---------------------------------------------------------------------------
 
class TestMetricsNoOpWhenAbsent:
    """When prometheus_client is not installed, all methods should be no-ops."""
 
    def _make_metrics_without_prometheus(self) -> McpMetrics:
        """Return a McpMetrics instance pretending prometheus_client is absent."""
        with patch("oct.mcp.metrics.PROMETHEUS_AVAILABLE", False):
            m = McpMetrics.__new__(McpMetrics)
            m.__init__()
        return m
 
    def test_no_op_record_call_does_not_raise(self):
        with patch("oct.mcp.metrics.PROMETHEUS_AVAILABLE", False):
            metrics = McpMetrics()
            metrics.record_call("oct_lint", "allowed", 0.05, 0)  # must not raise
 
    def test_no_op_record_rate_limit_hit_does_not_raise(self):
        with patch("oct.mcp.metrics.PROMETHEUS_AVAILABLE", False):
            metrics = McpMetrics()
            metrics.record_rate_limit_hit("oct_lint")
 
    def test_no_op_record_validation_failure_does_not_raise(self):
        with patch("oct.mcp.metrics.PROMETHEUS_AVAILABLE", False):
            metrics = McpMetrics()
            metrics.record_validation_failure("oct_lint")
 
    def test_no_op_start_server_does_not_raise(self):
        with patch("oct.mcp.metrics.PROMETHEUS_AVAILABLE", False):
            metrics = McpMetrics()
            metrics.start_server(9090)
 
 
# ---------------------------------------------------------------------------
# Prometheus-backed behavior
# ---------------------------------------------------------------------------
 
class TestMetricsWithPrometheus:
    """Tests using mock Prometheus Counter / Histogram objects."""
 
    def _make_metrics(self):
        """Build McpMetrics with mocked prometheus_client objects."""
        mock_counter_class = MagicMock()
        mock_histogram_class = MagicMock()
 
        # Counter() returns a mock with a .labels() chain
        mock_counter_instance = MagicMock()
        mock_counter_instance.labels.return_value = mock_counter_instance
        mock_counter_class.return_value = mock_counter_instance
 
        mock_histogram_instance = MagicMock()
        mock_histogram_instance.labels.return_value = mock_histogram_instance
        mock_histogram_class.return_value = mock_histogram_instance
 
        with patch("oct.mcp.metrics.PROMETHEUS_AVAILABLE", True):
            with patch("oct.mcp.metrics.Counter", mock_counter_class):
                with patch("oct.mcp.metrics.Histogram", mock_histogram_class):
                    metrics = McpMetrics()
 
        return metrics, mock_counter_instance, mock_histogram_instance
 
    @pytest.mark.skipif(not _is_prometheus_installed(), reason="prometheus_client not installed")
    def test_record_call_increments_tool_calls_counter(self):
        metrics = _isolated_metrics()
        # The counter should not raise; call it
        metrics.record_call("oct_lint", "allowed", 0.1, 0)
 
    @pytest.mark.skipif(not _is_prometheus_installed(), reason="prometheus_client not installed")
    def test_record_call_observes_histogram(self):
        metrics = _isolated_metrics()
        metrics.record_call("oct_lint", "allowed", 0.25, 0)
 
    @pytest.mark.skipif(not _is_prometheus_installed(), reason="prometheus_client not installed")
    def test_record_rate_limit_hit(self):
        metrics = _isolated_metrics()
        metrics.record_rate_limit_hit("oct_lint")
 
    @pytest.mark.skipif(not _is_prometheus_installed(), reason="prometheus_client not installed")
    def test_record_validation_failure(self):
        metrics = _isolated_metrics()
        metrics.record_validation_failure("oct_lint")
 
    def test_start_server_starts_daemon_thread(self):
        if not _is_prometheus_installed():
            pytest.skip("prometheus_client not installed")
 
        started_threads = []
 
        class MockThread:
            def __init__(self, target, args, daemon, kwargs=None):
                self.target = target
                self.args = args
                self.kwargs = kwargs or {}
                self.daemon = daemon
                self._started = False
 
            def start(self):
                self._started = True
                started_threads.append(self)
 
        with patch("oct.mcp.metrics.PROMETHEUS_AVAILABLE", True):
            with patch("oct.mcp.metrics.threading.Thread", MockThread):
                with patch("oct.mcp.metrics._start_http"):
                    metrics = _isolated_metrics()
                    metrics.start_server(9090)
 
        assert len(started_threads) == 1
        assert started_threads[0].daemon is True
        # Confirm the registry kwarg is propagated to start_http_server.
        assert "registry" in started_threads[0].kwargs
        assert started_threads[0].kwargs["registry"] is metrics._registry
 
    def test_start_server_noop_when_absent(self):
        """start_server with PROMETHEUS_AVAILABLE=False must not start any thread."""
        import threading
        threads_before = threading.active_count()
        with patch("oct.mcp.metrics.PROMETHEUS_AVAILABLE", False):
            metrics = McpMetrics()
            metrics.start_server(9090)
        threads_after = threading.active_count()
        assert threads_after == threads_before
 
    def test_record_call_with_mock_counter_labels_called(self):
        """Verify record_call calls .labels(tool=..., decision=...) on counter."""
        mock_counter = MagicMock()
        mock_histogram = MagicMock()
 
        # Bypass __init__ and inject mock attributes directly so we don't
        # need prometheus_client installed at the module level.
        with patch("oct.mcp.metrics.PROMETHEUS_AVAILABLE", True):
            metrics = McpMetrics.__new__(McpMetrics)
            metrics._tool_calls = mock_counter
            metrics._tool_duration = mock_histogram
            metrics._redactions = MagicMock()
            metrics._rate_limit_hits = MagicMock()
            metrics._validation_failures = MagicMock()
            metrics.record_call("oct_lint", "allowed", 0.1, 0)
 
        mock_counter.labels.assert_called_with(tool="oct_lint", decision="allowed")
 
    def test_record_call_increments_redactions_when_positive(self):
        """When redactions > 0, _redactions counter must be incremented."""
        mock_redact_counter = MagicMock()
 
        with patch("oct.mcp.metrics.PROMETHEUS_AVAILABLE", True):
            metrics = McpMetrics.__new__(McpMetrics)
            # Bypass __init__; manually set attributes
            metrics._tool_calls = MagicMock()
            metrics._tool_duration = MagicMock()
            metrics._redactions = mock_redact_counter
            metrics._rate_limit_hits = MagicMock()
            metrics._validation_failures = MagicMock()
            metrics.record_call("oct_lint", "allowed", 0.1, 3)
 
        mock_redact_counter.labels.assert_called_with(tool="oct_lint")
        mock_redact_counter.labels.return_value.inc.assert_called_with(3)
 
    def test_record_call_does_not_increment_redactions_when_zero(self):
        """When redactions == 0, _redactions counter must NOT be incremented."""
        mock_redact_counter = MagicMock()
 
        with patch("oct.mcp.metrics.PROMETHEUS_AVAILABLE", True):
            metrics = McpMetrics.__new__(McpMetrics)
            metrics._tool_calls = MagicMock()
            metrics._tool_duration = MagicMock()
            metrics._redactions = mock_redact_counter
            metrics._rate_limit_hits = MagicMock()
            metrics._validation_failures = MagicMock()
            metrics.record_call("oct_lint", "allowed", 0.1, 0)
 
        mock_redact_counter.labels.assert_not_called()
 
 
# ---------------------------------------------------------------------------
# E2E integration with _run_tool
# ---------------------------------------------------------------------------
 
class TestMetricsE2E:
    """Verify that _run_tool() calls the right metrics methods."""
 
    def _make_state(self, tmp_path: Path, mock_metrics: McpMetrics):
        from oct.mcp.audit import McpAuditLogger
        from oct.mcp.config import McpConfig
        from oct.mcp.executor import ToolExecutor
        from oct.mcp.policy import McpPolicy
        from oct.mcp.redactor import Redactor
        from oct.mcp.safety import SafetyGate
        from oct.mcp.sandbox import SandboxExecutor, SandboxResult
        from oct.mcp.server import RateLimiter, ServerState
 
        (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=0, stdout="ok", stderr=""
        )
        config = McpConfig()
        return ServerState(
            project_root=tmp_path,
            config=config,
            policy=McpPolicy(profile="default", safe_mode=False),
            executor=ToolExecutor(sandbox=mock_sandbox),
            redactor=Redactor(),
            audit_logger=McpAuditLogger(audit_log_path=tmp_path / "audit.log"),
            rate_limiter=RateLimiter(limit_per_minute=30),
            safety=SafetyGate(config),
            metrics=mock_metrics,
            session_id="metrics-test-session",
        )
 
    def test_run_tool_calls_record_call(self, tmp_path: Path):
        from oct.mcp.tools import _run_tool
        import oct.mcp.server as _server_module
 
        mock_metrics = MagicMock(spec=McpMetrics)
        state = self._make_state(tmp_path, mock_metrics)
        _server_module._server_state = state
        _run_tool("oct_lint", {})
        mock_metrics.record_call.assert_called_once()
 
    def test_run_tool_rate_limited_calls_record_rate_limit_hit(self, tmp_path: Path):
        from oct.mcp.server import RateLimiter
        from oct.mcp.tools import _run_tool
        import oct.mcp.server as _server_module
 
        mock_metrics = MagicMock(spec=McpMetrics)
        state = self._make_state(tmp_path, mock_metrics)
        # Replace rate limiter with one that's already exhausted
        exhausted_limiter = MagicMock(spec=RateLimiter)
        exhausted_limiter.consume.return_value = False
        state.rate_limiter = exhausted_limiter
        _server_module._server_state = state
        _run_tool("oct_lint", {})
        mock_metrics.record_rate_limit_hit.assert_called_once_with("oct_lint")
 
    def test_run_tool_validation_failure_calls_record_validation_failure(
        self, tmp_path: Path
    ):
        from oct.mcp.tools import _run_tool
        import oct.mcp.server as _server_module
 
        mock_metrics = MagicMock(spec=McpMetrics)
        state = self._make_state(tmp_path, mock_metrics)
        _server_module._server_state = state
        # Path traversal triggers validation failure
        _run_tool("oct_lint", {"paths": ["../../etc/passwd"]})
        mock_metrics.record_validation_failure.assert_called_once_with("oct_lint")