llm-ingestion-okf/tests/test_error_codes.py

"""Stable error codes: `IngestError.code` is the machine-readable API.

Every raise site carries a documented, stable code so consumers can assert
on sub-causes without binding message wording (messages may improve freely;
codes may not change). One test per code — this file IS the registry's
conformance suite.
"""

from __future__ import annotations

import hashlib
import importlib.util
import json
import sqlite3
import sys
import tempfile
import urllib.error
import warnings
from pathlib import Path
from typing import Any

import pytest

import llm_ingestion_okf.connectors as connectors
from llm_ingestion_okf.connectors import read_csv, read_http, read_sql, safe_resolve
from llm_ingestion_okf.errors import (
    ExtractionError,
    IngestError,
    ManifestError,
    MaterializationError,
    NetworkGateError,
    RenderError,
    SegmentationError,
    SourceError,
)
from llm_ingestion_okf.extract import extract_text
from llm_ingestion_okf.importer import (
    BundleDecision,
    ImportDecision,
    import_bundle,
    import_filename,
    import_slug,
)
from llm_ingestion_okf.inbox import (
    GateDecision,
    inbox_filename,
    inbox_slug,
    process_inbox,
    render_inbox_concept,
)
from llm_ingestion_okf.manifest import load_manifest, load_manifest_bytes
from llm_ingestion_okf.materialize import materialize_bundle
from llm_ingestion_okf.profiles import DEFAULT, SEGMENTED_V1
from llm_ingestion_okf.render import sql_value_to_text
from llm_ingestion_okf.segmentation import assert_plan_applies, parse_segmentation_plan

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

FIXTURES = Path(__file__).parent / "fixtures"

# The two pdf-parser codes are only reachable with the optional extra
# installed; the rejection code that replaces them without it is asserted
# unconditionally in tests/test_extract.py.
requires_extract = pytest.mark.skipif(
    importlib.util.find_spec("pdfplumber") is None,
    reason="the optional [extract] extra is not installed",
)


def inbox_concept(**overrides: Any) -> str:
    """A valid inbox concept render, one field at a time made invalid."""
    kwargs: dict[str, Any] = {
        "okf_type": "note",
        "title": "Note",
        "source_file": "note.md",
        "source_bytes": b"x",
        "ingested_at": INGESTED_AT,
    }
    kwargs.update(overrides)
    return render_inbox_concept("body", **kwargs)


def manifest_data(**overrides: Any) -> dict[str, Any]:
    data: dict[str, Any] = {
        "manifest_version": 1,
        "source": {"type": "file", "id": "src-1", "root": "data"},
        "bundle_summary": "A bundle.",
        "extractions": [
            {
                "id": "orders",
                "title": "Orders",
                "query": "orders.csv",
                "okf_type": "dataset",
                "max_rows": 100,
            }
        ],
    }
    data.update(overrides)
    return data


def load(data: dict[str, Any]) -> None:
    load_manifest_bytes(json.dumps(data).encode("utf-8"))


def code_of(excinfo: pytest.ExceptionInfo[IngestError]) -> str:
    return excinfo.value.code


# --- the attribute itself ---


def test_bare_ingest_error_defaults_to_unspecified() -> None:
    assert IngestError("something").code == "unspecified"


def test_code_does_not_change_the_message() -> None:
    exc = SourceError("the message", code="max_rows_exceeded")
    assert str(exc) == "the message"
    assert exc.code == "max_rows_exceeded"


# --- ManifestError codes ---


def test_manifest_unreadable(tmp_path: Path) -> None:
    with pytest.raises(ManifestError) as excinfo:
        load_manifest(tmp_path / "nope.json")
    assert code_of(excinfo) == "manifest_unreadable"


def test_manifest_unreadable_via_materialize(tmp_path: Path) -> None:
    with pytest.raises(ManifestError) as excinfo:
        materialize_bundle(tmp_path / "nope.json", tmp_path / "bundle", INGESTED_AT)
    assert code_of(excinfo) == "manifest_unreadable"


@pytest.mark.parametrize("raw", [b"\xff\xfe", b"not json"])
def test_manifest_invalid_json(raw: bytes) -> None:
    with pytest.raises(ManifestError) as excinfo:
        load_manifest_bytes(raw)
    assert code_of(excinfo) == "manifest_invalid_json"


def test_manifest_version_unsupported() -> None:
    with pytest.raises(ManifestError) as excinfo:
        load(manifest_data(manifest_version=2))
    assert code_of(excinfo) == "manifest_version_unsupported"


def test_source_type_unknown() -> None:
    with pytest.raises(ManifestError) as excinfo:
        load(manifest_data(source={"type": "ftp", "id": "x"}))
    assert code_of(excinfo) == "source_type_unknown"


def test_credential_embedded_in_base_url() -> None:
    source = {"type": "http", "id": "api-1", "base_url": "https://user:pw@example.test"}
    with pytest.raises(ManifestError) as excinfo:
        load(manifest_data(source=source))
    assert code_of(excinfo) == "credential_embedded"


def test_extraction_id_duplicate() -> None:
    extraction = manifest_data()["extractions"][0]
    with pytest.raises(ManifestError) as excinfo:
        load(manifest_data(extractions=[extraction, dict(extraction)]))
    assert code_of(excinfo) == "extraction_id_duplicate"


def test_okf_type_reserved() -> None:
    extraction = dict(manifest_data()["extractions"][0], okf_type="verdict")
    with pytest.raises(ManifestError) as excinfo:
        load(manifest_data(extractions=[extraction]))
    assert code_of(excinfo) == "okf_type_reserved"


@pytest.mark.parametrize(
    "data",
    [
        "not an object",
        manifest_data(bundle_summary=None),
        {k: v for k, v in manifest_data().items() if k != "source"},
        manifest_data(surprise="unknown field"),
        manifest_data(extractions=[]),
        manifest_data(source={"type": "file", "id": "UPPER", "root": "data"}),
        manifest_data(extractions=[dict(manifest_data()["extractions"][0], title="two\nlines")]),
        manifest_data(extractions=[dict(manifest_data()["extractions"][0], title="link [x]")]),
        manifest_data(extractions=[dict(manifest_data()["extractions"][0], max_rows=0)]),
        manifest_data(extractions=[dict(manifest_data()["extractions"][0], query="")]),
    ],
)
def test_manifest_schema_violations_share_one_code(data: Any) -> None:
    with pytest.raises(ManifestError) as excinfo:
        load_manifest_bytes(json.dumps(data).encode("utf-8"))
    assert code_of(excinfo) == "manifest_schema"


# --- SourceError codes ---


def test_path_escape(tmp_path: Path) -> None:
    with pytest.raises(SourceError) as excinfo:
        safe_resolve(tmp_path, "../outside")
    assert code_of(excinfo) == "path_escape"


def test_path_escape_covers_embedded_null(tmp_path: Path) -> None:
    # A NUL byte makes the OS path syscalls raise ValueError before any
    # boundary check runs. SourceError promises "always typed, never a
    # leaked OSError" — an untyped ValueError breaks that contract, so a
    # malformed target must fail closed under the same code as any other
    # target that cannot resolve inside the root.
    with pytest.raises(SourceError) as excinfo:
        safe_resolve(tmp_path, "bad\x00.md")
    assert code_of(excinfo) == "path_escape"


def test_source_root_missing(tmp_path: Path) -> None:
    with pytest.raises(SourceError) as excinfo:
        read_csv(tmp_path / "nope", "a.csv", max_rows=1)
    assert code_of(excinfo) == "source_root_missing"


def test_source_file_missing(tmp_path: Path) -> None:
    with pytest.raises(SourceError) as excinfo:
        read_csv(tmp_path, "nope.csv", max_rows=1)
    assert code_of(excinfo) == "source_file_missing"


def test_csv_no_header(tmp_path: Path) -> None:
    (tmp_path / "empty.csv").write_text("", encoding="utf-8")
    with pytest.raises(SourceError) as excinfo:
        read_csv(tmp_path, "empty.csv", max_rows=1)
    assert code_of(excinfo) == "csv_no_header"


def test_csv_ragged_row(tmp_path: Path) -> None:
    (tmp_path / "ragged.csv").write_text("a,b\n1\n", encoding="utf-8", newline="")
    with pytest.raises(SourceError) as excinfo:
        read_csv(tmp_path, "ragged.csv", max_rows=10)
    assert code_of(excinfo) == "csv_ragged_row"


def test_max_rows_exceeded_csv(tmp_path: Path) -> None:
    (tmp_path / "big.csv").write_text("a\n1\n2\n", encoding="utf-8", newline="")
    with pytest.raises(SourceError) as excinfo:
        read_csv(tmp_path, "big.csv", max_rows=1)
    assert code_of(excinfo) == "max_rows_exceeded"


def test_max_rows_exceeded_sql(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    db = tmp_path / "fixture.db"
    with sqlite3.connect(db) as conn:
        conn.execute("CREATE TABLE t (id INTEGER)")
        conn.executemany("INSERT INTO t VALUES (?)", [(1,), (2,)])
    monkeypatch.setenv("OKF_TEST_DB", str(db))
    with pytest.raises(SourceError) as excinfo:
        read_sql("OKF_TEST_DB", "SELECT id FROM t", max_rows=1)
    assert code_of(excinfo) == "max_rows_exceeded"


def test_max_rows_exceeded_http() -> None:
    with pytest.raises(SourceError) as excinfo:
        read_http("https://example.test", "/x", max_rows=1, get=lambda url, cred: "a\nb\n")
    assert code_of(excinfo) == "max_rows_exceeded"


def test_connection_ref_unset(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.delenv("OKF_TEST_DB", raising=False)
    with pytest.raises(SourceError) as excinfo:
        read_sql("OKF_TEST_DB", "SELECT 1", max_rows=1)
    assert code_of(excinfo) == "connection_ref_unset"


def test_database_missing(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setenv("OKF_TEST_DB", str(tmp_path / "nope.db"))
    with pytest.raises(SourceError) as excinfo:
        read_sql("OKF_TEST_DB", "SELECT 1", max_rows=1)
    assert code_of(excinfo) == "database_missing"


def test_sql_no_columns(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    db = tmp_path / "fixture.db"
    sqlite3.connect(db).close()
    monkeypatch.setenv("OKF_TEST_DB", str(db))
    # BEGIN succeeds read-only and yields no result columns.
    with pytest.raises(SourceError) as excinfo:
        read_sql("OKF_TEST_DB", "BEGIN", max_rows=1)
    assert code_of(excinfo) == "sql_no_columns"


def test_sql_failed(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    db = tmp_path / "fixture.db"
    sqlite3.connect(db).close()
    monkeypatch.setenv("OKF_TEST_DB", str(db))
    with pytest.raises(SourceError) as excinfo:
        read_sql("OKF_TEST_DB", "SELECT nope FROM missing", max_rows=1)
    assert code_of(excinfo) == "sql_failed"


def test_credential_ref_unset(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.delenv("OKF_TEST_TOKEN", raising=False)
    with pytest.raises(SourceError) as excinfo:
        read_http(
            "https://example.test",
            "/x",
            max_rows=1,
            credential_ref="OKF_TEST_TOKEN",
            get=lambda url, cred: "",
        )
    assert code_of(excinfo) == "credential_ref_unset"


def test_http_transport(monkeypatch: pytest.MonkeyPatch) -> None:
    def refuse(request: Any) -> Any:
        raise urllib.error.URLError("refused")

    monkeypatch.setattr(connectors, "urlopen", refuse)
    with pytest.raises(SourceError) as excinfo:
        connectors.urllib_get("https://example.test/x", None)
    assert code_of(excinfo) == "http_transport"


def test_fence_marker_in_body() -> None:
    with pytest.raises(SourceError) as excinfo:
        read_http("https://example.test", "/x", max_rows=10, get=lambda url, cred: "```\n")
    assert code_of(excinfo) == "fence_marker_in_body"


# --- RenderError codes ---


@pytest.mark.parametrize("value", [True, b"bytes"])
def test_unsupported_cell_type(value: object) -> None:
    with pytest.raises(RenderError) as excinfo:
        sql_value_to_text(value)
    assert code_of(excinfo) == "unsupported_cell_type"


def test_unknown_renderer() -> None:
    from dataclasses import replace

    from llm_ingestion_okf.inbox import _resolve_renderer
    from llm_ingestion_okf.profiles import DEFAULT

    profile = replace(DEFAULT, renderers={".md": "no-such-renderer"})
    with pytest.raises(MaterializationError) as excinfo:
        _resolve_renderer(profile, "note.md")
    assert code_of(excinfo) == "unknown_renderer"


# --- ExtractionError codes ---


def test_extractor_unknown() -> None:
    with pytest.raises(ExtractionError) as excinfo:
        extract_text("archive.zip", b"")
    assert code_of(excinfo) == "extractor_unknown"


def test_extractor_extra_missing(monkeypatch: pytest.MonkeyPatch) -> None:
    # The import probe rather than a suffix: `docx` proved this code reachable
    # only while the extra shipped no parser for it. Once it gains a converter,
    # `_UNPARSED_OPTIONAL_EXTENSIONS` empties and the membership branch becomes
    # unreachable — the probe is the path that survives.
    monkeypatch.setitem(sys.modules, "pdfplumber", None)
    with pytest.raises(ExtractionError) as excinfo:
        extract_text("report.pdf", b"%PDF-1.4")
    assert code_of(excinfo) == "extractor_extra_missing"


# The four converter codes are registered here BEFORE anything raises them, so
# no later step invents a spelling. Until the converter lands (the resolver in
# `_pandoc.py`, then the seam in `extract.py`), the falsifiable claim is exactly
# this: the code is documented in the registry and an `ExtractionError` carries
# it unchanged. Those steps replace each of these with a behavioural raise-site
# test; a code that never gains one would show up here as a test that still
# only reads a docstring.


@pytest.mark.parametrize(
    "code",
    [
        "extractor_binary_missing",
        "extractor_binary_version",
        "extractor_convert_error",
        "extractor_empty_conversion",
    ],
)
def test_converter_code_is_registered_and_carried(code: str) -> None:
    registry = ExtractionError.__doc__ or ""
    assert f"`{code}`" in registry, f"{code} is not documented in the registry"
    assert ExtractionError("x", code=code).code == code


def test_extractor_ocr_group_missing(monkeypatch: pytest.MonkeyPatch) -> None:
    """The OCR engine's own absence, which is not the extra's absence.

    Reached through the import probe for the same reason as the code above:
    `None` in `sys.modules` is what CPython treats as a failed import, so the
    test states the same thing on a machine where the group IS installed.
    """
    monkeypatch.setitem(sys.modules, "rapidocr", None)
    with pytest.raises(ExtractionError) as excinfo:
        extract_text("scan.pdf", (FIXTURES / "no-text-layer.pdf").read_bytes(), ocr=True)
    assert code_of(excinfo) == "extractor_ocr_group_missing"


@requires_extract
def test_extractor_empty_pdf() -> None:
    with pytest.raises(ExtractionError) as excinfo:
        extract_text("scan.pdf", (FIXTURES / "no-text-layer.pdf").read_bytes())
    assert code_of(excinfo) == "extractor_empty_pdf"


@requires_extract
def test_extractor_pdf_error() -> None:
    with pytest.raises(ExtractionError) as excinfo:
        extract_text("broken.pdf", b"not a pdf at all")
    assert code_of(excinfo) == "extractor_pdf_error"


def test_extractor_decode_error() -> None:
    with pytest.raises(ExtractionError) as excinfo:
        extract_text("note.txt", b"\xffbad")
    assert code_of(excinfo) == "extractor_decode_error"


def test_extractor_empty_csv() -> None:
    with pytest.raises(ExtractionError) as excinfo:
        extract_text("empty.csv", b"")
    assert code_of(excinfo) == "extractor_empty_csv"


# --- MaterializationError codes ---


def test_ingested_at_invalid(tmp_path: Path) -> None:
    with pytest.raises(MaterializationError) as excinfo:
        materialize_bundle(tmp_path / "m.json", tmp_path / "bundle", "2026-07-17")
    assert code_of(excinfo) == "ingested_at_invalid"


def test_collision_unstamped(tmp_path: Path) -> None:
    src = tmp_path / "src"
    src.mkdir()
    (src / "manifest.json").write_text(json.dumps(manifest_data()), encoding="utf-8")
    (src / "data").mkdir()
    (src / "data" / "orders.csv").write_text("a\n1\n", encoding="utf-8", newline="")
    bundle = tmp_path / "bundle"
    bundle.mkdir()
    (bundle / "ingest-orders.md").write_text("curated, no stamp\n", encoding="utf-8")
    with pytest.raises(MaterializationError) as excinfo:
        materialize_bundle(src / "manifest.json", bundle, INGESTED_AT)
    assert code_of(excinfo) == "collision_unstamped"


def test_inbox_slug_empty() -> None:
    with pytest.raises(MaterializationError) as excinfo:
        inbox_slug("!!!.md")
    assert code_of(excinfo) == "inbox_slug_empty"


def test_inbox_slug_too_long() -> None:
    with pytest.raises(MaterializationError) as excinfo:
        inbox_filename("a" * 247)
    assert code_of(excinfo) == "inbox_slug_too_long"


def test_inbox_slug_collision(tmp_path: Path) -> None:
    # Two dropped names reducing to one generated filename: reported per file,
    # so this code surfaces in InboxResult.failed rather than as a raise.
    inbox = tmp_path / "inbox"
    inbox.mkdir()
    (inbox / "note.md").write_text("a\n", encoding="utf-8")
    (inbox / "note.txt").write_text("b\n", encoding="utf-8")
    result = process_inbox(
        inbox,
        tmp_path / "bundle",
        INGESTED_AT,
        okf_type="note",
        gate=lambda text: GateDecision(sanitized_text=text, disposition="warn"),
    )
    assert {entry.error.code for entry in result.failed} == {"inbox_slug_collision"}


def test_inbox_title_invalid() -> None:
    with pytest.raises(MaterializationError) as excinfo:
        inbox_concept(title="broken [link]")
    assert code_of(excinfo) == "inbox_title_invalid"


def test_inbox_source_file_invalid() -> None:
    with pytest.raises(MaterializationError) as excinfo:
        inbox_concept(source_file="two\nlines.md")
    assert code_of(excinfo) == "inbox_source_file_invalid"


def test_okf_type_reserved_at_the_inbox_door() -> None:
    # Same reserved layer as the manifest code above, enforced at Door B.
    with pytest.raises(MaterializationError) as excinfo:
        inbox_concept(okf_type="verdict")
    assert code_of(excinfo) == "okf_type_reserved"


# --- MaterializationError codes at Door C ---


def approve_all(bundle: dict[str, str], *, origin: str, channel: str) -> BundleDecision:
    """A gate that clears every concept at the non-blocking floor."""
    return BundleDecision(
        concepts=tuple(ImportDecision(path=path, disposition="warn") for path in sorted(bundle))
    )


def place(source: Path, relpath: str, content: str = "body\n") -> None:
    path = source / relpath
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(content, encoding="utf-8", newline="")


def run_import(tmp_path: Path, **overrides: Any) -> Any:
    kwargs: dict[str, Any] = {"origin": "external", "channel": "automatic", "gate": approve_all}
    kwargs.update(overrides)
    return import_bundle(tmp_path / "source", tmp_path / "bundle", INGESTED_AT, **kwargs)


def test_import_path_empty() -> None:
    with pytest.raises(MaterializationError) as excinfo:
        import_slug("!!!.md")
    assert code_of(excinfo) == "import_path_empty"


def test_import_path_too_long() -> None:
    with pytest.raises(MaterializationError) as excinfo:
        import_filename("a" * 250)
    assert code_of(excinfo) == "import_path_too_long"


def test_import_slug_collision(tmp_path: Path) -> None:
    # Two concept paths reducing to one generated filename: reported per
    # concept, so this code surfaces in ImportResult.failed rather than as a raise.
    place(tmp_path / "source", "a/b.md")
    place(tmp_path / "source", "a-b.md")
    result = run_import(tmp_path)
    assert {entry.error.code for entry in result.failed} == {"import_slug_collision"}


def test_import_label_invalid(tmp_path: Path) -> None:
    place(tmp_path / "source", "report [v2].md")
    result = run_import(tmp_path)
    assert [entry.error.code for entry in result.failed] == ["import_label_invalid"]


def test_collision_unstamped_at_the_import_door(tmp_path: Path) -> None:
    # Same code as Door A's stamp collision: the generated name is occupied by
    # content this library cannot prove is its own to replace.
    bundle = tmp_path / "bundle"
    bundle.mkdir()
    (bundle / "import-note.md").write_text("curated\n", encoding="utf-8")
    place(tmp_path / "source", "note.md")
    result = run_import(tmp_path)
    assert [entry.error.code for entry in result.failed] == ["collision_unstamped"]


@pytest.mark.parametrize(("origin", "channel"), [("nope", "automatic"), ("external", "nope")])
def test_import_provenance_invalid(tmp_path: Path, origin: str, channel: str) -> None:
    place(tmp_path / "source", "a.md")
    with pytest.raises(MaterializationError) as excinfo:
        run_import(tmp_path, origin=origin, channel=channel)
    assert code_of(excinfo) == "import_provenance_invalid"


# --- NetworkGateError codes ---


def test_network_opt_in_missing(tmp_path: Path) -> None:
    src = tmp_path / "src"
    src.mkdir()
    data = manifest_data(source={"type": "http", "id": "api-1", "base_url": "https://x.test"})
    data["extractions"][0]["query"] = "/orders"
    (src / "manifest.json").write_text(json.dumps(data), encoding="utf-8")
    with pytest.raises(NetworkGateError) as excinfo:
        materialize_bundle(src / "manifest.json", tmp_path / "bundle", INGESTED_AT)
    assert code_of(excinfo) == "network_opt_in_missing"


# --- SegmentationError codes ---
#
# These arrived during the 1-to-N voyage, when the byte-stability pin over
# `tests/` was the verification mechanism and editing this file would have
# broken it, so they were covered in the new segmentation modules instead. The
# pin was released by the end-to-end gate run of 2026-09-01, and the registry
# convention above applies again: one test per code, and this file is where a
# reader confirms the registry is complete. The behavioural tests in
# `tests/test_segmentation.py` and `tests/test_segmented_inbox.py` stay where
# they are -- they exercise the parser and the door, which is a different
# question from whether every documented code has a raise site.


def segmentation_entry(**overrides: Any) -> dict[str, Any]:
    payload: dict[str, Any] = {
        "segment_id": "s1",
        "path": "krav/3-1/brannkonsept.md",
        "title": "Brannkonsept",
        "okf_type": "requirement",
        "span": [0, 40],
        "ingested_at": INGESTED_AT,
    }
    payload.update(overrides)
    return payload


def segmentation_payload(**overrides: Any) -> dict[str, Any]:
    payload: dict[str, Any] = {
        "version": "1",
        "source_sha256": "a" * 64,
        "text_sha256": "c" * 64,
        "extractor_id": "text",
        "extractor_version": "1.0.0",
        "adjudicated_at": INGESTED_AT,
        "entries": [segmentation_entry()],
    }
    payload.update(overrides)
    return payload


def parse_segmentation_fails(payload: dict[str, Any]) -> str:
    with pytest.raises(SegmentationError) as excinfo:
        parse_segmentation_plan(payload)
    return excinfo.value.code


def run_segmented_inbox(tmp_path: Path, *, plan_overrides: dict[str, Any], **kwargs: Any) -> None:
    inbox = tmp_path / "round"
    inbox.mkdir(parents=True, exist_ok=True)
    body = "Brannkonsept: krav til seksjonering.\n"
    (inbox / "q500.md").write_text(body, encoding="utf-8", newline="")
    payload = segmentation_payload(
        source_sha256=hashlib.sha256(body.encode("utf-8")).hexdigest(),
        entries=[segmentation_entry(span=[0, len(body)])],
    )
    payload.update(plan_overrides)
    process_inbox(
        inbox,
        tmp_path / "bundle",
        INGESTED_AT,
        okf_type="requirement",
        gate=lambda text: GateDecision(sanitized_text=text, disposition="warn"),
        segmentation=parse_segmentation_plan(payload),
        **kwargs,
    )


def test_segmentation_plan_invalid() -> None:
    assert parse_segmentation_fails(segmentation_payload(entries=[])) == "segmentation_plan_invalid"


def test_segmentation_path_invalid() -> None:
    payload = segmentation_payload(entries=[segmentation_entry(path="/krav/absolutt.md")])
    assert parse_segmentation_fails(payload) == "segmentation_path_invalid"


def test_segmentation_span_invalid() -> None:
    payload = segmentation_payload(entries=[segmentation_entry(span=[40, 40])])
    assert parse_segmentation_fails(payload) == "segmentation_span_invalid"


def test_segmentation_duplicate_id() -> None:
    payload = segmentation_payload(
        entries=[segmentation_entry(), segmentation_entry(path="krav/3-2/roemning.md")]
    )
    assert parse_segmentation_fails(payload) == "segmentation_duplicate_id"


def test_segmentation_extractor_mismatch() -> None:
    plan = parse_segmentation_plan(segmentation_payload())
    with pytest.raises(SegmentationError) as excinfo:
        assert_plan_applies(
            plan,
            source_sha256="b" * 64,
            text_sha256=plan.text_sha256,
            extractor_id=plan.extractor_id,
            extractor_version=plan.extractor_version,
        )
    assert code_of(excinfo) == "segmentation_extractor_mismatch"


def test_segmentation_unsupported_profile(tmp_path: Path) -> None:
    with pytest.raises(SegmentationError) as excinfo:
        run_segmented_inbox(tmp_path, plan_overrides={}, profile=DEFAULT)
    assert code_of(excinfo) == "segmentation_unsupported_profile"


def test_segmentation_plan_unmatched(tmp_path: Path) -> None:
    with pytest.raises(SegmentationError) as excinfo:
        run_segmented_inbox(
            tmp_path,
            plan_overrides={"source_sha256": "0" * 64},
            profile=SEGMENTED_V1,
            root_frontmatter_values={"bundle_id": "b-1"},
        )
    assert code_of(excinfo) == "segmentation_plan_unmatched"


# --- asset codes (0.10.0) --------------------------------------------------
#
# One test per code, like every code above it. These five are the only codes in
# the registry that a caller is expected to COUNT rather than to act on: an
# image a reader could not carry becomes a row in the run log and a line in the
# concept, never a failed document.


def test_asset_type_unknown() -> None:
    from llm_ingestion_okf import assets

    with pytest.raises(ExtractionError) as excinfo:
        assets.read_image(b"%PDF-1.7\n", name="figur.png")
    assert excinfo.value.code == "asset_type_unknown"


def test_asset_samples_invalid() -> None:
    from llm_ingestion_okf import assets

    with pytest.raises(ExtractionError) as excinfo:
        assets.encode_png(8, 8, b"\x00", channels=1)
    assert excinfo.value.code == "asset_samples_invalid"


def test_asset_too_large() -> None:
    """One code, three ways to be over the bound: a DECLARED size, a file, and
    a stream that decompresses past it. The third arrived in 0.10.1 after an
    independent review measured a 408 516-byte PDF declaring 1x1 being carried
    at 892 MB of peak RSS."""
    import zlib

    from llm_ingestion_okf import assets

    with pytest.raises(ExtractionError) as excinfo:
        assets.check_size(20_000, 20_000, name="declared")
    assert excinfo.value.code == "asset_too_large"

    with pytest.raises(ExtractionError) as excinfo:
        assets.check_payload(assets.MAX_IMAGE_BYTES + 1, name="file")
    assert excinfo.value.code == "asset_too_large"

    with pytest.raises(ExtractionError) as excinfo:
        assets.inflated_size(zlib.compress(b"\x00" * 4096, 9), name="stream", limit=16)
    assert excinfo.value.code == "asset_too_large"


def test_asset_pdf_unbounded() -> None:
    """A filter whose output cannot be measured before it is produced.

    Its own code because it says something different from `asset_too_large`:
    that one reports a measurement that came out over the bound, this one
    reports that no measurement was possible, so the picture was refused
    unread. Reached through the real PDF path rather than a helper, because
    the decision is which FILTER the stream declares.
    """
    from llm_ingestion_okf.extract import extract_document

    pytest.importorskip("pdfplumber")
    sys.path.insert(0, str(Path(__file__).parent))
    from test_asset_limits import _bomb

    extracted = extract_document(
        "lzw.pdf", _bomb(4, payload=b"\xff" * 512, filters="/LZWDecode"), assets=True
    )
    assert [item.code for item in extracted.rejected] == ["asset_pdf_unbounded"]
    registry = ExtractionError.__doc__ or ""
    assert "`asset_pdf_unbounded`" in registry


def test_asset_size_invalid() -> None:
    """A declared size that is not a size. Its own code because it says
    something different about the document than `asset_too_large` does."""
    from llm_ingestion_okf import assets

    with pytest.raises(ExtractionError) as excinfo:
        assets.check_size(-1, 40_000_000_000, name="negative")
    assert excinfo.value.code == "asset_size_invalid"


def test_asset_remote() -> None:
    from llm_ingestion_okf.extract import extract_document

    document = extract_document(
        "side.html",
        b'<html><body><img src="https://example.invalid/x.png" alt="x"></body></html>',
        assets=True,
    )
    assert [item.code for item in document.rejected] == ["asset_remote"]


def test_asset_unresolved() -> None:
    from llm_ingestion_okf.extract import extract_document

    document = extract_document(
        "side.html",
        b'<html><body><img src="mangler.png" alt="x"></body></html>',
        assets=True,
    )
    assert [item.code for item in document.rejected] == ["asset_unresolved"]


def test_asset_pdf_unsupported() -> None:
    """A PDF image this encoder refuses rather than approximates.

    A 1-bit stencil: carrying it at 8 bits would be a decision about what black
    means, and a wrong one is indistinguishable from a right one in the output.
    """
    pytest.importorskip("pdfplumber")
    from llm_ingestion_okf.extract import extract_document

    data = (
        Path(__file__).parent / "fixtures" / "image-inbox" / "kapittel-84-tabell.pdf"
    ).read_bytes()
    stencil = data.replace(
        b"/BitsPerComponent 8 /Filter /FlateDecode", b"/BitsPerComponent 1 /Filter /FlateDecode", 1
    )
    assert stencil != data
    with warnings.catch_warnings():
        warnings.simplefilter("ignore")
        document = extract_document("krav.pdf", stencil, assets=True)
    assert "asset_pdf_unsupported" in [item.code for item in document.rejected]


def test_asset_collision() -> None:
    """Two different pictures reducing to one asset name, refused in the run.

    Constructed rather than found: the name carries 12 hex of the digest of its
    own bytes, so reaching this by accident is a 48-bit collision. The code
    exists because resolving it silently would lose one of the two pictures
    while every pointer to it kept showing the other.
    """
    from llm_ingestion_okf.assets import ExtractedImage, asset_name
    from llm_ingestion_okf.inbox import _write_assets

    image = ExtractedImage(b"AAAA", "f.png", "image/png", ".png", 1, 1)
    # One name already holding DIFFERENT bytes, which is what a 48-bit digest
    # collision would look like from inside the run.
    seen = {asset_name(image): b"BBBB"}
    with tempfile.TemporaryDirectory() as root:
        with pytest.raises(MaterializationError) as excinfo:
            _write_assets(Path(root) / "bundle", [image], seen)
    assert excinfo.value.code == "asset_collision"