llm-ingestion-okf/tests/test_okf_watch.py

"""The upstream OKF watch: it must be able to prove it can find.

This watch reports a NEGATIVE result almost every time it runs. That is the
whole hazard: a run that never reached the network, or whose query silently
matched nothing, produces the same "no change" as a run that genuinely found
nothing. Trust in an absence that was never measured is worse than no watch,
so every test here is about keeping those two outcomes distinguishable.

The git-backed tests build their own repository in tmp_path. Skipping when
`~/repos/_okf-canonical` is missing would preserve nothing on the machine
where it exists, and pointing them at the real mirror would make the suite
depend on upstream's history.
"""

from __future__ import annotations

import json
import subprocess
import sys
from pathlib import Path

import pytest

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "tools"))

import okf_watch

# --- the exit-status / empty-result seam -----------------------------------


def _git(repo: Path, *args: str) -> str:
    return subprocess.run(
        ["git", *args], cwd=repo, capture_output=True, text=True, check=True
    ).stdout


@pytest.fixture
def repo(tmp_path: Path) -> Path:
    """A repository with one quiet commit and one normative commit."""
    r = tmp_path / "upstream"
    r.mkdir()
    _git(r, "init", "-q", "-b", "main")
    _git(r, "config", "user.email", "t@example.invalid")
    _git(r, "config", "user.name", "t")
    (r / "SPEC.md").write_text("Version 0.2\n", encoding="utf-8")
    (r / "CONTRIBUTING.md").write_text("hello\n", encoding="utf-8")
    _git(r, "add", "-A")
    _git(r, "commit", "-q", "-m", "base")
    (r / "CONTRIBUTING.md").write_text("hello again\n", encoding="utf-8")
    _git(r, "add", "-A")
    _git(r, "commit", "-q", "-m", "docs: contributing")
    (r / "SPEC.md").write_text("Version 0.2 tightened\n", encoding="utf-8")
    _git(r, "add", "-A")
    _git(r, "commit", "-q", "-m", "spec: tighten timestamps")
    return r


def test_a_failing_git_call_raises_instead_of_returning_empty(repo: Path) -> None:
    """The defect this watch exists to avoid, at its smallest.

    `git log <nonexistent>` writes to stderr and exits non-zero. A helper that
    returned its empty stdout would hand the caller "no commits" — a negative
    manufactured by a broken query.
    """
    with pytest.raises(okf_watch.WatchError) as exc:
        okf_watch.git(repo, "log", "--format=%H", "no-such-ref..HEAD")
    assert "exit" in str(exc.value)


def test_an_empty_result_from_a_successful_call_is_not_an_error(repo: Path) -> None:
    """The other half of the seam: a real zero must still be readable as zero."""
    head = okf_watch.git(repo, "rev-parse", "HEAD")
    assert okf_watch.git(repo, "log", "--format=%H", f"{head}..{head}") == ""


# --- classification --------------------------------------------------------


def test_a_spec_change_is_normative() -> None:
    normative, quiet = okf_watch.classify(["SPEC.md", "CONTRIBUTING.md"])
    assert normative == ["SPEC.md"]
    assert quiet == ["CONTRIBUTING.md"]


def test_the_reference_reader_is_normative_because_is_stale_reversed_there() -> None:
    normative, _ = okf_watch.classify(["src/reference_agent/bundle/document.py"])
    assert normative == ["src/reference_agent/bundle/document.py"]


def test_the_viewer_is_quiet_but_the_readme_is_not() -> None:
    """README carries the relocation notice. That is how the move was announced."""
    normative, quiet = okf_watch.classify(
        ["src/reference_agent/viewer/static/viz.css", "bundles/acme_retail/viz.html", "README.md"]
    )
    assert normative == ["README.md"]
    assert quiet == ["src/reference_agent/viewer/static/viz.css", "bundles/acme_retail/viz.html"]


def test_an_unknown_path_defaults_to_normative() -> None:
    """Quiet is an enumerated list; signal is not.

    A watch that enumerates what matters cannot see a change upstream has not
    invented yet, and fails in the silent direction. Over-firing is visible;
    under-firing is not.
    """
    normative, quiet = okf_watch.classify(["some/new/thing-upstream-adds-later.md"])
    assert normative == ["some/new/thing-upstream-adds-later.md"]
    assert quiet == []


# --- proving it can find ---------------------------------------------------


def test_the_self_check_passes_on_a_range_that_really_changed_the_spec(repo: Path) -> None:
    base = okf_watch.git(repo, "rev-parse", "HEAD~1")
    head = okf_watch.git(repo, "rev-parse", "HEAD")
    okf_watch.prove_can_find(repo, base, head)  # must not raise


def test_the_self_check_fails_loudly_when_the_query_finds_nothing(repo: Path) -> None:
    """An empty known-positive means the query is broken, not that upstream is quiet."""
    head = okf_watch.git(repo, "rev-parse", "HEAD")
    with pytest.raises(okf_watch.WatchError) as exc:
        okf_watch.prove_can_find(repo, head, head)
    assert "known-positive" in str(exc.value)


def test_the_self_check_fails_loudly_when_the_range_is_unreachable(repo: Path) -> None:
    with pytest.raises(okf_watch.WatchError):
        okf_watch.prove_can_find(repo, "deadbeef", "HEAD")


# --- idempotence: report on change, not on state ---------------------------


def test_a_first_normative_change_is_reported(tmp_path: Path) -> None:
    state = okf_watch.State.load(tmp_path / "s.json", pin="PIN")
    assert okf_watch.decide(state, ["aaa", "bbb"]) == ["aaa", "bbb"]


def test_the_same_upstream_state_is_not_reported_twice(tmp_path: Path) -> None:
    path = tmp_path / "s.json"
    state = okf_watch.State.load(path, pin="PIN")
    okf_watch.decide(state, ["aaa"])
    state.record(["aaa"])
    state.save(path)
    again = okf_watch.State.load(path, pin="PIN")
    assert okf_watch.decide(again, ["aaa"]) == []


def test_only_the_new_commits_are_reported(tmp_path: Path) -> None:
    """Upstream advancing must not re-announce what was already announced."""
    path = tmp_path / "s.json"
    state = okf_watch.State.load(path, pin="PIN")
    state.record(["aaa"])
    state.save(path)
    later = okf_watch.State.load(path, pin="PIN")
    assert okf_watch.decide(later, ["aaa", "bbb"]) == ["bbb"]


def test_moving_our_pin_resets_what_counts_as_already_reported(tmp_path: Path) -> None:
    """A pin move means we absorbed everything behind it. Nothing before it is news."""
    path = tmp_path / "s.json"
    state = okf_watch.State.load(path, pin="OLDPIN")
    state.record(["aaa"])
    state.save(path)
    assert json.loads(path.read_text(encoding="utf-8"))["pin"] == "OLDPIN"
    after = okf_watch.State.load(path, pin="NEWPIN")
    assert after.reported == []


# --- what gets sent --------------------------------------------------------


def test_the_coord_body_is_ascii_because_coord_send_requires_it() -> None:
    body = okf_watch.render_body(
        pin="ad30107",
        head="beef123",
        commits=[("beef123", "spec: tighten timestamps")],
        normative=["SPEC.md"],
        quiet=["CONTRIBUTING.md"],
    )
    body.encode("ascii")  # raises UnicodeEncodeError if not


def test_the_body_names_the_repository_it_watched() -> None:
    """ "No change" is only meaningful with the object named. So is a hit."""
    body = okf_watch.render_body(
        pin="ad30107", head="beef123", commits=[("beef123", "x")], normative=["SPEC.md"], quiet=[]
    )
    assert okf_watch.CANONICAL_REPO in body
    assert "ad30107" in body and "beef123" in body


def test_the_body_carries_the_denominator_not_just_the_hits() -> None:
    body = okf_watch.render_body(
        pin="p",
        head="h",
        commits=[("a", "one"), ("b", "two")],
        normative=["SPEC.md"],
        quiet=["CONTRIBUTING.md", "LICENSE.md"],
    )
    assert "1 of 3" in body


# --- the override that fired a false alarm ---------------------------------


def test_an_overridden_pin_cannot_send_a_real_message(tmp_path: Path, repo: Path) -> None:
    """Measured 2026-08-23: this exact combination delivered two false alarms.

    `--pin` exists so the reporting branch can be exercised on demand against
    real upstream history. Exercising it must never be indistinguishable from
    upstream actually moving, so the override forces dry-run rather than
    trusting the operator to pair two flags correctly.
    """
    args = okf_watch.parse_args(["--pin", "deadbeef", "--state", str(tmp_path / "s.json")])
    assert args.dry_run is True


def test_the_real_pin_leaves_dry_run_alone() -> None:
    assert okf_watch.parse_args([]).dry_run is False