test_mcp_audit.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# tests/tests_mcp/test_mcp_audit.py
 
"""
Purpose
-------
Regression tests for :mod:`oct.mcp.audit` — ``McpAuditRecord`` and
``McpAuditLogger``.
 
Responsibilities
----------------
- Verify JSONL write and rotation (size and count thresholds).
- Verify pruning of old audit files.
- Verify session/request ID propagation and redaction flag.
 
Diagnostics
-----------
Domain: MCP-TESTS
Levels:
    L2 — test lifecycle
    L3 — assertion details
    L4 — deep tracing
 
Contracts
---------
- All audit files are written to ``tmp_path``; no real logs directory
  is touched.
"""
 
from __future__ import annotations
 
import json
import time
from pathlib import Path
 
import pytest
 
from oct.mcp.audit import (
    MCP_AUDIT_MAX_FILES,
    MCP_AUDIT_MAX_SIZE_BYTES,
    McpAuditLogger,
    McpAuditRecord,
    _hash_project_root,
    _utc_now_iso,
)
 
 
# -------------------------------------------------------------------
# McpAuditRecord
# -------------------------------------------------------------------
 
class TestMcpAuditRecord:
    def test_defaults(self):
        r = McpAuditRecord()
        assert r.timestamp == ""
        assert r.session_id == ""
        assert r.request_id == ""
        assert r.tool == ""
        assert r.args == {}
        assert r.args_redacted is True
        assert r.decision == ""
        assert r.policy_rule == ""
        assert r.exit_code is None
        assert r.duration_ms == 0
        assert r.output_size_bytes == 0
        assert r.output_truncated is False
        assert r.output_hash_sha256 == ""
        assert r.redactions_applied == 0
        assert r.user == ""
        assert r.project_root_hash == ""
 
    def test_all_fields_json_serialisable(self):
        r = McpAuditRecord(
            timestamp="2026-01-01T00:00:00Z",
            session_id="s1",
            request_id="r1",
            tool="oct_lint",
            args={"paths": []},
            decision="allowed",
            policy_rule="manifest_auto_approve",
            exit_code=0,
            duration_ms=42,
            output_size_bytes=100,
            redactions_applied=3,
            user="alice",
            project_root_hash="abcd1234",
        )
        from dataclasses import asdict
        line = json.dumps(asdict(r))
        assert "oct_lint" in line
 
 
# -------------------------------------------------------------------
# Helpers
# -------------------------------------------------------------------
 
class TestHelpers:
    def test_utc_now_iso_format(self):
        ts = _utc_now_iso()
        assert ts.endswith("Z")
        assert "T" in ts
 
    def test_hash_project_root_length(self, tmp_path: Path):
        h = _hash_project_root(tmp_path)
        assert len(h) == 8
 
    def test_hash_project_root_hex(self, tmp_path: Path):
        h = _hash_project_root(tmp_path)
        int(h, 16)  # must be valid hex
 
    def test_hash_project_root_stable(self, tmp_path: Path):
        h1 = _hash_project_root(tmp_path)
        h2 = _hash_project_root(tmp_path)
        assert h1 == h2
 
    def test_hash_different_roots_differ(self, tmp_path: Path):
        a = tmp_path / "a"
        b = tmp_path / "b"
        a.mkdir()
        b.mkdir()
        assert _hash_project_root(a) != _hash_project_root(b)
 
 
# -------------------------------------------------------------------
# McpAuditLogger — basic write
# -------------------------------------------------------------------
 
class TestMcpAuditLoggerWrite:
    def test_creates_audit_dir(self, tmp_path: Path):
        audit_dir = tmp_path / "audit_dir"
        logger = McpAuditLogger(audit_log_path=audit_dir / "audit.log")
        r = McpAuditRecord(tool="oct_lint", timestamp=_utc_now_iso())
        logger.write(r)
        assert audit_dir.exists()
 
    def test_creates_audit_log_file(self, tmp_path: Path):
        log = tmp_path / "audit.log"
        logger = McpAuditLogger(audit_log_path=log)
        logger.write(McpAuditRecord(tool="oct_lint", timestamp=_utc_now_iso()))
        assert log.exists()
 
    def test_writes_jsonl_line(self, tmp_path: Path):
        log = tmp_path / "audit.log"
        logger = McpAuditLogger(audit_log_path=log)
        logger.write(McpAuditRecord(tool="oct_health", timestamp=_utc_now_iso()))
        lines = log.read_text(encoding="utf-8").strip().splitlines()
        assert len(lines) == 1
        record = json.loads(lines[0])
        assert record["tool"] == "oct_health"
 
    def test_multiple_writes_multiple_lines(self, tmp_path: Path):
        log = tmp_path / "audit.log"
        logger = McpAuditLogger(audit_log_path=log)
        for i in range(5):
            logger.write(McpAuditRecord(tool=f"oct_lint_{i}", timestamp=_utc_now_iso()))
        lines = log.read_text(encoding="utf-8").strip().splitlines()
        assert len(lines) == 5
 
    def test_session_id_preserved(self, tmp_path: Path):
        log = tmp_path / "audit.log"
        logger = McpAuditLogger(audit_log_path=log)
        logger.write(McpAuditRecord(
            tool="oct_lint",
            session_id="sess-abc",
            request_id="req-xyz",
            timestamp=_utc_now_iso(),
        ))
        record = json.loads(log.read_text(encoding="utf-8").strip())
        assert record["session_id"] == "sess-abc"
        assert record["request_id"] == "req-xyz"
 
    def test_args_redacted_flag(self, tmp_path: Path):
        log = tmp_path / "audit.log"
        logger = McpAuditLogger(audit_log_path=log)
        logger.write(McpAuditRecord(
            tool="oct_lint",
            args={"paths": ["src/"]},
            args_redacted=True,
            timestamp=_utc_now_iso(),
        ))
        record = json.loads(log.read_text(encoding="utf-8").strip())
        assert record["args_redacted"] is True
 
    def test_timestamp_auto_filled_when_empty(self, tmp_path: Path):
        log = tmp_path / "audit.log"
        logger = McpAuditLogger(audit_log_path=log)
        # timestamp is empty — logger should fill it.
        logger.write(McpAuditRecord(tool="oct_lint"))
        record = json.loads(log.read_text(encoding="utf-8").strip())
        assert record["timestamp"].endswith("Z")
 
    def test_current_path_set_after_write(self, tmp_path: Path):
        log = tmp_path / "audit.log"
        logger = McpAuditLogger(audit_log_path=log)
        assert logger.current_path is None
        logger.write(McpAuditRecord(tool="oct_lint", timestamp=_utc_now_iso()))
        assert logger.current_path is not None
 
    def test_never_raises_on_bad_dir(self):
        # Writing to an invalid path should not raise.
        logger = McpAuditLogger(audit_log_path="/\x00/invalid/audit.log")
        # Should silently fail, not raise.
        logger.write(McpAuditRecord(tool="oct_lint", timestamp=_utc_now_iso()))
 
 
# -------------------------------------------------------------------
# McpAuditLogger — rotation
# -------------------------------------------------------------------
 
class TestMcpAuditLoggerRotation:
    def test_rotation_creates_new_file(self, tmp_path: Path):
        log = tmp_path / "audit.log"
        # Very small rotation threshold to trigger rotation easily.
        logger = McpAuditLogger(
            audit_log_path=log,
            max_size_bytes=10,  # tiny threshold
        )
        logger.write(McpAuditRecord(tool="oct_lint", timestamp=_utc_now_iso()))
        path_after_first = logger.current_path
        # Write again — should rotate since file > 10 bytes.
        logger.write(McpAuditRecord(tool="oct_health", timestamp=_utc_now_iso()))
        path_after_second = logger.current_path
        assert path_after_second != path_after_first
 
    def test_rotated_files_use_mcp_audit_prefix(self, tmp_path: Path):
        log = tmp_path / "audit.log"
        logger = McpAuditLogger(
            audit_log_path=log,
            max_size_bytes=5,
        )
        for _ in range(3):
            logger.write(McpAuditRecord(tool="oct_lint", timestamp=_utc_now_iso()))
        # At least one rotated file should have the mcp-audit-* prefix.
        rotated = list(tmp_path.glob("mcp-audit-*.jsonl"))
        assert len(rotated) >= 1
 
 
# -------------------------------------------------------------------
# McpAuditLogger — pruning
# -------------------------------------------------------------------
 
class TestMcpAuditLoggerPruning:
    def test_pruning_keeps_max_files(self, tmp_path: Path):
        log = tmp_path / "audit.log"
        max_files = 5
        logger = McpAuditLogger(
            audit_log_path=log,
            max_size_bytes=5,  # rotate every write
            max_files=max_files,
        )
        # Write enough to create max_files + 5 rotated files.
        for i in range(max_files + 5):
            logger.write(McpAuditRecord(tool="oct_lint", timestamp=_utc_now_iso()))
            time.sleep(0.001)  # ensure distinct timestamps
 
        rotated = list(tmp_path.glob("mcp-audit-*.jsonl"))
        assert len(rotated) <= max_files
 
    def test_audit_dir_property(self, tmp_path: Path):
        log = tmp_path / "subdir" / "audit.log"
        logger = McpAuditLogger(audit_log_path=log)
        assert logger.audit_dir == tmp_path / "subdir"