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 urllib.error
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"


# --- 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() -> None:
    # `docx` rather than `pdf`: the extra now ships a pdf parser, so the type
    # that still has none is what proves this code is reachable. The pdf
    # import-probe path is covered in tests/test_extract.py.
    with pytest.raises(ExtractionError) as excinfo:
        extract_text("doc.docx", b"binary")
    assert code_of(excinfo) == "extractor_extra_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,
        "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 / "n500.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,
            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"