llm-ingestion-okf/tests/test_accounting_gate.py

"""The content-accounting gate's own tests (`tools/okf_accounting_gate.py`).

The gate was written RED at `0b00de4`, before `okf build` had a source
inventory or a per-element account. These tests are GREEN and prove three things.

1. The fasit is independent. `tools/okf_witness.py` imports no
   `llm_ingestion_okf` module -- checked on the live import graph of a process
   that ran every witness, with a control that shows the check fires -- and
   the committed inventories are exactly what the witness counts today, pinned
   again to hand counts on four documents.
2. Every row CAN turn green and CAN turn red, each boundary driven from both
   sides with synthetic build output (the door the capability must open).
3. Run against the real `okf build`, the fixture rows are green. They were
   red at `0b00de4` on the three defects the gate was ordered for (no
   inventory, a file carried AND rejected, a rejected document logged as "0
   carried of 0 found"); `test_a_file_carried_through_a_document_and_rejected_
   is_double_booked` and the row 4 tests keep those defects detectable.
"""

from __future__ import annotations

import json
import subprocess
import sys
from pathlib import Path
from typing import Any

import pytest

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

import okf_accounting_gate as gate  # noqa: E402
import okf_witness as witness  # noqa: E402

TABLE = [
    ".csv",
    ".docx",
    ".htm",
    ".html",
    ".json",
    ".md",
    ".odt",
    ".pdf",
    ".pptx",
    ".rtf",
    ".txt",
    ".xlsx",
    ".xml",
]


# --- 1. the fasit ------------------------------------------------------------

_IMPORT_PROBE = """
import sys
sys.path.insert(0, {tools!r})
{preload}
import okf_witness as w
from pathlib import Path
fixtures = Path({fixtures!r})
w.witness_inbox(fixtures / "corpus")
w.witness_inbox(fixtures / "rejected")
w.count_sts_json((fixtures / "witness" / "prosess-84-sts.twin.json").read_bytes())
w.pdf_poppler(fixtures / "corpus" / "prosess-84-tabell.pdf")
print(sorted(m for m in sys.modules if m.split(".")[0] == "llm_ingestion_okf"))
"""


def _package_modules_after_witness(preload: str) -> list[str]:
    script = _IMPORT_PROBE.format(tools=str(TOOLS), fixtures=str(gate.FIXTURES), preload=preload)
    out = subprocess.run(
        [sys.executable, "-c", script], capture_output=True, text=True, check=True
    ).stdout
    result: list[str] = json.loads(out.strip().splitlines()[-1].replace("'", '"'))
    return result


def test_the_witness_imports_no_module_of_the_package_it_judges() -> None:
    pytest.importorskip("pdfplumber")
    assert _package_modules_after_witness("") == []


def test_the_import_check_fires_when_the_package_is_loaded() -> None:
    pytest.importorskip("pdfplumber")
    assert "llm_ingestion_okf" in _package_modules_after_witness("import llm_ingestion_okf")


@pytest.mark.parametrize(
    ("inbox", "committed"),
    [(gate.CORPUS, gate.INVENTORY), (gate.REJECTED, gate.REJECTED_INVENTORY)],
)
def test_the_committed_fasit_is_what_the_witness_counts(inbox: Path, committed: Path) -> None:
    pytest.importorskip("pdfplumber")
    assert witness.witness_inbox(inbox) == gate.load_inventory(committed)


def test_the_fixture_corpus_covers_every_readme_file_type() -> None:
    inventory = gate.load_inventory(gate.INVENTORY)
    assert gate.readme_types() == TABLE
    assert {entry["suffix"] for entry in inventory["documents"].values()} == set(TABLE)


# Counted by hand from the fixture bytes, not by the witness.
HAND_COUNTS = {
    "notat.md": {
        "code_block": 1,
        "heading": 2,
        "image": 1,
        "paragraph": 2,
        "table": 1,
        "table_row": 2,
    },
    "side.htm": {"cell": 4, "heading": 2, "image": 0, "list_item": 2, "paragraph": 1, "table": 1},
    "krav-rikt-tekstformat.rtf": {"cell": 56, "image": 0, "paragraph": 3, "table_row": 24},
    "prosess-84-sts.xml": {
        "cell": 0,
        "footnote": 0,
        "image": 2,
        "list_item": 0,
        "paragraph": 2,
        "section": 2,
        "section_label": 2,
        "table": 0,
        "table_label": 0,
        "title": 2,
    },
}


@pytest.mark.parametrize("name", sorted(HAND_COUNTS))
def test_the_witness_matches_a_hand_count(name: str) -> None:
    inventory = witness.witness_file(gate.CORPUS, gate.CORPUS / name)
    assert inventory.elements == HAND_COUNTS[name]


def test_a_fenced_heading_is_not_a_heading_to_the_witness() -> None:
    elements, _ = witness.count_markdown("# Real\n\n```bash\n# not one\n```\n")
    assert elements["heading"] == 1
    assert elements["code_block"] == 1


def test_the_sts_image_reference_resolves_through_the_graphics_directory() -> None:
    inventory = witness.witness_file(gate.CORPUS, gate.STS_FIXTURE)
    assert [ref.target for ref in inventory.images] == [
        "graphics/tabell-84-2.png",
        "graphics/figur-84-1.png",
    ]


def test_a_reference_above_the_document_resolves_to_nothing(tmp_path: Path) -> None:
    (tmp_path / "secret.png").write_bytes(b"x")
    (tmp_path / "docs").mkdir()
    document = tmp_path / "docs" / "a.html"
    assert witness.resolve_local(tmp_path, document, "../secret.png") is None


def test_the_witness_refuses_a_doctype() -> None:
    with pytest.raises(witness.WitnessRefused):
        witness.count_sts_xml(b'<!DOCTYPE x [<!ENTITY a "b">]><standard/>')


# --- 2. every row can go both ways -------------------------------------------


def _inventory() -> dict[str, Any]:
    return {
        "documents": {
            "a.md": {"suffix": ".md", "elements": {"heading": 2, "image": 1}, "images": []},
        },
        "files": {"graphics/x.png": {"pointed_at_by": ["a.md"]}},
    }


def _build(
    *,
    accounting: dict[str, Any] | None = None,
    sources: set[str] | None = None,
    assets: set[str] | None = None,
    log: str = "",
    exit_code: int = 0,
) -> gate.Build:
    return gate.Build(
        exit_code=exit_code,
        log=log,
        accounting=accounting,
        source_files={"a.md"} if sources is None else sources,
        asset_prefixes=assets or set(),
    )


def _declared(heading: int = 2, image: int = 1, fate: str = "rejected") -> dict[str, Any]:
    return {
        "accounting_version": 1,
        "documents": [
            {
                "source_file": "a.md",
                "status": "persisted",
                "code": None,
                "inventory": {"heading": 2, "image": 1},
                "fates": {
                    "heading": {"carried": heading},
                    "image": {"pointer": image},
                },
            }
        ],
        "files": [{"source_file": "graphics/x.png", "fate": fate, "code": "extractor_unknown"}],
    }


def _corpus(tmp_path: Path) -> Path:
    (tmp_path / "graphics").mkdir()
    (tmp_path / "graphics" / "x.png").write_bytes(b"png bytes")
    return tmp_path


def test_row1_is_green_when_every_type_has_a_fasit() -> None:
    inventory = {
        "documents": {f"f{s}": {"suffix": s, "elements": {}} for s in TABLE},
        "files": {},
    }
    row = gate.row1(TABLE, inventory, inventory)
    assert (row.k, row.m, row.status) == (13, 13, gate.GREEN)


def test_row1_is_red_when_one_type_lacks_a_fasit() -> None:
    inventory = {
        "documents": {f"f{s}": {"suffix": s, "elements": {}} for s in TABLE[:-1]},
        "files": {},
    }
    row = gate.row1(TABLE, inventory, inventory)
    assert (row.k, row.m, row.status) == (12, 13, gate.RED)


def test_row1_is_red_when_the_committed_fasit_is_stale() -> None:
    inventory = {"documents": {"f.md": {"suffix": ".md", "elements": {"heading": 1}}}}
    fresh = {"documents": {"f.md": {"suffix": ".md", "elements": {"heading": 2}}}}
    row = gate.row1([".md"], inventory, fresh)
    assert (row.k, row.status) == (0, gate.RED)


def test_row2_is_red_without_the_door() -> None:
    row = gate.row2([".md"], _inventory(), _build(), door=False)
    assert (row.k, row.m, row.status) == (0, 1, gate.RED)


def test_row2_is_green_when_the_declared_inventory_equals_the_witness() -> None:
    row = gate.row2([".md"], _inventory(), _build(accounting=_declared()), door=True)
    assert (row.k, row.m, row.status) == (1, 1, gate.GREEN)


def test_row2_is_red_when_the_declared_inventory_is_one_off() -> None:
    declared = _declared()
    declared["documents"][0]["inventory"]["heading"] = 3
    row = gate.row2([".md"], _inventory(), _build(accounting=declared), door=True)
    assert (row.k, row.status) == (0, gate.RED)


def test_row3_is_green_when_every_element_and_file_has_one_fate(tmp_path: Path) -> None:
    units = gate.account(_inventory(), _build(accounting=_declared()), _corpus(tmp_path))
    row = gate.row3(units, door=True)
    assert (row.k, row.m, row.status) == (2, 2, gate.GREEN)


@pytest.mark.parametrize(("heading", "u", "d"), [(1, 1, 0), (3, 0, 1)])
def test_row3_is_red_one_element_either_side(tmp_path: Path, heading: int, u: int, d: int) -> None:
    units = gate.account(
        _inventory(), _build(accounting=_declared(heading=heading)), _corpus(tmp_path)
    )
    document = units[0]
    assert (document.unaccounted, document.double) == (u, d)
    assert gate.row3(units, door=True).status == gate.RED


def test_row3_is_red_when_no_fate_is_declared(tmp_path: Path) -> None:
    units = gate.account(_inventory(), _build(), _corpus(tmp_path))
    assert units[0].unaccounted == 3
    assert gate.row3(units, door=False).status == gate.RED


def test_a_file_carried_through_a_document_and_rejected_is_double_booked(tmp_path: Path) -> None:
    corpus = _corpus(tmp_path)
    carried = {gate._sha12(corpus / "graphics" / "x.png")}
    units = gate.account(_inventory(), _build(accounting=_declared(), assets=carried), corpus)
    assert (units[1].unaccounted, units[1].double) == (0, 1)


def test_a_file_carried_through_a_document_and_declared_carried_is_clean(tmp_path: Path) -> None:
    corpus = _corpus(tmp_path)
    carried = {gate._sha12(corpus / "graphics" / "x.png")}
    build = _build(accounting=_declared(fate="carried"), assets=carried)
    assert gate.account(_inventory(), build, corpus)[1].clean


def test_a_file_declared_carried_without_its_bytes_is_unaccounted(tmp_path: Path) -> None:
    build = _build(accounting=_declared(fate="carried"))
    unit = gate.account(_inventory(), build, _corpus(tmp_path))[1]
    assert (unit.unaccounted, unit.double) == (1, 0)


def test_an_unpointed_file_sharing_bytes_with_a_carried_one_is_not_carried(
    tmp_path: Path,
) -> None:
    corpus = _corpus(tmp_path)
    (corpus / "graphics" / "twin.png").write_bytes(b"png bytes")
    inventory = _inventory()
    inventory["files"]["graphics/twin.png"] = {"pointed_at_by": []}
    carried = {gate._sha12(corpus / "graphics" / "x.png")}
    units = gate.account(inventory, _build(assets=carried), corpus)
    assert [(u.name, u.double) for u in units[1:]] == [
        ("graphics/twin.png", 0),
        ("graphics/x.png", 1),
    ]


def test_without_the_door_double_booking_is_derived_from_conservation(tmp_path: Path) -> None:
    corpus = _corpus(tmp_path)
    carried = {gate._sha12(corpus / "graphics" / "x.png")}
    assert gate.account(_inventory(), _build(assets=carried), corpus)[1].double == 1
    assert gate.account(_inventory(), _build(), corpus)[1].clean


_HONEST_LOG = (
    "* **Images**: 0 carried of 1 found, written to `assets/`.\n"
    "* a.md: 3 elements found in the source, 0 carried: document rejected `fail_secure`\n"
)


def _rejected_inventory() -> dict[str, Any]:
    inventory = _inventory()
    inventory["documents"]["a.md"]["images"] = [{"kind": "local"}]
    return inventory


def test_row4_is_green_when_the_log_names_what_the_rejected_document_held() -> None:
    row = gate.row4(_rejected_inventory(), _build(sources=set(), log=_HONEST_LOG))
    assert (row.k, row.m, row.status) == (1, 1, gate.GREEN)


@pytest.mark.parametrize(
    "log",
    [
        _HONEST_LOG.replace("0 carried of 1 found", "0 carried of 0 found"),
        _HONEST_LOG.replace("3 elements", "2 elements"),
        _HONEST_LOG.replace(" `fail_secure`", ""),
    ],
)
def test_row4_is_red_when_the_log_understates_the_rejected_document(log: str) -> None:
    row = gate.row4(_rejected_inventory(), _build(sources=set(), log=log))
    assert (row.k, row.status) == (0, gate.RED)


def test_row4_cannot_be_green_when_nothing_was_rejected() -> None:
    row = gate.row4(_rejected_inventory(), _build(log=_HONEST_LOG))
    assert (row.m, row.status) == (0, gate.RED)


def test_row5_is_green_when_the_witnesses_agree() -> None:
    row = gate.row5([("pair", gate.compare({"p": 3}, {"p": 3}))], [])
    assert (row.k, row.m, row.status) == (1, 1, gate.GREEN)


def test_row5_is_red_with_both_numbers_when_they_disagree_by_one() -> None:
    row = gate.row5([("pair", gate.compare({"p": 3}, {"p": 4}))], [])
    assert row.status == gate.RED
    assert row.details == ["pair: p: 3 vs 4"]


def test_row5_is_red_when_a_witness_is_missing() -> None:
    assert gate.row5([("pair", gate.compare({"p": 3}, None))], []).status == gate.RED


def test_row6_without_its_source_is_red_locally_and_skipped_in_ci(tmp_path: Path) -> None:
    missing = tmp_path / "absent"
    local = gate.row6(missing, ci=False)
    ci = gate.row6(missing, ci=True)
    assert (local.status, local.fails) == (gate.RED, True)
    assert (ci.status, ci.fails) == (gate.SKIPPED, False)
    assert "source missing" in ci.reason


def test_the_proposed_exceptions_are_not_applied() -> None:
    assert gate.APPROVED_EXCEPTIONS == frozenset()
    assert "NOT APPROVED" in gate.render([])


def test_bad_usage_exits_two() -> None:
    with pytest.raises(SystemExit) as exc:
        gate.main(["--no-such-flag"])
    assert exc.value.code == 2


# --- 3. the real build at this commit ----------------------------------------


@pytest.fixture(scope="module")
def real_rows() -> list[gate.Row]:
    pytest.importorskip("pdfplumber")
    pytest.importorskip("pypandoc")
    return gate.evaluate(r761=None, ci=True, consume=False)


def test_the_door_exists() -> None:
    assert gate.door_available()


def test_the_real_gate_is_green_on_every_fixture_row(real_rows: list[gate.Row]) -> None:
    """Rows 1-5 against the real `okf build`; row 6 needs R761 and is skipped
    here. Written red at 0b00de4 (rows 2, 3, 4 red), green once the build
    accounts for every element."""
    assert [(r.number, r.status) for r in real_rows] == [
        (1, gate.GREEN),
        (2, gate.GREEN),
        (3, gate.GREEN),
        (4, gate.GREEN),
        (5, gate.GREEN),
        (6, gate.SKIPPED),
    ]
    assert real_rows[2].reason.startswith("u = 0 unaccounted, d = 0 double-booked")