pr-reviewer/tests/test_handbook.py

"""Tests for app.handbook: git checkout, page reading, and the digest provider.

All network and LLM calls are mocked (subprocess.run and fireworks_complete).
"""

from __future__ import annotations

import base64
import json
from pathlib import Path

import pytest

from app import handbook as hb
from app.handbook import (
    HandbookError,
    HandbookProvider,
    _extract_checklist,
    ensure_checkout,
    read_pages,
)


class FakeCompleted:
    def __init__(self, returncode: int = 0, stdout: str = "", stderr: str = "") -> None:
        self.returncode = returncode
        self.stdout = stdout
        self.stderr = stderr


def _fail(*a, **k):
    raise AssertionError("should not have been called")


# --- ensure_checkout -------------------------------------------------------- #


def test_ensure_checkout_clones_when_absent(monkeypatch, tmp_path) -> None:
    cache = tmp_path / "hb"
    calls: list[list[str]] = []

    def fake_run(cmd, **kw):
        calls.append(cmd)
        return FakeCompleted(0, stdout="abcdef123\n" if "rev-parse" in cmd else "")

    monkeypatch.setattr(hb.subprocess, "run", fake_run)
    head = ensure_checkout(cache, "https://x/hb.git", token="tok")

    assert head == "abcdef123"
    clone = next(c for c in calls if "clone" in c)
    # token goes via a Basic-auth http.extraHeader, never embedded in the URL
    expected = (
        "Authorization: Basic " + base64.b64encode(b"x-access-token:tok").decode()
    )
    assert any(expected in p for p in clone)
    assert all("tok@" not in p for p in clone)


def test_ensure_checkout_pulls_when_present(monkeypatch, tmp_path) -> None:
    cache = tmp_path / "hb"
    (cache / ".git").mkdir(parents=True)
    calls: list[list[str]] = []

    def fake_run(cmd, **kw):
        calls.append(cmd)
        return FakeCompleted(0, stdout="deadbeef\n" if "rev-parse" in cmd else "")

    monkeypatch.setattr(hb.subprocess, "run", fake_run)
    head = ensure_checkout(cache, "https://x/hb.git", token=None)

    assert head == "deadbeef"
    assert any("fetch" in c for c in calls)
    assert any("reset" in c for c in calls)
    assert not any("clone" in c for c in calls)


def test_ensure_checkout_raises_on_failure(monkeypatch, tmp_path) -> None:
    monkeypatch.setattr(
        hb.subprocess, "run", lambda cmd, **kw: FakeCompleted(1, stderr="boom")
    )
    with pytest.raises(HandbookError):
        ensure_checkout(tmp_path / "hb", "https://x/hb.git", token=None)


# --- _extract_checklist ----------------------------------------------------- #


def test_extract_checklist_strips_reasoning_preamble() -> None:
    raw = "We need to produce...\nreasoning here\n<CHECKLIST>\nNAMING\n- kebab-case\n</CHECKLIST>"
    assert _extract_checklist(raw) == "NAMING\n- kebab-case"


def test_extract_checklist_falls_back_without_tags() -> None:
    assert _extract_checklist("  NAMING\n- kebab-case  ") == "NAMING\n- kebab-case"


# --- read_pages ------------------------------------------------------------- #


def test_read_pages_concats_present_files(tmp_path) -> None:
    (tmp_path / "naming-conventions.md").write_text("kebab-case rules")
    (tmp_path / "code-review.md").write_text("review rubric")
    text = read_pages(tmp_path)
    assert "kebab-case rules" in text
    assert "review rubric" in text
    assert "naming-conventions.md" in text  # header included


def test_read_pages_raises_when_none_found(tmp_path) -> None:
    with pytest.raises(HandbookError):
        read_pages(tmp_path)


def test_read_pages_truncates(monkeypatch, tmp_path) -> None:
    monkeypatch.setattr(hb, "DISTILL_MAX_BYTES", 50)
    (tmp_path / "code-review.md").write_text("X" * 500)
    assert "[handbook truncated]" in read_pages(tmp_path)


# --- HandbookProvider ------------------------------------------------------- #


def _provider(tmp_path, make_cfg, **over):
    cfg = make_cfg(
        HANDBOOK_CACHE_DIR=str(tmp_path / "hb"),
        HANDBOOK_DIGEST_PATH=str(tmp_path / "digest.json"),
        **over,
    )
    return HandbookProvider(cfg), cfg


def _stub_pipeline(monkeypatch, digest="DIGEST", head="sha123"):
    monkeypatch.setattr(hb, "ensure_checkout", lambda *a, **k: head)
    monkeypatch.setattr(hb, "read_pages", lambda *a, **k: "PAGES")
    monkeypatch.setattr(hb, "fireworks_complete", lambda *a, **k: digest)


def test_refresh_distills_when_stale(monkeypatch, tmp_path, make_cfg) -> None:
    prov, cfg = _provider(tmp_path, make_cfg)
    _stub_pipeline(monkeypatch, digest="DIGEST OUTPUT", head="sha123")

    assert prov.current_digest() is None
    assert prov.refresh_if_stale() is True
    assert prov.current_digest() == "DIGEST OUTPUT"
    assert prov.status()["head_sha"] == "sha123"
    assert prov.status()["digest_chars"] == len("DIGEST OUTPUT")
    saved = json.loads(Path(cfg.HANDBOOK_DIGEST_PATH).read_text())
    assert saved["digest"] == "DIGEST OUTPUT"


def test_refresh_skips_when_fresh(monkeypatch, tmp_path, make_cfg) -> None:
    prov, _ = _provider(tmp_path, make_cfg)
    n = {"calls": 0}

    monkeypatch.setattr(hb, "ensure_checkout", lambda *a, **k: "s")
    monkeypatch.setattr(hb, "read_pages", lambda *a, **k: "P")

    def counting(*a, **k):
        n["calls"] += 1
        return "D"

    monkeypatch.setattr(hb, "fireworks_complete", counting)

    prov.refresh_if_stale()
    prov.refresh_if_stale()  # still fresh -> no second distill
    assert n["calls"] == 1


def test_refresh_disabled_is_noop(monkeypatch, tmp_path, make_cfg) -> None:
    prov, _ = _provider(tmp_path, make_cfg, HANDBOOK_ENABLED=False)
    monkeypatch.setattr(hb, "ensure_checkout", _fail)
    assert prov.refresh_if_stale() is False


def test_refresh_keeps_last_good_on_error(monkeypatch, tmp_path, make_cfg) -> None:
    prov, _ = _provider(tmp_path, make_cfg)
    _stub_pipeline(monkeypatch, digest="GOOD")
    prov.refresh_if_stale()
    assert prov.current_digest() == "GOOD"

    # Force stale + clear backoff, then make the pull fail.
    prov._distilled_at = 0.0
    prov._last_attempt = 0.0
    monkeypatch.setattr(hb, "ensure_checkout", _fail_hb)
    assert prov.refresh_if_stale() is False
    assert prov.current_digest() == "GOOD"  # last good retained


def test_refresh_backs_off_after_failure(monkeypatch, tmp_path, make_cfg) -> None:
    prov, _ = _provider(tmp_path, make_cfg)
    attempts = {"n": 0}

    def boom(*a, **k):
        attempts["n"] += 1
        raise HandbookError("net down")

    monkeypatch.setattr(hb, "ensure_checkout", boom)
    prov.refresh_if_stale()  # attempt 1 fails
    prov.refresh_if_stale()  # within backoff window -> not attempted again
    assert attempts["n"] == 1


def test_load_cache_on_init(tmp_path, make_cfg) -> None:
    p = tmp_path / "digest.json"
    p.write_text(
        json.dumps(
            {"digest": "CACHED", "head_sha": "h1", "distilled_at": 9_999_999_999}
        )
    )
    cfg = make_cfg(HANDBOOK_CACHE_DIR=str(tmp_path / "hb"), HANDBOOK_DIGEST_PATH=str(p))
    prov = HandbookProvider(cfg)
    assert prov.current_digest() == "CACHED"
    assert prov.status()["head_sha"] == "h1"


def _fail_hb(*a, **k):
    raise HandbookError("net down")