"""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\nNAMING\n- kebab-case\n" 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")