llm-ingestion-okf/tests/test_asset_gate.py

"""THE GATE: does a bundle carry the images its sources declare? (0.10.0)

Written BEFORE the capability, and red on purpose. Its whole job is to state
one number per reader with the DENOMINATOR read out of the source document
rather than out of this package -- "carried 0 of 9" is a measurement; "no image
support" is a sentence. The denominators here are computed by opening the
fixture bytes and counting what the FORMAT says is there (`page.images`,
`word/media/`, `ppt/media/`, `<img`, `<graphic`), so the gate keeps working
when the fixtures change and cannot drift into asserting our own output back at
us.

**Measured at `332961a`**, built from `git archive` rather than from the
editable tree -- an editable install reads `src/` live, so a "before" run taken
in this working tree would have been measuring the change it was supposed to
predate:

    carried 0 of 2 local (2 declared)  kapittel-7-tabell.pdf
    carried 0 of 1 local (1 declared)  kapittel-7-notat.docx
    carried 0 of 1 local (1 declared)  kapittel-7-presentasjon.pptx
    carried 0 of 2 local (3 declared)  kapittel-7-web.html
    carried 0 of 2 local (2 declared)  kapittel-7-sts.xml
    ---------------------------------------------------------------
    carried 0 of 8 local images across 5 documents (9 declared),
    and the bundle held no `assets/` directory at all.

Two of the seven files in the fixture inbox are the PNGs the HTML and STS
documents point at, and the run reports them as `extractor_unknown: 2/7` at
both commits. That is deliberate and unchanged: `.png` as a DROPPED FILE is a
separate question with its own order, and an image reached through a document
is this one.

WHY THE GATE IS A BUNDLE-LEVEL COUNT and not a per-function assertion: the
failure this exists to stop is not "the reader returned nothing". It is a
bundle that looks complete and is not -- a real reference standard saying
"... er gitt i tabell 7-2" over an empty space, with `okf check` green
and `okf quality` unable to see it. So the gate runs the real command, walks
the written bundle, and asks whether the bytes are there.

THE REMOTE IMAGE IS PART OF THE DENOMINATOR AND NOT PART OF THE TARGET. The
HTML fixture carries three `<img>`; one points at `https://example.invalid/`.
Extraction opens no socket -- network access here is an explicit per-run opt-in
and extraction is not on that path -- so that image is FOUND, counted, and
carried as a pointer without a file. A gate that quietly dropped it from the
denominator would report 2 of 2 and hide the one case a reader most needs told.
"""

from __future__ import annotations

import re
import warnings
import zipfile
from pathlib import Path

import pytest

from llm_ingestion_okf import cli, corpus
from llm_ingestion_okf.assets import ASSETS_DIR, IMAGE_POINTER

FIXTURES = Path(__file__).parent / "fixtures" / "image-inbox"
BUNDLE_ID = "asset-gate-fixture"
OKF_VERSION = "0.2"

#: Every document in the fixture inbox, with the reader it exercises.
DOCUMENTS = (
    "kapittel-7-tabell.pdf",
    "kapittel-7-notat.docx",
    "kapittel-7-presentasjon.pptx",
    "kapittel-7-web.html",
    "kapittel-7-sts.xml",
)


# --- denominators, read out of the source ----------------------------------


def declared_images(path: Path) -> int:
    """How many images the SOURCE says it holds, by the format's own rule.

    Never a constant: a constant is this package asserting its own expectation,
    and it goes stale the moment a fixture is regenerated. Each branch reads the
    container the way the format defines it, which is also the number a person
    checking the bundle by hand would arrive at.
    """
    suffix = path.suffix.lower()
    data = path.read_bytes()
    if suffix == ".pdf":
        pdfplumber = pytest.importorskip("pdfplumber")
        with warnings.catch_warnings():
            warnings.simplefilter("ignore")
            with pdfplumber.open(path) as pdf:
                return sum(len(page.images) for page in pdf.pages)
    if suffix in (".docx", ".pptx", ".xlsx"):
        with zipfile.ZipFile(path) as archive:
            return sum(1 for name in archive.namelist() if "/media/" in name)
    text = data.decode("utf-8")
    if suffix in (".html", ".htm"):
        return len(re.findall(r"<img\b", text))
    if suffix == ".xml":
        return len(re.findall(r"<(?:inline-)?graphic\b", text))
    raise AssertionError(f"no declared-image rule for {path.name}")


def local_images(path: Path) -> int:
    """The declared images MINUS the ones whose source is off this machine.

    Separate from :func:`declared_images` and both are reported: the difference
    between them is exactly the network boundary, and collapsing the two would
    turn a boundary into an absence.
    """
    declared = declared_images(path)
    if path.suffix.lower() in (".html", ".htm"):
        remote = len(re.findall(r'<img\b[^>]*src="(?:https?:)?//', path.read_text("utf-8")))
        return declared - remote
    return declared


# --- the run ---------------------------------------------------------------


def _inbox(tmp_path: Path) -> Path:
    inbox = tmp_path / "inbox"
    (inbox / "graphics").mkdir(parents=True)
    for source in sorted(FIXTURES.rglob("*")):
        if source.is_file():
            target = inbox / source.relative_to(FIXTURES)
            target.write_bytes(source.read_bytes())
    return inbox


def _build(inbox: Path, bundle: Path, *extra: str) -> int:
    return cli.main(
        [
            "build",
            str(inbox),
            "--bundle",
            str(bundle),
            "--bundle-id",
            BUNDLE_ID,
            "--okf-version",
            OKF_VERSION,
            *extra,
        ]
    )


def _concepts(bundle: Path) -> list[Path]:
    return [
        path
        for path in sorted(bundle.rglob("*.md"))
        if path.name not in {"index.md", corpus.LOG_NAME}
    ]


def carried_per_source(bundle: Path) -> dict[str, int]:
    """Pointers in the bundle whose asset file is actually present, per source.

    A pointer whose bytes are missing is NOT carried. Counting pointers alone
    would let an empty `assets/` pass the gate, which is the same defect one
    level down -- a bundle that looks complete and is not.
    """
    counts: dict[str, int] = {}
    for concept in _concepts(bundle):
        text = concept.read_text("utf-8")
        source = ""
        for line in text.splitlines():
            if line.startswith("source_file:"):
                source = line.split(":", 1)[1].strip().strip("'\"")
                break
        for match in IMAGE_POINTER.finditer(text):
            if (bundle / ASSETS_DIR / match.group("asset")).is_file():
                counts[source] = counts.get(source, 0) + 1
    return counts


@pytest.fixture(scope="module")
def built(tmp_path_factory: pytest.TempPathFactory) -> tuple[Path, dict[str, int]]:
    pytest.importorskip("pdfplumber")
    pytest.importorskip("pypandoc")
    pytest.importorskip("llm_ingestion_guard")
    root = tmp_path_factory.mktemp("asset-gate")
    inbox = _inbox(root)
    bundle = root / "bundle"
    with warnings.catch_warnings():
        warnings.simplefilter("ignore")
        assert _build(inbox, bundle) == 0
    return bundle, carried_per_source(bundle)


# --- the gate --------------------------------------------------------------


@pytest.mark.parametrize("document", DOCUMENTS)
def test_every_local_image_reaches_the_bundle(
    document: str, built: tuple[Path, dict[str, int]]
) -> None:
    """carried == local, per source document, with both numbers printed."""
    bundle, carried = built
    want = local_images(FIXTURES / document)
    got = carried.get(document, 0)
    assert got == want, (
        f"{document}: carried {got} of {want} local images "
        f"({declared_images(FIXTURES / document)} declared by the source)"
    )


def test_the_bundle_carries_every_local_image_of_every_document(
    built: tuple[Path, dict[str, int]],
) -> None:
    """The whole-bundle row, so the gate reports one number a person can quote."""
    bundle, carried = built
    want = sum(local_images(FIXTURES / document) for document in DOCUMENTS)
    got = sum(carried.values())
    assert got == want, f"carried {got} of {want} local images across {len(DOCUMENTS)} documents"


def test_a_remote_image_is_a_pointer_without_a_file_never_a_silent_drop(
    built: tuple[Path, dict[str, int]],
) -> None:
    """The network boundary, stated in the artifact rather than implied by absence."""
    bundle, _ = built
    html = [path for path in _concepts(bundle) if "kapittel-7-web.html" in path.read_text("utf-8")]
    assert html, "the html document produced no concept at all"
    body = "\n".join(path.read_text("utf-8") for path in html)
    assert "https://example.invalid/ekstern.png" in body, (
        "a remote image must leave a pointer naming where it was, so a reader "
        "learns the document had a figure this bundle does not hold"
    )
    assert "not carried" in body


def test_the_asset_bytes_are_the_sources_own_bytes(
    built: tuple[Path, dict[str, int]],
) -> None:
    """A carried image is byte-identical to the file the source shipped.

    Only the two files the fixture carries as real files can be checked this
    way; the PDF and office images arrive inside a container. That is enough to
    pin the property that matters -- nothing re-encodes an image that already
    is one.
    """
    bundle, _ = built
    written = {path.read_bytes() for path in (bundle / ASSETS_DIR).glob("*")}
    for original in sorted((FIXTURES / "graphics").glob("*.png")):
        assert original.read_bytes() in written, original.name