Kjell Tore Guttormsen
9d296ca686
Consumer tests were binding message wording (pytest.raises(match=...)) to distinguish sub-causes within one error type. Every raise site now carries a documented, stable code attribute; the registry lives in the errors.py class docstrings. Codes are stable API; message text is explicitly not. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
316 lines
11 KiB
Python
316 lines
11 KiB
Python
"""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], 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_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"
|