ops-warden/tests/test_audit.py

"""Tests for unified audit trail (WARDEN-WP-0022)."""
from __future__ import annotations

import json
from datetime import datetime, timedelta, timezone
from pathlib import Path
from unittest.mock import patch

import pytest
from typer.testing import CliRunner

from warden.audit import (
    AuditError,
    collect_activity,
    fetch_hub_notes,
    read_events,
    record_event,
)
from warden.cli import app

runner = CliRunner()


def test_record_and_read_event(tmp_path: Path) -> None:
    record_event(
        tmp_path,
        kind="sign",
        action="issue",
        subject="agt-test",
        target="agt-test",
        decision_id="dec-1",
        backend="local",
    )
    events = read_events(tmp_path)
    assert len(events) == 1
    assert events[0]["kind"] == "sign"
    assert events[0]["subject"] == "agt-test"
    assert events[0]["decision_id"] == "dec-1"


def test_read_events_filters_by_kind_and_since(tmp_path: Path) -> None:
    record_event(tmp_path, kind="sign", action="issue", subject="a", target="a")
    record_event(tmp_path, kind="access", action="fetch", subject="op", target="need-1")
    since = datetime.now(timezone.utc) - timedelta(hours=1)
    sign_only = read_events(tmp_path, since=since, kinds={"sign"})
    assert len(sign_only) == 1
    assert sign_only[0]["kind"] == "sign"


def test_secret_guard_rejects_token_prefix(tmp_path: Path) -> None:
    with pytest.raises(AuditError, match="secret"):
        record_event(
            tmp_path,
            kind="access",
            action="fetch",
            subject="ghp_abc123456789012345678901234567890",
            target="need",
        )


def test_secret_guard_rejects_high_entropy(tmp_path: Path) -> None:
    with pytest.raises(AuditError, match="high-entropy"):
        record_event(
            tmp_path,
            kind="access",
            action="fetch",
            subject="operator",
            target="need",
            note="9f3a8c2d1b0e7f6a5c4d3b2a1f0e9d8c7b6a5948372615049382716059483",
        )


def test_rotation_when_log_exceeds_limit(tmp_path: Path, monkeypatch) -> None:
    import warden.audit as audit_mod

    monkeypatch.setattr(audit_mod, "_MAX_BYTES", 50)
    for i in range(5):
        record_event(tmp_path, kind="worker", action="tick", subject="worker", target=str(i))
    assert (tmp_path / "audit.jsonl").exists()
    assert (tmp_path / "audit.jsonl.1").exists()


def test_collect_activity_merges_legacy_logs(tmp_path: Path) -> None:
    ts = datetime.now(timezone.utc).isoformat()
    (tmp_path / "signatures.log").write_text(
        json.dumps(
            {
                "timestamp": ts,
                "actor": "agt-legacy",
                "actor_type": "agt",
                "backend": "vault",
            }
        )
        + "\n"
    )
    (tmp_path / "access-audit.log").write_text(
        json.dumps(
            {
                "timestamp": ts,
                "action": "fetch",
                "need_id": "openbao-api-key",
                "owner_repo": "railiance-platform",
                "subject": "operator",
                "exit_code": 0,
            }
        )
        + "\n"
    )
    events = collect_activity(tmp_path, days=7)
    kinds = {e["kind"] for e in events}
    assert "sign" in kinds
    assert "access" in kinds
    assert any(e.get("source") == "signatures.log" for e in events)


def test_fetch_hub_notes_filters_ops_warden(tmp_path: Path) -> None:
    payload = [
        {
            "created_at": datetime.now(timezone.utc).isoformat(),
            "summary": "ops-warden: worker tick complete",
            "author": "codex",
            "event_type": "note",
        },
        {
            "created_at": datetime.now(timezone.utc).isoformat(),
            "summary": "unrelated repo change",
            "author": "codex",
            "event_type": "note",
        },
    ]
    with patch("httpx.get") as mock_get:
        mock_get.return_value.raise_for_status = lambda: None
        mock_get.return_value.json.return_value = payload
        notes = fetch_hub_notes(days=7, hub_url="http://127.0.0.1:8000")
    assert len(notes) == 1
    assert notes[0]["kind"] == "hub"


def test_activity_cli_json(tmp_path: Path, monkeypatch) -> None:
    state_dir = tmp_path / "state"
    state_dir.mkdir()
    cfg = tmp_path / "warden.yaml"
    cfg.write_text(f"backend: local\nca_key: {tmp_path / 'ca'}\nstate_dir: {state_dir}\n")
    (tmp_path / "ca").write_text("fake")
    monkeypatch.setenv("WARDEN_CONFIG", str(cfg))
    record_event(state_dir, kind="sign", action="issue", subject="agt-cli", target="agt-cli")

    result = runner.invoke(app, ["activity", "--days", "1", "--json"])
    assert result.exit_code == 0
    data = json.loads(result.stdout)
    assert isinstance(data, list)
    assert data[0]["kind"] == "sign"