rdb-discovery/tests/test_guardrails.py

from pathlib import Path
from datetime import datetime, timedelta

import os
import pytest

from rdb_discovery.guardrails import (
    check_repeated_reads,
    check_repeated_commands,
    check_repeated_commands_telemetry,
    check_repeated_reads_telemetry,
    check_no_recent_file_changes,
    check_no_test_run_recorded,
    check_run_log_updated,
    run_all_guardrails,
    format_report,
)


# -- check_repeated_reads --

def test_check_repeated_reads_ok(tmp_path: Path) -> None:
    """No agent history files → status ok."""
    result = check_repeated_reads(tmp_path)
    assert result["status"] == "ok"


def test_check_repeated_reads_warning(tmp_path: Path) -> None:
    """AGENT_HANDOFF.md mentions the same file >2 times."""
    ah = tmp_path / "AGENT_HANDOFF.md"
    ah.write_text(
        "# Agent Handoff\n\n"
        "- README.md\n- README.md\n- README.md\n"
        "- TASKS.md\n",
        encoding="utf-8",
    )
    result = check_repeated_reads(tmp_path)
    assert result["status"] == "warning"
    assert "README.md (3x)" in result["details"]


def test_check_repeated_reads_no_over_threshold(tmp_path: Path) -> None:
    """Files mentioned <=2 times should not trigger warning."""
    ah = tmp_path / "AGENT_HANDOFF.md"
    ah.write_text(
        "# Agent Handoff\n\n"
        "- README.md\n- README.md\n"
        "- TASKS.md\n- TASKS.md\n",
        encoding="utf-8",
    )
    result = check_repeated_reads(tmp_path)
    assert result["status"] == "ok"


# -- check_repeated_commands --

def test_check_repeated_commands_ok_no_log(tmp_path: Path) -> None:
    result = check_repeated_commands(tmp_path)
    assert result["status"] == "ok"


def test_check_repeated_commands_ok(tmp_path: Path) -> None:
    rl = tmp_path / "RUN_LOG.md"
    rl.write_text(
        "| date | event1 |\n"
        "| date | event2 |\n",
        encoding="utf-8",
    )
    result = check_repeated_commands(tmp_path)
    assert result["status"] == "ok"


def test_check_repeated_commands_warning(tmp_path: Path) -> None:
    rl = tmp_path / "RUN_LOG.md"
    line = "| date | Task completed |\n"
    rl.write_text(
        "# RUN_LOG\n\n| Date | Event | Task | Notes |\n" + line * 5,
        encoding="utf-8",
    )
    result = check_repeated_commands(tmp_path)
    assert result["status"] == "warning"


# -- check_repeated_commands_telemetry --

def test_check_repeated_commands_telemetry_no_events(tmp_path: Path) -> None:
    """No telemetry events → status ok with fallback message."""
    result = check_repeated_commands_telemetry(tmp_path)
    assert result["status"] == "ok"
    assert "No telemetry data available" in result["details"]


def test_check_repeated_commands_telemetry_warning(tmp_path: Path) -> None:
    """Same command repeated >3 times in telemetry → warning."""
    from rdb_discovery.telemetry import record_event

    for _ in range(5):
        record_event(tmp_path, "command", "rdb prompt")
    result = check_repeated_commands_telemetry(tmp_path)
    assert result["status"] == "warning"
    assert "rdb prompt" in result["details"]


def test_check_repeated_commands_telemetry_no_over_threshold(tmp_path: Path) -> None:
    """Same command repeated <=3 times → ok."""
    from rdb_discovery.telemetry import record_event

    for _ in range(3):
        record_event(tmp_path, "command", "rdb prompt")
    result = check_repeated_commands_telemetry(tmp_path)
    assert result["status"] == "ok"


# -- check_repeated_reads_telemetry --

def test_check_repeated_reads_telemetry_no_events(tmp_path: Path) -> None:
    """No telemetry events → status ok with fallback message."""
    result = check_repeated_reads_telemetry(tmp_path)
    assert result["status"] == "ok"
    assert "No telemetry data available" in result["details"]


def test_check_repeated_reads_telemetry_no_read_events(tmp_path: Path) -> None:
    """Only command events, no read events → ok with message."""
    from rdb_discovery.telemetry import record_event

    record_event(tmp_path, "command", "rdb prompt")
    result = check_repeated_reads_telemetry(tmp_path)
    assert result["status"] == "ok"
    assert "No file read events found" in result["details"]


def test_check_repeated_reads_telemetry_warning(tmp_path: Path) -> None:
    """Same file read >3 times in telemetry → warning."""
    from rdb_discovery.telemetry import record_event

    for _ in range(5):
        record_event(tmp_path, "read", "README.md")
    result = check_repeated_reads_telemetry(tmp_path)
    assert result["status"] == "warning"
    assert "README.md" in result["details"]


def test_check_repeated_reads_telemetry_no_over_threshold(tmp_path: Path) -> None:
    """Same file read <=3 times → ok."""
    from rdb_discovery.telemetry import record_event

    for _ in range(3):
        record_event(tmp_path, "read", "TASKS.md")
    result = check_repeated_reads_telemetry(tmp_path)
    assert result["status"] == "ok"


# -- check_no_recent_file_changes --

def test_check_no_recent_file_changes_ok(tmp_path: Path) -> None:
    """Files modified recently → ok."""
    for rel in ["TASKS.md", "PROJECT_STATE.md"]:
        (tmp_path / rel).write_text("x", encoding="utf-8")

    # Patch _last_modified to return a recent time
    import rdb_discovery.guardrails as mod
    original_now = mod._now

    def fake_now() -> datetime:
        return datetime.now()

    mod._now = fake_now
    try:
        result = check_no_recent_file_changes(tmp_path)
        assert result["status"] == "ok"
    finally:
        mod._now = original_now


def test_check_no_recent_file_changes_warning(tmp_path: Path) -> None:
    """Files modified >48h ago → warning."""
    (tmp_path / "TASKS.md").write_text("x", encoding="utf-8")

    import rdb_discovery.guardrails as mod

    old_ts = datetime.now() - timedelta(hours=72)
    # Set a fake mtime via os.utime (avoids pathlib.touch times= on macOS)
    ts = old_ts.timestamp()
    (tmp_path / "TASKS.md").touch()
    os.utime(str(tmp_path / "TASKS.md"), (ts, ts))

    result = check_no_recent_file_changes(tmp_path)
    assert result["status"] == "warning"


# -- check_no_test_run_recorded --

def test_check_no_test_run_recorded_ok(tmp_path: Path) -> None:
    rl = tmp_path / "RUN_LOG.md"
    rl.write_text("# RUN\n\n| date | pytest passed |\n", encoding="utf-8")
    result = check_no_test_run_recorded(tmp_path)
    assert result["status"] == "ok"


def test_check_no_test_run_recorded_warning(tmp_path: Path) -> None:
    rl = tmp_path / "RUN_LOG.md"
    rl.write_text("# RUN\n\n| date | task done |\n", encoding="utf-8")
    result = check_no_test_run_recorded(tmp_path)
    assert result["status"] == "warning"


def test_check_no_test_run_recorded_no_log(tmp_path: Path) -> None:
    result = check_no_test_run_recorded(tmp_path)
    assert result["status"] == "ok"


# -- check_run_log_updated --

def test_check_run_log_updated_ok(tmp_path: Path) -> None:
    tasks = tmp_path / "TASKS.md"
    tasks.write_text(
        "## TASK-001 - First\nStatus: Done\n",
        encoding="utf-8",
    )
    rl = tmp_path / "RUN_LOG.md"
    rl.write_text(
        "# RUN\n\n| date | Task completed | TASK-001 |\n",
        encoding="utf-8",
    )
    result = check_run_log_updated(tmp_path)
    assert result["status"] == "ok"


def test_check_run_log_updated_warning(tmp_path: Path) -> None:
    tasks = tmp_path / "TASKS.md"
    tasks.write_text(
        "## TASK-001 - First\nStatus: Done\n",
        encoding="utf-8",
    )
    rl = tmp_path / "RUN_LOG.md"
    rl.write_text("# RUN\n\n| date | something else |\n", encoding="utf-8")
    result = check_run_log_updated(tmp_path)
    assert result["status"] == "warning"


def test_check_run_log_updated_no_files(tmp_path: Path) -> None:
    result = check_run_log_updated(tmp_path)
    assert result["status"] == "ok"


# -- run_all_guardrails --

def test_run_all_guardrails_returns_keys() -> None:
    import tempfile
    with tempfile.TemporaryDirectory() as td:
        root = Path(td)
        # Create minimal files so checks don't return 'ok' for missing file
        (root / "RUN_LOG.md").write_text("# RUN\n", encoding="utf-8")
        result = run_all_guardrails(root)
        assert "checks" in result
        assert "overall_status" in result


def test_run_all_guardrails_overall_all_clear(tmp_path: Path) -> None:
    """All checks pass → overall 'all clear'."""
    # Create files with recent mtimes and ok content
    (tmp_path / "TASKS.md").write_text("# TASKS\n", encoding="utf-8")
    (tmp_path / "RUN_LOG.md").write_text("# RUN\n| date | pytest passed |\n", encoding="utf-8")

    import rdb_discovery.guardrails as mod

    old_now = mod._now
    now = datetime.now()

    class FakePath:
        def __init__(self, path: Path):
            self._path = path

        def exists(self) -> bool:
            return self._path.exists()

        def stat(self):
            class StatResult:
                st_mtime = now.timestamp()
            return StatResult()

    orig_last_modified = mod._last_modified

    def fake_last_modified(root: Path, rel: str) -> datetime | None:
        p = root / rel
        if p.exists():
            return now
        return None

    mod._now = lambda: now
    mod._last_modified = fake_last_modified

    try:
        result = run_all_guardrails(tmp_path)
        # Should be clear or have a non-error overall status (guardrails may flag other things)
        assert result["overall_status"] in ("all clear", "review recommended")
    finally:
        mod._now = old_now
        mod._last_modified = orig_last_modified


# -- format_report --

def test_format_report_includes_overall() -> None:
    results = {
        "checks": {"Test check": {"status": "ok", "details": "fine"}},
        "overall_status": "all clear",
    }
    report = format_report(results)
    assert "# Guardrail Report" in report
    assert "all clear" in report


def test_format_report_with_warning() -> None:
    results = {
        "checks": {"Test check": {"status": "warning", "details": "watch out"}},
        "overall_status": "review recommended",
    }
    report = format_report(results)
    assert "review recommended" in report