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 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 (
    IngestError,
    ManifestError,
    MaterializationError,
    NetworkGateError,
    RenderError,
    SourceError,
)
from llm_ingestion_okf.manifest import load_manifest, load_manifest_bytes
from llm_ingestion_okf.materialize import materialize_bundle
from llm_ingestion_okf.render import sql_value_to_text

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


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"


# --- 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"


# --- 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"