Kjell Tore Guttormsen
af1849f0b3
TDD step 3: read_csv executes a file-source extraction — boundary-checked path resolution via commonpath on canonical paths (rejects .. traversal, absolute paths, symlink escapes, and prefix-collision siblings), utf-8-sig decoding, streaming max_rows cap as a typed error (never silent truncation), ragged-row rejection, and verbatim cells (escaping stays in the §5 renderers). New SourceError in the typed hierarchy. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01QeqhJpYQyghASjiJo5EhGg
154 lines
4.7 KiB
Python
154 lines
4.7 KiB
Python
"""The `file` connector (ingest-spec §4, §8).
|
|
|
|
CSV under `root` with fail-closed boundary-checked path resolution,
|
|
utf-8-sig decoding (BOM never leaks into the first header cell), streaming
|
|
max_rows cap (error, never silent truncation), and ragged-row rejection.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from llm_ingestion_okf.connectors import read_csv
|
|
from llm_ingestion_okf.errors import IngestError, SourceError
|
|
|
|
|
|
def write_csv(root: Path, name: str, text: str) -> Path:
|
|
path = root / name
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(text, encoding="utf-8", newline="")
|
|
return path
|
|
|
|
|
|
# --- error hierarchy ---
|
|
|
|
|
|
def test_source_error_is_ingest_error() -> None:
|
|
assert issubclass(SourceError, IngestError)
|
|
|
|
|
|
# --- happy path ---
|
|
|
|
|
|
def test_reads_header_and_rows_in_source_order(tmp_path: Path) -> None:
|
|
write_csv(tmp_path, "orders.csv", "a,b\n1,x\n2,y\n")
|
|
header, rows = read_csv(tmp_path, "orders.csv", max_rows=10)
|
|
assert header == ["a", "b"]
|
|
assert rows == [["1", "x"], ["2", "y"]]
|
|
|
|
|
|
def test_reads_csv_in_subdirectory(tmp_path: Path) -> None:
|
|
write_csv(tmp_path, "sub/orders.csv", "a\n1\n")
|
|
header, rows = read_csv(tmp_path, "sub/orders.csv", max_rows=10)
|
|
assert header == ["a"]
|
|
assert rows == [["1"]]
|
|
|
|
|
|
def test_cells_are_verbatim_text(tmp_path: Path) -> None:
|
|
# No escaping at the connector layer — that is render_table's job.
|
|
write_csv(tmp_path, "t.csv", 'a\n"pipe|and\\slash"\n')
|
|
_, rows = read_csv(tmp_path, "t.csv", max_rows=10)
|
|
assert rows == [["pipe|and\\slash"]]
|
|
|
|
|
|
def test_quoted_cell_with_embedded_newline_survives_verbatim(tmp_path: Path) -> None:
|
|
write_csv(tmp_path, "t.csv", 'a,b\n"line1\nline2",x\n')
|
|
_, rows = read_csv(tmp_path, "t.csv", max_rows=10)
|
|
assert rows == [["line1\nline2", "x"]]
|
|
|
|
|
|
def test_bom_never_leaks_into_first_header_cell(tmp_path: Path) -> None:
|
|
write_csv(tmp_path, "t.csv", "\ufeffa,b\n1,2\n")
|
|
header, _ = read_csv(tmp_path, "t.csv", max_rows=10)
|
|
assert header == ["a", "b"]
|
|
|
|
|
|
# --- boundary check: fail-closed path resolution (the OKF path rule) ---
|
|
|
|
|
|
def test_dotdot_traversal_rejected(tmp_path: Path) -> None:
|
|
root = tmp_path / "root"
|
|
root.mkdir()
|
|
write_csv(tmp_path, "outside.csv", "a\n1\n")
|
|
with pytest.raises(SourceError):
|
|
read_csv(root, "../outside.csv", max_rows=10)
|
|
|
|
|
|
def test_absolute_query_path_rejected(tmp_path: Path) -> None:
|
|
root = tmp_path / "root"
|
|
root.mkdir()
|
|
outside = write_csv(tmp_path, "outside.csv", "a\n1\n")
|
|
with pytest.raises(SourceError):
|
|
read_csv(root, str(outside), max_rows=10)
|
|
|
|
|
|
def test_symlink_escape_rejected(tmp_path: Path) -> None:
|
|
root = tmp_path / "root"
|
|
root.mkdir()
|
|
write_csv(tmp_path, "outside.csv", "a\n1\n")
|
|
os.symlink(tmp_path / "outside.csv", root / "link.csv")
|
|
with pytest.raises(SourceError):
|
|
read_csv(root, "link.csv", max_rows=10)
|
|
|
|
|
|
def test_prefix_collision_sibling_rejected(tmp_path: Path) -> None:
|
|
root = tmp_path / "data"
|
|
root.mkdir()
|
|
write_csv(tmp_path, "data-evil/t.csv", "a\n1\n")
|
|
with pytest.raises(SourceError):
|
|
read_csv(root, "../data-evil/t.csv", max_rows=10)
|
|
|
|
|
|
# --- input-shape failures: typed errors, never leaked OSError ---
|
|
|
|
|
|
def test_missing_root_rejected(tmp_path: Path) -> None:
|
|
with pytest.raises(SourceError):
|
|
read_csv(tmp_path / "nope", "t.csv", max_rows=10)
|
|
|
|
|
|
def test_root_that_is_a_file_rejected(tmp_path: Path) -> None:
|
|
file_root = write_csv(tmp_path, "afile", "x")
|
|
with pytest.raises(SourceError):
|
|
read_csv(file_root, "t.csv", max_rows=10)
|
|
|
|
|
|
def test_query_not_resolving_to_file_rejected(tmp_path: Path) -> None:
|
|
with pytest.raises(SourceError):
|
|
read_csv(tmp_path, "missing.csv", max_rows=10)
|
|
|
|
|
|
def test_empty_csv_rejected(tmp_path: Path) -> None:
|
|
write_csv(tmp_path, "t.csv", "")
|
|
with pytest.raises(SourceError):
|
|
read_csv(tmp_path, "t.csv", max_rows=10)
|
|
|
|
|
|
def test_ragged_row_rejected(tmp_path: Path) -> None:
|
|
write_csv(tmp_path, "t.csv", "a,b\n1\n")
|
|
with pytest.raises(SourceError):
|
|
read_csv(tmp_path, "t.csv", max_rows=10)
|
|
|
|
|
|
# --- max_rows cap (spec §8): error, never silent truncation ---
|
|
|
|
|
|
def test_row_count_at_cap_is_allowed(tmp_path: Path) -> None:
|
|
write_csv(tmp_path, "t.csv", "a\n1\n2\n")
|
|
_, rows = read_csv(tmp_path, "t.csv", max_rows=2)
|
|
assert len(rows) == 2
|
|
|
|
|
|
def test_exceeding_cap_is_an_error(tmp_path: Path) -> None:
|
|
write_csv(tmp_path, "t.csv", "a\n1\n2\n3\n")
|
|
with pytest.raises(SourceError):
|
|
read_csv(tmp_path, "t.csv", max_rows=2)
|
|
|
|
|
|
def test_header_does_not_count_toward_cap(tmp_path: Path) -> None:
|
|
write_csv(tmp_path, "t.csv", "a\n1\n")
|
|
_, rows = read_csv(tmp_path, "t.csv", max_rows=1)
|
|
assert rows == [["1"]]
|