llm-ingestion-okf/tests/test_corpus_run.py

"""The corpus harness: it reports numbers with denominators, or it fails.

Three properties, all of them reactions to measured defects rather than good
intentions:

- **K1b is a COMMAND, not a sentence in a report.** The conservation identity
  `merged + Sigma(coded rejections) == N` is checked by the harness, which
  EXITS NON-ZERO when it does not hold. A prose assertion is something a reader
  has to trust; an exit status is something a pipeline cannot ignore.
- **`N` is computed, never typed.** A literal `43` keeps passing after the
  corpus changes, and the number it then reports is a fact about a directory
  that no longer exists.
- **Three counts, never one.** The guard sits between extraction and persist,
  so a healthy persisted count can hide a pile of quarantines. Extracted,
  gated and persisted are reported separately for that reason.

The negative control is the point of this module. A harness that can never
fail proves nothing, so `test_a_file_neither_merged_nor_coded_fails_the_run`
hands the conservation check an inventory it must reject -- if that test ever
passes silently, every green run above it becomes meaningless.
"""

from __future__ import annotations

import sys
from pathlib import Path

import pytest

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

import okf_corpus_run  # noqa: E402

INGESTED_AT = "2026-07-25T12:00:00Z"

SUBSTANTIVE = "Krav til seksjonering av bygget.\n\nEn andre setning som baerer innhold.\n"
DEGENERATE = "   \n\t\n   \n"


def corpus(tmp_path: Path, files: dict[str, str]) -> Path:
    root = tmp_path / "corpus"
    root.mkdir(parents=True, exist_ok=True)
    for name, text in files.items():
        (root / name).write_text(text, encoding="utf-8", newline="")
    return root


def test_the_denominator_is_the_directory_not_a_literal(tmp_path: Path) -> None:
    root = corpus(tmp_path, {"a.md": SUBSTANTIVE, "b.md": SUBSTANTIVE, "c.md": SUBSTANTIVE})
    report = okf_corpus_run.measure(root, tmp_path / "bundle", ingested_at=INGESTED_AT)
    assert report.n == 3
    assert report.n == len(list(root.iterdir()))


def test_the_conservation_identity_holds_and_the_run_exits_zero(tmp_path: Path) -> None:
    root = corpus(
        tmp_path,
        {"a.md": SUBSTANTIVE, "b.md": DEGENERATE, "c.md": SUBSTANTIVE, "d.bin": "x"},
    )
    code = okf_corpus_run.main(
        ["--corpus", str(root), "--report", str(tmp_path / "r.md"), "--ingested-at", INGESTED_AT]
    )
    assert code == 0
    report = okf_corpus_run.measure(root, tmp_path / "bundle2", ingested_at=INGESTED_AT)
    assert report.merged + report.rejected == report.n
    assert not report.unaccounted


def test_a_file_neither_merged_nor_coded_fails_the_run(tmp_path: Path) -> None:
    """The negative control. Without it a green run proves nothing.

    The check is handed an inventory where one name is in neither column --
    exactly what a silently dropped file looks like from the outside -- and it
    must both refuse and NAME the file.
    """
    unaccounted = okf_corpus_run.unaccounted_names(
        dropped=("a.md", "b.md", "vanished.md"),
        merged=("a.md",),
        coded=("b.md",),
    )
    assert unaccounted == ("vanished.md",)
    assert okf_corpus_run.unaccounted_names(dropped=("a.md",), merged=("a.md",), coded=()) == ()


def test_the_harness_exits_non_zero_and_names_the_unaccounted_file(
    tmp_path: Path, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch
) -> None:
    root = corpus(tmp_path, {"a.md": SUBSTANTIVE, "b.md": SUBSTANTIVE})
    real = okf_corpus_run.measure

    def lose_one(*args: object, **kwargs: object):
        report = real(*args, **kwargs)  # type: ignore[arg-type]
        return okf_corpus_run.replace(report, unaccounted=("b.md",))

    monkeypatch.setattr(okf_corpus_run, "measure", lose_one)
    code = okf_corpus_run.main(
        ["--corpus", str(root), "--report", str(tmp_path / "r.md"), "--ingested-at", INGESTED_AT]
    )
    assert code != 0
    assert "b.md" in capsys.readouterr().err


def test_the_degenerate_rule_is_reproducible_from_its_statement(tmp_path: Path) -> None:
    """Zero characters after stripping whitespace. A definition, not a threshold."""
    assert okf_corpus_run.is_degenerate("")
    assert okf_corpus_run.is_degenerate("   \n\t  ")
    assert not okf_corpus_run.is_degenerate("x")
    root = corpus(tmp_path, {"a.md": SUBSTANTIVE, "b.md": DEGENERATE})
    report = okf_corpus_run.measure(root, tmp_path / "bundle", ingested_at=INGESTED_AT)
    assert report.substantive == 1
    assert report.degenerate == 1
    assert report.substantive + report.degenerate == report.merged


def test_the_three_counts_are_reported_separately(tmp_path: Path) -> None:
    """A healthy persisted count can hide a pile of quarantines."""
    root = corpus(tmp_path, {"a.md": SUBSTANTIVE, "b.bin": "x"})
    report = okf_corpus_run.measure(root, tmp_path / "bundle", ingested_at=INGESTED_AT)
    assert (report.extracted, report.gated, report.persisted) == (1, 1, 1)
    assert report.n == 2
    text = report.render()
    for label in ("extracted", "gated", "persisted", "denominator"):
        assert label in text


def test_the_report_names_the_resolved_converter_and_its_version(tmp_path: Path) -> None:
    """The vendored binary is bypassed silently otherwise -- measured three times."""
    root = corpus(tmp_path, {"a.md": SUBSTANTIVE})
    report = okf_corpus_run.measure(root, tmp_path / "bundle", ingested_at=INGESTED_AT)
    text = report.render()
    assert "converter" in text.lower()
    assert okf_corpus_run.converter_identity()[1] in text


def test_wall_time_per_file_is_reported(tmp_path: Path) -> None:
    """The only evidence the scale requirement will ever have."""
    root = corpus(tmp_path, {"a.md": SUBSTANTIVE, "b.md": SUBSTANTIVE})
    report = okf_corpus_run.measure(root, tmp_path / "bundle", ingested_at=INGESTED_AT)
    assert report.seconds_total >= 0.0
    assert "per file" in report.render()