test_audit.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# tests/tests_git/test_audit.py
 
"""
Purpose
-------
Unit tests for :mod:`oct.git.audit` — the rotating JSONL audit logger
and the :func:`audited` decorator for Phase 4B ``oct git`` commands
(G-B5 / G-B6).
 
Responsibilities
----------------
- Verify :class:`AuditRecord` serialises to JSONL as one-object-per-line.
- Verify the size-rotation threshold is honoured: the very first write
  that lifts a file at/above :data:`AUDIT_MAX_SIZE_BYTES` goes into a
  fresh file.
- Verify the count-rotation threshold: no more than
  :data:`AUDIT_MAX_FILES` files live in ``logs/`` at once.
- Verify credential redaction applies to every ``args`` entry.
- Verify :func:`audited` writes a row on both success and exception,
  never swallows exceptions from the wrapped function, and records
  ``duration_ms`` monotonically.
 
Diagnostics
-----------
Domain: GIT-TESTS
Levels:
    L2 — test lifecycle
    L3 — assertion details
    L4 — deep tracing
 
Contracts
---------
- No real git subprocess, no real Click group — the decorator is tested
  against a minimal SimpleNamespace stand-in for ``ctx``.
- Every test uses a ``tmp_path`` subdirectory as its project root so
  the real ``logs/`` directory is never touched.
"""
 
from __future__ import annotations
 
import json
from pathlib import Path
from types import SimpleNamespace
from unittest import mock
 
import pytest
 
from oct.git import audit as audit_mod
from oct.git.audit import (
    AUDIT_MAX_FILES,
    AUDIT_MAX_SIZE_BYTES,
    AuditLogger,
    AuditRecord,
    _redact_args,
    _resolve_project_root_from_ctx,
    audited,
)
 
 
# ---------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------
 
 
@pytest.fixture
def project_root(tmp_path: Path) -> Path:
    """A throwaway project root. The logs dir is NOT pre-created so we
    can assert the logger materialises it on demand."""
    root = tmp_path / "proj"
    root.mkdir()
    return root
 
 
@pytest.fixture
def make_record():
    """Factory returning a minimal populated :class:`AuditRecord`."""
 
    def _make(**overrides) -> AuditRecord:
        base = dict(
            timestamp="2026-04-11T12:00:00Z",
            command="oct git check",
            args=["oct", "git", "check", "--staged-only"],
            branch="feature/x",
            staged_files=3,
            checks_passed=True,
            secrets_detected=False,
            lint_violations=0,
            format_violations=0,
            exit_code=0,
            duration_ms=120,
        )
        base.update(overrides)
        return AuditRecord(**base)
 
    return _make
 
 
# ---------------------------------------------------------------------
# _redact_args
# ---------------------------------------------------------------------
 
 
class TestRedactArgs:
    def test_empty_returns_empty_list(self):
        assert _redact_args(None) == []
        assert _redact_args([]) == []
        assert _redact_args(()) == []
 
    def test_plain_args_pass_through(self):
        assert _redact_args(["oct", "git", "check"]) == ["oct", "git", "check"]
 
    def test_http_url_with_token_is_redacted(self):
        argv = ["git", "clone", "https://user:t0ken@example.com/x.git"]
        out = _redact_args(argv)
        assert out[0] == "git"
        assert out[1] == "clone"
        assert "t0ken" not in out[2]
        assert "***@example.com" in out[2]
 
    def test_non_string_entries_stringified(self):
        out = _redact_args([1, None, "ok"])
        assert out == ["1", "None", "ok"]
 
 
# ---------------------------------------------------------------------
# AuditRecord <-> JSON
# ---------------------------------------------------------------------
 
 
class TestAuditRecordSerialisation:
    def test_asdict_roundtrip(self, make_record):
        rec = make_record()
        serialised = json.dumps(rec.__dict__)  # cheap proxy for asdict
        loaded = json.loads(serialised)
        assert loaded["command"] == "oct git check"
        assert loaded["exit_code"] == 0
 
    def test_defaults_are_json_safe(self):
        rec = AuditRecord()
        # Must not raise.
        json.dumps({
            "timestamp": rec.timestamp,
            "command": rec.command,
            "args": rec.args,
            "branch": rec.branch,
            "staged_files": rec.staged_files,
            "checks_passed": rec.checks_passed,
            "secrets_detected": rec.secrets_detected,
            "lint_violations": rec.lint_violations,
            "format_violations": rec.format_violations,
            "exit_code": rec.exit_code,
            "duration_ms": rec.duration_ms,
        })
 
 
# ---------------------------------------------------------------------
# AuditLogger — basic write
# ---------------------------------------------------------------------
 
 
class TestAuditLoggerWrite:
    def test_write_creates_logs_dir(self, project_root, make_record):
        assert not (project_root / "logs").exists()
        logger = AuditLogger(project_root)
        logger.write(make_record())
        assert (project_root / "logs").is_dir()
 
    def test_write_creates_one_jsonl_file(self, project_root, make_record):
        logger = AuditLogger(project_root)
        logger.write(make_record())
        files = list((project_root / "logs").glob("git-audit-*.jsonl"))
        assert len(files) == 1
 
    def test_write_appends_one_line_per_record(self, project_root, make_record):
        logger = AuditLogger(project_root)
        logger.write(make_record(exit_code=0))
        logger.write(make_record(exit_code=1))
        logger.write(make_record(exit_code=5))
        files = list((project_root / "logs").glob("git-audit-*.jsonl"))
        assert len(files) == 1
        lines = files[0].read_text(encoding="utf-8").strip().splitlines()
        assert len(lines) == 3
        for line in lines:
            parsed = json.loads(line)
            assert "exit_code" in parsed
 
    def test_write_redacts_args_defensively(self, project_root, make_record):
        logger = AuditLogger(project_root)
        rec = make_record(args=["https://user:tok@example.com/r.git"])
        logger.write(rec)
        files = list((project_root / "logs").glob("git-audit-*.jsonl"))
        content = files[0].read_text(encoding="utf-8")
        assert "tok" not in content
        assert "***@example.com" in content
 
    def test_write_never_raises_on_io_error(self, project_root, make_record):
        logger = AuditLogger(project_root)
        # Break the logs dir by pointing it at something unwriteable-ish:
        # swap the Path for a file rather than a directory.
        bogus = project_root / "logs"
        bogus.write_text("not a dir")
        # Must not raise.
        logger.write(make_record())
 
    def test_current_path_none_until_first_write(self, project_root, make_record):
        logger = AuditLogger(project_root)
        assert logger.current_path is None
        logger.write(make_record())
        assert logger.current_path is not None
        assert logger.current_path.name.startswith("git-audit-")
        assert logger.current_path.name.endswith(".jsonl")
 
 
# ---------------------------------------------------------------------
# AuditLogger — size rotation
# ---------------------------------------------------------------------
 
 
class TestSizeRotation:
    def test_rotation_opens_new_file_when_size_exceeds_threshold(
        self, project_root, make_record, monkeypatch
    ):
        # Shrink the threshold so rotation is cheap to trigger.
        monkeypatch.setattr(audit_mod, "AUDIT_MAX_SIZE_BYTES", 200)
        logger = AuditLogger(project_root)
        # Two records below threshold, then keep writing until we roll over.
        logger.write(make_record(command="tiny"))
        first_path = logger.current_path
        assert first_path is not None
 
        # Stuff the file so the next write triggers rotation.
        with first_path.open("a", encoding="utf-8") as fh:
            fh.write("x" * 300)
 
        logger.write(make_record(command="after-rotation"))
        second_path = logger.current_path
        assert second_path is not None
        assert second_path != first_path
 
        files = sorted((project_root / "logs").glob("git-audit-*.jsonl"))
        assert len(files) == 2
 
        # The "after-rotation" record must be in the NEW file.
        new_content = second_path.read_text(encoding="utf-8")
        assert "after-rotation" in new_content
        old_content = first_path.read_text(encoding="utf-8")
        assert "after-rotation" not in old_content
 
    def test_rotation_threshold_default_is_5mb(self):
        # Regression: blueprint §12.2 pins the 5 MB default.
        assert AUDIT_MAX_SIZE_BYTES == 5 * 1024 * 1024
 
 
# ---------------------------------------------------------------------
# AuditLogger — count rotation
# ---------------------------------------------------------------------
 
 
class TestCountRotation:
    def test_max_files_default_is_twenty(self):
        assert AUDIT_MAX_FILES == 20
 
    def test_old_files_are_pruned_when_count_exceeds_max(
        self, project_root, make_record, monkeypatch
    ):
        # Seed logs/ with 25 fake audit files, each with a unique timestamp.
        logs = project_root / "logs"
        logs.mkdir()
        seeded = []
        for i in range(25):
            # Lexical order == chronological order by construction.
            name = f"git-audit-20260101-{i:06d}.jsonl"
            p = logs / name
            p.write_text(f"seed {i}\n", encoding="utf-8")
            seeded.append(p)
 
        # Shrink size threshold so the first real write creates a new file
        # and triggers pruning without us having to write 5 MB of text.
        monkeypatch.setattr(audit_mod, "AUDIT_MAX_SIZE_BYTES", 50)
        logger = AuditLogger(project_root)
        logger.write(make_record(command="newest"))
 
        remaining = sorted(logs.glob("git-audit-*.jsonl"))
        # After writing one new file + pruning down to the limit.
        assert len(remaining) == AUDIT_MAX_FILES
 
        # The newest file (the one just written) must still be there.
        assert logger.current_path in remaining
 
        # The oldest of the 25 seeds must be gone.
        assert seeded[0] not in remaining
 
    def test_prune_is_no_op_when_under_limit(
        self, project_root, make_record, monkeypatch
    ):
        monkeypatch.setattr(audit_mod, "AUDIT_MAX_SIZE_BYTES", 50)
        logger = AuditLogger(project_root)
        for _ in range(5):
            logger.write(make_record())
        files = list((project_root / "logs").glob("git-audit-*.jsonl"))
        # 5 writes under a 50-byte threshold rotate aggressively but must
        # never prune because we stay under AUDIT_MAX_FILES.
        assert 1 <= len(files) <= AUDIT_MAX_FILES
 
 
# ---------------------------------------------------------------------
# _resolve_project_root_from_ctx
# ---------------------------------------------------------------------
 
 
class TestResolveProjectRoot:
    def test_reads_project_root_key(self, tmp_path):
        ctx = SimpleNamespace(obj={"project_root": tmp_path})
        assert _resolve_project_root_from_ctx(ctx) == tmp_path
 
    def test_reads_root_key_fallback(self, tmp_path):
        ctx = SimpleNamespace(obj={"root": tmp_path})
        assert _resolve_project_root_from_ctx(ctx) == tmp_path
 
    def test_none_ctx_obj_falls_through(self, tmp_path, monkeypatch):
        ctx = SimpleNamespace(obj=None)
        monkeypatch.chdir(tmp_path)
        result = _resolve_project_root_from_ctx(ctx)
        # Either get_project_root returns something, or we fall to cwd.
        assert isinstance(result, Path)
 
 
# ---------------------------------------------------------------------
# @audited decorator
# ---------------------------------------------------------------------
 
 
class TestAuditedDecorator:
    def test_writes_audit_row_on_success(self, project_root, make_record):
        @audited("oct git test-cmd")
        def fn(ctx):
            ctx.obj["_audit_record"] = make_record(command="oct git test-cmd")
            return "ok"
 
        ctx = SimpleNamespace(obj={"project_root": project_root})
        result = fn(ctx)
        assert result == "ok"
        files = list((project_root / "logs").glob("git-audit-*.jsonl"))
        assert len(files) == 1
        content = files[0].read_text(encoding="utf-8")
        assert "oct git test-cmd" in content
 
    def test_writes_audit_row_on_exception(self, project_root):
        @audited("oct git crash-cmd")
        def fn(ctx):
            # Command crashes before populating the record.
            raise RuntimeError("boom")
 
        ctx = SimpleNamespace(obj={"project_root": project_root})
        with pytest.raises(RuntimeError, match="boom"):
            fn(ctx)
        files = list((project_root / "logs").glob("git-audit-*.jsonl"))
        assert len(files) == 1
        content = files[0].read_text(encoding="utf-8")
        loaded = json.loads(content.strip().splitlines()[0])
        assert loaded["command"] == "oct git crash-cmd"
        assert loaded["exit_code"] is None
        assert loaded["checks_passed"] is False
 
    def test_duration_ms_is_recorded(self, project_root, make_record):
        @audited("oct git time-cmd")
        def fn(ctx):
            ctx.obj["_audit_record"] = make_record(command="oct git time-cmd")
            return 0
 
        ctx = SimpleNamespace(obj={"project_root": project_root})
        fn(ctx)
        files = list((project_root / "logs").glob("git-audit-*.jsonl"))
        loaded = json.loads(files[0].read_text(encoding="utf-8").strip())
        # Duration is non-negative and an int.
        assert isinstance(loaded["duration_ms"], int)
        assert loaded["duration_ms"] >= 0
 
    def test_creates_ctx_obj_if_missing(self, project_root):
        @audited("oct git no-obj")
        def fn(ctx):
            ctx.obj["_audit_record"] = AuditRecord(
                timestamp="2026-04-11T00:00:00Z",
                command="oct git no-obj",
                exit_code=0,
            )
            return None
 
        ctx = SimpleNamespace(obj=None, project_root=project_root)
        # Also cover fallback resolution via ctx.obj setdefault
        ctx.obj = {"project_root": project_root}  # seed so resolver works
        fn(ctx)
 
    def test_ctx_obj_audit_record_never_none_after_call(
        self, project_root, make_record
    ):
        @audited("oct git populate")
        def fn(ctx):
            ctx.obj["_audit_record"] = make_record(command="oct git populate")
            return "fine"
 
        ctx = SimpleNamespace(obj={"project_root": project_root})
        fn(ctx)
        assert ctx.obj["_audit_record"] is not None
        assert ctx.obj["_audit_record"].command == "oct git populate"