Kjell Tore Guttormsen
b7f5ce3800
The slug inherited the whole filename stem with no length bound, so a long dropped name produced a filename the filesystem cannot hold — surfacing as an untyped OSError at the write, with an errno that differs per platform (63 on macOS, 36 on Linux). The inbox contract promises a typed per-file outcome, so the gate belongs in the pure function that forms the name. Refusal, not truncation: truncating is lossy AND collision-prone — two long names sharing a prefix would reduce to one filename and the second write would silently claim the first file. Same posture as `inbox_slug_empty`: the library never invents a filename the operator did not give it. The message carries the actual size and the limit, since the fix is to rename the dropped file. Limit verified empirically on APFS 2026-07-25: a 255-byte component writes, a 258-byte one raises errno 63. ext4 and NTFS cap at the same 255. New MaterializationError code `inbox_slug_too_long`, registered in the errors docstring and in the error-code conformance suite.
402 lines
14 KiB
Python
402 lines
14 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 (
|
|
ExtractionError,
|
|
IngestError,
|
|
ManifestError,
|
|
MaterializationError,
|
|
NetworkGateError,
|
|
RenderError,
|
|
SourceError,
|
|
)
|
|
from llm_ingestion_okf.extract import extract_text
|
|
from llm_ingestion_okf.inbox import inbox_filename, inbox_slug, 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.render import sql_value_to_text
|
|
|
|
INGESTED_AT = "2026-07-17T12:00:00Z"
|
|
|
|
|
|
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:
|
|
with pytest.raises(ExtractionError) as excinfo:
|
|
extract_text("doc.pdf", b"binary")
|
|
assert code_of(excinfo) == "extractor_extra_missing"
|
|
|
|
|
|
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_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"
|
|
|
|
|
|
# --- 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"
|