Kjell Tore Guttormsen
4f45fe6037
New: test_ingest_sql.py (connector mechanics + typed rendering + dispatch), test_ingest_sql_loadbearing.py (SQL seams: typed-NULL, typed-REAL, navigability via unchanged okf), test_ingest_golden_sql.py + examples/ingest-golden-sql/ (byte-frozen sqlite golden; NULL->empty, 4200.0->4200.0 and escape discriminate a typed impl from naive ones). Updated: sql-source refusal test repointed to http (I6); manifest schema-breadth comment. Each new seam proven RED at its detach point. Refs: shared/ingest-spec.md 11 golden + load-bearing table
367 lines
15 KiB
Python
367 lines
15 KiB
Python
"""I2 Steps 2–4 tests — CSV connector, deterministic materialization, index semantics.
|
||
|
||
Covers the §4 connector rules (fail-closed ``root`` boundary via ``safe_resolve``, streaming
|
||
``max_rows`` cap — error, never silent truncation §8), the §5 rendering rules (text-verbatim
|
||
cells, escape order ``\\`` BEFORE ``|``, newlines → single space with CRLF as ONE unit,
|
||
header cells escaped identically to data cells, LF-only + exactly one trailing newline), the
|
||
§5/§7 provenance frontmatter (exact key order, verbatim ``ingested_at``, ``{stem}@{hash16}``
|
||
stamp), and the §3/§6 bundle semantics (bundle_summary index, managed-link removal + in-place
|
||
label refresh, curated-collision gate, §10 idempotence). Golden regression lives separately
|
||
in tests/test_ingest_golden.py; the detach-RED seam quartet in tests/test_ingest_loadbearing.py.
|
||
"""
|
||
|
||
import copy
|
||
import hashlib
|
||
import json
|
||
import logging
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import pytest
|
||
|
||
from portfolio_optimiser import okf
|
||
from portfolio_optimiser.ingest import IngestError, materialize, read_csv, render_table
|
||
from portfolio_optimiser.retrieval import PathSecurityError
|
||
|
||
_INGESTED_AT = "2026-07-03T12:00:00Z"
|
||
|
||
# --- local CSV-catalogue builder (conftest's LLM fixtures are irrelevant to ingest) ---
|
||
|
||
|
||
def _catalogue(tmp_path: Path, files: dict[str, bytes]) -> Path:
|
||
root = tmp_path / "catalogue"
|
||
root.mkdir()
|
||
for name, content in files.items():
|
||
(root / name).write_bytes(content)
|
||
return root
|
||
|
||
|
||
def _read(root: Path, query: str = "data.csv", max_rows: int = 100) -> Any:
|
||
return read_csv(root, query, max_rows=max_rows)
|
||
|
||
|
||
# --- Step 2: connector + renderer ---
|
||
|
||
|
||
def test_table_rendering_header_separator_rows_in_source_order(tmp_path: Path) -> None:
|
||
root = _catalogue(tmp_path, {"data.csv": b"a,b\n3,4\n1,2\n"})
|
||
header, rows = _read(root)
|
||
assert render_table(header, rows) == "| a | b |\n| --- | --- |\n| 3 | 4 |\n| 1 | 2 |\n"
|
||
|
||
|
||
def test_cell_escaping_backslash(tmp_path: Path) -> None:
|
||
assert render_table(["h"], [["a\\b"]]).splitlines()[2] == "| a\\\\b |"
|
||
|
||
|
||
def test_cell_escaping_pipe(tmp_path: Path) -> None:
|
||
assert render_table(["h"], [["a|b"]]).splitlines()[2] == "| a\\|b |"
|
||
|
||
|
||
def test_cell_escaping_embedded_newline_to_single_space(tmp_path: Path) -> None:
|
||
root = _catalogue(tmp_path, {"data.csv": b'h\n"x\ny"\n'})
|
||
header, rows = _read(root)
|
||
assert rows == [["x\ny"]] # csv preserves the quoted newline...
|
||
assert render_table(header, rows).splitlines()[2] == "| x y |" # ...rendering collapses it
|
||
|
||
|
||
def test_escape_order_backslash_before_pipe(tmp_path: Path) -> None:
|
||
# Input cell `\|`: escaping `\`→`\\` FIRST then `|`→`\|` yields `\\\|`; the reverse
|
||
# order would double-escape the introduced backslash.
|
||
assert render_table(["h"], [["\\|"]]).splitlines()[2] == "| \\\\\\| |"
|
||
|
||
|
||
def test_crlf_embedded_newline_is_one_space(tmp_path: Path) -> None:
|
||
root = _catalogue(tmp_path, {"data.csv": b'h\n"x\r\ny"\n'})
|
||
header, rows = _read(root)
|
||
assert render_table(header, rows).splitlines()[2] == "| x y |" # ONE space, not two
|
||
|
||
|
||
def test_header_cells_escaped_like_data_cells(tmp_path: Path) -> None:
|
||
root = _catalogue(tmp_path, {"data.csv": b'"h|1","h\n2"\nv1,v2\n'})
|
||
header, rows = _read(root)
|
||
assert render_table(header, rows).splitlines()[0] == "| h\\|1 | h 2 |"
|
||
|
||
|
||
def test_header_only_csv_renders_no_data_rows(tmp_path: Path) -> None:
|
||
root = _catalogue(tmp_path, {"data.csv": b"a,b\n"})
|
||
header, rows = _read(root)
|
||
assert render_table(header, rows) == "| a | b |\n| --- | --- |\n"
|
||
|
||
|
||
def test_empty_csv_raises(tmp_path: Path) -> None:
|
||
root = _catalogue(tmp_path, {"data.csv": b""})
|
||
with pytest.raises(IngestError):
|
||
_read(root)
|
||
|
||
|
||
def test_ragged_row_raises(tmp_path: Path) -> None:
|
||
root = _catalogue(tmp_path, {"data.csv": b"a,b\n1\n"})
|
||
with pytest.raises(IngestError):
|
||
_read(root)
|
||
|
||
|
||
def test_max_rows_cap_is_an_error_not_truncation(tmp_path: Path) -> None:
|
||
root = _catalogue(tmp_path, {"data.csv": b"a\n1\n2\n3\n"})
|
||
with pytest.raises(IngestError):
|
||
_read(root, max_rows=2)
|
||
|
||
|
||
def test_path_escape_raises(tmp_path: Path) -> None:
|
||
root = _catalogue(tmp_path, {"data.csv": b"a\n"})
|
||
(tmp_path / "outside.csv").write_bytes(b"a\n1\n")
|
||
with pytest.raises(PathSecurityError):
|
||
_read(root, query="../outside.csv")
|
||
|
||
|
||
def test_missing_root_raises_ingest_error(tmp_path: Path) -> None:
|
||
with pytest.raises(IngestError):
|
||
read_csv(tmp_path / "no-such-root", "data.csv", max_rows=10)
|
||
|
||
|
||
def test_missing_query_file_raises_ingest_error(tmp_path: Path) -> None:
|
||
root = _catalogue(tmp_path, {"data.csv": b"a\n"})
|
||
with pytest.raises(IngestError):
|
||
_read(root, query="absent.csv")
|
||
|
||
|
||
def test_bom_never_leaks_into_first_header_cell(tmp_path: Path) -> None:
|
||
root = _catalogue(tmp_path, {"data.csv": b"\xef\xbb\xbfa,b\n1,2\n"})
|
||
header, _rows = _read(root)
|
||
assert header == ["a", "b"]
|
||
|
||
|
||
# --- Step 3: materialization — provenance frontmatter, LF bytes, staging ---
|
||
|
||
|
||
def _project(
|
||
tmp_path: Path,
|
||
*,
|
||
extractions: list[dict[str, Any]] | None = None,
|
||
source: dict[str, Any] | None = None,
|
||
files: dict[str, bytes] | None = None,
|
||
) -> tuple[Path, Path]:
|
||
"""Write a manifest + CSV catalogue under tmp_path; return (manifest_path, bundle_dir)."""
|
||
catalogue = tmp_path / "catalogue"
|
||
catalogue.mkdir(exist_ok=True)
|
||
for name, content in (files or {"costs.csv": b"item,cost\nled,120\nhvac,80\n"}).items():
|
||
(catalogue / name).write_bytes(content)
|
||
manifest = {
|
||
"manifest_version": 1,
|
||
# Relative root — pinned decision: resolves against the manifest file's directory.
|
||
"source": source or {"type": "file", "id": "prosjekt-arkiv", "root": "catalogue"},
|
||
"bundle_summary": "Cost extracts from the project archive.",
|
||
"extractions": extractions
|
||
or [
|
||
{
|
||
"id": "costs",
|
||
"title": "Project costs",
|
||
"query": "costs.csv",
|
||
"okf_type": "dataset",
|
||
"max_rows": 100,
|
||
}
|
||
],
|
||
}
|
||
manifest_path = tmp_path / "manifest.json"
|
||
manifest_path.write_text(json.dumps(manifest), encoding="utf-8")
|
||
return manifest_path, tmp_path / "bundle"
|
||
|
||
|
||
def test_frontmatter_keys_in_exact_spec_order_raw_text(tmp_path: Path) -> None:
|
||
manifest_path, bundle_dir = _project(tmp_path)
|
||
(written,) = materialize(manifest_path, bundle_dir, ingested_at=_INGESTED_AT)
|
||
lines = written.read_text(encoding="utf-8").splitlines()
|
||
keys = [line.partition(":")[0] for line in lines[1:8]]
|
||
# §5: exactly these keys, in exactly this order.
|
||
assert lines[0] == "---" and lines[8] == "---"
|
||
assert keys == [
|
||
"type",
|
||
"title",
|
||
"source_system",
|
||
"source_query",
|
||
"ingested_at",
|
||
"ingest_manifest",
|
||
"generated",
|
||
]
|
||
|
||
|
||
def test_provenance_roundtrips_via_unchanged_parse_frontmatter(tmp_path: Path) -> None:
|
||
manifest_path, bundle_dir = _project(tmp_path)
|
||
(written,) = materialize(manifest_path, bundle_dir, ingested_at=_INGESTED_AT)
|
||
fm = okf.parse_frontmatter(written)
|
||
assert fm["type"] == "dataset"
|
||
assert fm["title"] == "Project costs"
|
||
assert fm["source_system"] == "prosjekt-arkiv"
|
||
assert fm["source_query"] == "costs.csv"
|
||
assert fm["ingested_at"] == _INGESTED_AT # verbatim (§5)
|
||
assert fm["generated"] == "true" # parse_frontmatter returns strings, never booleans
|
||
expected = "manifest@" + hashlib.sha256(manifest_path.read_bytes()).hexdigest()[:16]
|
||
assert fm["ingest_manifest"] == expected
|
||
|
||
|
||
def test_generated_file_is_lf_only_with_one_trailing_newline(tmp_path: Path) -> None:
|
||
manifest_path, bundle_dir = _project(tmp_path)
|
||
(written,) = materialize(manifest_path, bundle_dir, ingested_at=_INGESTED_AT)
|
||
data = written.read_bytes()
|
||
assert b"\r" not in data
|
||
assert data.endswith(b"\n") and not data.endswith(b"\n\n")
|
||
|
||
|
||
def test_invalid_ingested_at_raises_value_error(tmp_path: Path) -> None:
|
||
manifest_path, bundle_dir = _project(tmp_path)
|
||
# §5 format is ISO-8601 UTC with a Z suffix — validated by regex (datetime.fromisoformat
|
||
# rejects 'Z' on Python 3.10, the repo floor).
|
||
for bad in ("2026-07-03 12:00:00", "2026-07-03T12:00:00+00:00", "2026-07-03", ""):
|
||
with pytest.raises(ValueError):
|
||
materialize(manifest_path, bundle_dir, ingested_at=bad)
|
||
|
||
|
||
def test_materialize_creates_nonexistent_nested_bundle_dir(tmp_path: Path) -> None:
|
||
manifest_path, _ = _project(tmp_path)
|
||
nested = tmp_path / "deep" / "nested" / "bundle"
|
||
(written,) = materialize(manifest_path, nested, ingested_at=_INGESTED_AT)
|
||
assert written.is_file() and written.parent == nested
|
||
|
||
|
||
def test_source_call_logged_with_id_timestamp_rowcount(
|
||
tmp_path: Path, caplog: pytest.LogCaptureFixture
|
||
) -> None:
|
||
manifest_path, bundle_dir = _project(tmp_path)
|
||
with caplog.at_level(logging.INFO, logger="portfolio_optimiser.ingest"):
|
||
materialize(manifest_path, bundle_dir, ingested_at=_INGESTED_AT)
|
||
# §8: which source, when (the deterministic ingested_at argument), row count.
|
||
joined = " ".join(record.getMessage() for record in caplog.records)
|
||
assert "prosjekt-arkiv" in joined and _INGESTED_AT in joined and "rows=2" in joined
|
||
|
||
|
||
def test_two_runs_with_identical_inputs_are_byte_identical(tmp_path: Path) -> None:
|
||
manifest_path, bundle_dir = _project(tmp_path)
|
||
materialize(manifest_path, bundle_dir, ingested_at=_INGESTED_AT)
|
||
first = {p.name: p.read_bytes() for p in sorted(bundle_dir.iterdir())}
|
||
materialize(manifest_path, bundle_dir, ingested_at=_INGESTED_AT)
|
||
second = {p.name: p.read_bytes() for p in sorted(bundle_dir.iterdir())}
|
||
assert first == second # §10 idempotence at file level
|
||
|
||
|
||
def test_http_source_has_no_connector(tmp_path: Path) -> None:
|
||
# The polymorphic schema VALIDATES http (brief assumption 1); executing it is the gated
|
||
# I6 extension point — attempting to materialize is an explicit refusal, never a silent
|
||
# no-op. (sql now EXECUTES, I4; see tests/test_ingest_sql.py.)
|
||
manifest_path, bundle_dir = _project(
|
||
tmp_path, source={"type": "http", "id": "api", "base_url": "https://host/api"}
|
||
)
|
||
with pytest.raises(IngestError):
|
||
materialize(manifest_path, bundle_dir, ingested_at=_INGESTED_AT)
|
||
|
||
|
||
# --- Step 4: index generation + stamped-replacement semantics ---
|
||
|
||
_TWO_EXTRACTIONS = [
|
||
{
|
||
"id": "costs",
|
||
"title": "Project costs",
|
||
"query": "costs.csv",
|
||
"okf_type": "dataset",
|
||
"max_rows": 100,
|
||
},
|
||
{
|
||
"id": "meta",
|
||
"title": "Catalogue metadata",
|
||
"query": "meta.csv",
|
||
"okf_type": "reference",
|
||
"max_rows": 10,
|
||
},
|
||
]
|
||
_TWO_FILES = {
|
||
"costs.csv": b"item,cost\nled,120\n",
|
||
"meta.csv": b"key,value\nowner,anlegg\n",
|
||
}
|
||
|
||
|
||
def _bundle_bytes(bundle_dir: Path) -> dict[str, bytes]:
|
||
return {p.name: p.read_bytes() for p in sorted(bundle_dir.iterdir()) if p.is_file()}
|
||
|
||
|
||
def test_fresh_dir_gets_index_with_bundle_summary_and_links_in_manifest_order(
|
||
tmp_path: Path,
|
||
) -> None:
|
||
manifest_path, bundle_dir = _project(
|
||
tmp_path, extractions=copy.deepcopy(_TWO_EXTRACTIONS), files=dict(_TWO_FILES)
|
||
)
|
||
materialize(manifest_path, bundle_dir, ingested_at=_INGESTED_AT)
|
||
index = (bundle_dir / "index.md").read_text(encoding="utf-8")
|
||
assert index.startswith("Cost extracts from the project archive.\n")
|
||
link_lines = [line for line in index.splitlines() if line.startswith("- [")]
|
||
assert link_lines == [
|
||
"- [Project costs](ingest-costs.md)",
|
||
"- [Catalogue metadata](ingest-meta.md)",
|
||
]
|
||
|
||
|
||
def test_curated_index_line_preserved_byte_for_byte(tmp_path: Path) -> None:
|
||
manifest_path, bundle_dir = _project(tmp_path)
|
||
bundle_dir.mkdir()
|
||
curated = "Hand-written summary.\n\n- [Curated note](note.md)\n"
|
||
(bundle_dir / "index.md").write_bytes(curated.encode("utf-8"))
|
||
materialize(manifest_path, bundle_dir, ingested_at=_INGESTED_AT)
|
||
index = (bundle_dir / "index.md").read_text(encoding="utf-8")
|
||
assert index.startswith(curated) # §6: every unmanaged line preserved verbatim
|
||
assert "- [Project costs](ingest-costs.md)" in index
|
||
|
||
|
||
def test_removed_extraction_drops_file_and_index_link(tmp_path: Path) -> None:
|
||
manifest_path, bundle_dir = _project(
|
||
tmp_path, extractions=copy.deepcopy(_TWO_EXTRACTIONS), files=dict(_TWO_FILES)
|
||
)
|
||
materialize(manifest_path, bundle_dir, ingested_at=_INGESTED_AT)
|
||
# Re-materialize with `meta` dropped from the manifest.
|
||
manifest_path2, _ = _project(tmp_path, files=dict(_TWO_FILES))
|
||
materialize(manifest_path2, bundle_dir, ingested_at=_INGESTED_AT)
|
||
assert not (bundle_dir / "ingest-meta.md").exists()
|
||
index = (bundle_dir / "index.md").read_text(encoding="utf-8")
|
||
assert "](ingest-meta.md)" not in index # §6: link to removed ingest file is gone
|
||
assert "- [Project costs](ingest-costs.md)" in index # sibling link intact
|
||
|
||
|
||
def test_changed_title_refreshes_label_in_place(tmp_path: Path) -> None:
|
||
manifest_path, bundle_dir = _project(
|
||
tmp_path, extractions=copy.deepcopy(_TWO_EXTRACTIONS), files=dict(_TWO_FILES)
|
||
)
|
||
materialize(manifest_path, bundle_dir, ingested_at=_INGESTED_AT)
|
||
renamed = copy.deepcopy(_TWO_EXTRACTIONS)
|
||
renamed[0]["title"] = "Renamed costs"
|
||
manifest_path2, _ = _project(tmp_path, extractions=renamed, files=dict(_TWO_FILES))
|
||
materialize(manifest_path2, bundle_dir, ingested_at=_INGESTED_AT)
|
||
link_lines = [
|
||
line
|
||
for line in (bundle_dir / "index.md").read_text(encoding="utf-8").splitlines()
|
||
if line.startswith("- [")
|
||
]
|
||
# §6: the label IS the title — refreshed in place, position preserved (no stale label,
|
||
# no move-to-end).
|
||
assert link_lines == [
|
||
"- [Renamed costs](ingest-costs.md)",
|
||
"- [Catalogue metadata](ingest-meta.md)",
|
||
]
|
||
|
||
|
||
def test_unstamped_collision_fails_and_leaves_bundle_unmodified(tmp_path: Path) -> None:
|
||
manifest_path, bundle_dir = _project(tmp_path)
|
||
bundle_dir.mkdir()
|
||
(bundle_dir / "index.md").write_bytes(b"Curated.\n")
|
||
# A curated file occupying a generated filename WITHOUT the ingest stamp (§3): never
|
||
# overwrite curated content.
|
||
(bundle_dir / "ingest-costs.md").write_bytes(b"---\ntype: note\n---\n\nCurated body.\n")
|
||
before = _bundle_bytes(bundle_dir)
|
||
with pytest.raises(IngestError):
|
||
materialize(manifest_path, bundle_dir, ingested_at=_INGESTED_AT)
|
||
assert _bundle_bytes(bundle_dir) == before
|
||
|
||
|
||
def test_rematerialization_twice_is_byte_identical_including_index(tmp_path: Path) -> None:
|
||
manifest_path, bundle_dir = _project(
|
||
tmp_path, extractions=copy.deepcopy(_TWO_EXTRACTIONS), files=dict(_TWO_FILES)
|
||
)
|
||
materialize(manifest_path, bundle_dir, ingested_at=_INGESTED_AT)
|
||
first = _bundle_bytes(bundle_dir)
|
||
materialize(manifest_path, bundle_dir, ingested_at=_INGESTED_AT)
|
||
assert _bundle_bytes(bundle_dir) == first # §10, index.md included
|