"""I6 load-bearing seams — the ``http`` path end-to-end (ingest spec §11's detach-RED regime). Each test is built so the seam under test is the ONLY thing preventing the observable artifact (the Fase-2 green-but-dead trap). Every test injects a canned ``http_get`` (no socket, no credentials), so the suite runs with the network cable pulled. - NETWORK GATE (§8/SC1): http is refused fail-fast unless the per-run ``allow_network`` flag is set — and refused BEFORE any source access. Asserted on BOTH branches with the SAME manifest and SAME recording ``get``: flag off → ``IngestError`` + ``get`` NEVER called; flag on → succeeds + ``get`` called. Removing the whole gate → the refuse assertion goes RED (get called, no error); reverting to the *unconditional* refusal (dropping ``and not allow_network``) → the allow assertion goes RED (raises when it should fetch). A refuse-only test would false-green on the second detach. - FENCED-NOT-TABLE (§5 http seam): the body is rendered verbatim inside a fenced code block, NOT a markdown table — a raw ``|`` and ``\`` survive un-escaped. RED if the http path degrades to ``render_table``. - NAVIGABILITY (§2/§11): an http-generated bundle is consumable by the UNCHANGED ``okf.navigate_bundle``/``bundle_context`` — reachable via index cross-links, classified by ``okf_type``, provenance riding through. RED when index linking is detached. - OVER-CAP WRITES NOTHING (§8/SC max_rows): an over-cap response raises during in-memory staging, before any disk mutation — no ``ingest-*.md`` is written. RED if silent truncation replaces the raise. """ from __future__ import annotations import json from pathlib import Path from typing import Any import pytest from portfolio_optimiser import okf from portfolio_optimiser.ingest import IngestError, materialize _INGESTED_AT = "2026-07-04T12:00:00Z" def _extraction( ext_id: str, query: str, okf_type: str, *, title: str, max_rows: int = 10 ) -> dict[str, Any]: return { "id": ext_id, "title": title, "query": query, "okf_type": okf_type, "max_rows": max_rows, } def _http_manifest( tmp_path: Path, extractions: list[dict[str, Any]], *, credential_ref: str | None = None ) -> Path: manifest = { "manifest_version": 1, "source": { "type": "http", "id": "api", "base_url": "https://api.example.test/v1", "credential_ref": credential_ref, }, "bundle_summary": "HTTP extracts from a local mock endpoint (http source type).", "extractions": extractions, } path = tmp_path / "manifest.json" path.write_text(json.dumps(manifest), encoding="utf-8") return path def _fixed_get(body: str) -> Any: def get(url: str, credential: str | None) -> str: return body return get # --- LOAD-BEARING: network gate (both branches, same manifest + same get) ------------------------ def test_http_network_gate_refuses_without_flag_and_fetches_with_flag(tmp_path: Path) -> None: manifest = _http_manifest( tmp_path, [_extraction("status", "status", "dataset", title="Service status")] ) calls: list[tuple[str, str | None]] = [] def recording_get(url: str, credential: str | None) -> str: calls.append((url, credential)) return "service ok" # Refuse branch: flag off → IngestError AND get NEVER called (refused before any source access). with pytest.raises(IngestError): materialize( manifest, tmp_path / "b_refuse", ingested_at=_INGESTED_AT, allow_network=False, http_get=recording_get, ) assert calls == [] # §8: no network touched when the flag is absent # Allow branch: flag on → succeeds (writes ingest-status.md) AND get WAS called. written = materialize( manifest, tmp_path / "b_allow", ingested_at=_INGESTED_AT, allow_network=True, http_get=recording_get, ) assert any(p.name == "ingest-status.md" for p in written) assert len(calls) == 1 # the injected transport was exercised # --- LOAD-BEARING: fenced-not-table (verbatim body, no escaping) --------------------------------- def test_http_body_is_fenced_verbatim_not_table_escaped(tmp_path: Path) -> None: manifest = _http_manifest( tmp_path, [_extraction("status", "status", "dataset", title="Service status")] ) payload = r'{"state": "degraded | partial", "path": "c:\temp\cache"}' materialize( manifest, tmp_path / "b", ingested_at=_INGESTED_AT, allow_network=True, http_get=_fixed_get(payload), ) body = (tmp_path / "b" / "ingest-status.md").read_text(encoding="utf-8") assert "```" in body # rendered inside a fenced code block assert payload in body # verbatim: raw | and \ survive assert "\\|" not in body # RED if the http path degraded to render_table (pipe-escaped) # --- LOAD-BEARING: navigability through unchanged OKF -------------------------------------------- def test_http_generated_bundle_is_navigable_via_unchanged_okf(tmp_path: Path) -> None: manifest = _http_manifest( tmp_path, [_extraction("status", "status", "reference", title="Service status")] ) bundle_dir = tmp_path / "b" materialize( manifest, bundle_dir, ingested_at=_INGESTED_AT, allow_network=True, http_get=_fixed_get("service ok\nall systems green"), ) bundle = okf.navigate_bundle(str(bundle_dir)) by_name = {f.name: f for f in bundle.files} assert "ingest-status.md" in by_name assert by_name["ingest-status.md"].type == "reference" assert by_name["ingest-status.md"].frontmatter["ingest_manifest"] # provenance rides through context = okf.bundle_context(bundle) assert "Service status" in context and "service ok" in context # index + body reached # --- LOAD-BEARING: over-cap writes nothing (memory-staging) -------------------------------------- def test_http_over_cap_raises_and_writes_nothing(tmp_path: Path) -> None: manifest = _http_manifest( tmp_path, [_extraction("status", "status", "dataset", title="Service status", max_rows=2)] ) bundle_dir = tmp_path / "b" with pytest.raises(IngestError): # §8: error, never silent truncation materialize( manifest, bundle_dir, ingested_at=_INGESTED_AT, allow_network=True, http_get=_fixed_get("a\nb\nc"), # 3 lines > max_rows=2 ) # Over-cap raises during in-memory staging, before the first disk mutation. assert not list(bundle_dir.glob("ingest-*.md"))