124 lines
4.7 KiB
Python
124 lines
4.7 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
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import pytest
|
||
|
||
from portfolio_optimiser.ingest import IngestError, read_csv, render_table
|
||
from portfolio_optimiser.retrieval import PathSecurityError
|
||
|
||
# --- 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"]
|