portfolio-optimiser/tests/test_ingest_materialize.py

"""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 llm_ingestion_okf import MaterializationError, SourceError

from conftest import expected_generated_stamp
from portfolio_optimiser import okf
from portfolio_optimiser.ingest import IngestError, materialize, read_csv, render_table

_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")
    # Fail-closed containment is unchanged; the library raises its own typed refusal
    # (SourceError, code="path_escape") where the repo-local seam raised PathSecurityError.
    with pytest.raises(SourceError) as exc:
        _read(root, query="../outside.csv")
    assert exc.value.code == "path_escape"


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"] == expected_generated_stamp(_INGESTED_AT)  # 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_is_refused(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). The refusal is unchanged; its type moved
    # from ValueError to the library's MaterializationError.
    for bad in ("2026-07-03 12:00:00", "2026-07-03T12:00:00+00:00", "2026-07-03", ""):
        with pytest.raises(MaterializationError) as exc:
            materialize(manifest_path, bundle_dir, ingested_at=bad)
        assert exc.value.code == "ingested_at_invalid"
    # The refusal is fail-fast: nothing was written before the format was checked.
    assert not bundle_dir.exists()


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)
    # The §8 audit log is still emitted per source call, but the CHANNEL moved with the
    # implementation: "portfolio_optimiser.ingest" -> "llm_ingestion_okf.materialize"
    # (accepted 2026-07-20; nothing in the repo consumed the old logger name).
    with caplog.at_level(logging.INFO, logger="llm_ingestion_okf.materialize"):
        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
    # §8 also bounds what may be logged: never cell contents.
    assert "led-retrofit" not 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_is_refused_without_network_flag(tmp_path: Path) -> None:
    # The polymorphic schema VALIDATES http (brief assumption 1). Since I6 the http connector
    # EXISTS, but materialize refuses it fail-fast unless the per-run allow_network flag is set —
    # the manifest cannot grant itself network access (§8). Default (no flag) → explicit refusal,
    # never a silent no-op. (The allow-branch is covered in tests/test_ingest_http_loadbearing.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