diff --git a/tests/test_track_a_source.py b/tests/test_track_a_source.py new file mode 100644 index 0000000..16fe0e8 --- /dev/null +++ b/tests/test_track_a_source.py @@ -0,0 +1,187 @@ +"""The Hop-A source adapter against the §10 golden fixture — happy path from the +canonical bytes, tamper paths from surgically corrupted copies of them. + +Every tamper case starts from the REAL fixture and breaks exactly one declared fact +(hash, member set, sample_count, dtype, major version, target, month set, cell count, +identifier length) — proving the adapter rejects precisely what the contract says it +must, not merely malformed junk. +""" + +import io +import json +import zipfile +from pathlib import Path + +import numpy as np +import pytest + +from views_postprocessing.unfao import track_a_source as tas + +FIX = Path(__file__).resolve().parent / "fixtures" / "wire_contract" +SHARD = (FIX / "fixture_run_0__lr_ged_sb__m000543.tap.zip").read_bytes() +MANIFEST = json.loads((FIX / "fixture_run_0__lr_ged_sb__manifest.json").read_text()) +SHARD_NAME = MANIFEST["shards"][0]["name"] +SHARD_SHA = MANIFEST["shards"][0]["sha256"] + + +def _retouched_shard(**member_overrides: bytes) -> bytes: + """The fixture shard with named members replaced — everything else byte-identical.""" + with zipfile.ZipFile(io.BytesIO(SHARD)) as zf: + members = {name: zf.read(name) for name in zf.namelist()} + members.update(member_overrides) + buf = io.BytesIO() + with zipfile.ZipFile(buf, "w", zipfile.ZIP_STORED) as zf: + for name, data in members.items(): + zf.writestr(name, data) + return buf.getvalue() + + +def _header(**overrides) -> bytes: + with zipfile.ZipFile(io.BytesIO(SHARD)) as zf: + header = json.loads(zf.read("metadata.json")) + header.update(overrides) + return json.dumps(header).encode() + + +def _sha(data: bytes) -> str: + import hashlib + + return hashlib.sha256(data).hexdigest() + + +# --- happy path: the golden fixture is the oracle ----------------------------------- + + +def test_read_manifest_accepts_the_fixture_manifest(): + manifest = tas.read_manifest((FIX / "fixture_run_0__lr_ged_sb__manifest.json").read_bytes()) + assert manifest["run_id"] == "fixture_run_0" + assert manifest["sidecar_sha256"] is None # Erratum E1 + + +def test_read_shard_round_trips_the_fixture(): + frame, header = tas.read_shard(SHARD, expected_sha256=SHARD_SHA) + assert frame.n_rows == 6 and frame.sample_count == 4 + assert frame.values.dtype == np.float32 + np.testing.assert_array_equal(np.asarray(frame.index.unit), np.arange(100001, 100007)) + np.testing.assert_array_equal(np.asarray(frame.index.time), np.full(6, 543)) + assert header["contract_version"] == "1.5" and header["target"] == "lr_ged_sb" + + +def test_frames_for_target_assembles_the_run(): + frame = tas.frames_for_target(MANIFEST, {SHARD_NAME: SHARD}) + assert frame.n_rows == 6 and frame.sample_count == 4 + + +# --- §3.2 hash verification ----------------------------------------------------------- + + +def test_tampered_bytes_rejected(): + with pytest.raises(tas.TrackASourceError, match="sha256 mismatch"): + tas.read_shard(SHARD + b"\x00", expected_sha256=SHARD_SHA) + + +# --- §3.1 member set ------------------------------------------------------------------- + + +def test_missing_member_rejected(): + with zipfile.ZipFile(io.BytesIO(SHARD)) as zf: + members = {n: zf.read(n) for n in zf.namelist() if n != "identifiers.npz"} + buf = io.BytesIO() + with zipfile.ZipFile(buf, "w", zipfile.ZIP_STORED) as zf: + for name, data in members.items(): + zf.writestr(name, data) + bad = buf.getvalue() + with pytest.raises(tas.TrackASourceError, match="archive members"): + tas.read_shard(bad, expected_sha256=_sha(bad)) + + +# --- header-vs-payload (per-hop ingest asserts) ---------------------------------------- + + +def test_header_lying_about_sample_count_rejected(): + bad = _retouched_shard(**{"metadata.json": _header(sample_count=1024)}) + with pytest.raises(tas.TrackASourceError, match="sample_count=1024"): + tas.read_shard(bad, expected_sha256=_sha(bad)) + + +def test_header_lying_about_dtype_rejected(): + bad = _retouched_shard(**{"metadata.json": _header(dtype="float64")}) + with pytest.raises(tas.TrackASourceError, match="dtype"): + tas.read_shard(bad, expected_sha256=_sha(bad)) + + +def test_identifier_length_mismatch_rejected(): + with zipfile.ZipFile(io.BytesIO(SHARD)) as zf: + with zipfile.ZipFile(io.BytesIO(zf.read("identifiers.npz"))) as ids: + time = np.load(io.BytesIO(ids.read("time.npy"))) + unit = np.load(io.BytesIO(ids.read("unit.npy"))) + short_ids = io.BytesIO() + with zipfile.ZipFile(short_ids, "w", zipfile.ZIP_STORED) as zf: + for name, arr in (("time.npy", time[:-1]), ("unit.npy", unit)): + buf = io.BytesIO() + np.save(buf, arr) + zf.writestr(name, buf.getvalue()) + bad = _retouched_shard(**{"identifiers.npz": short_ids.getvalue()}) + with pytest.raises(tas.TrackASourceError, match="identifier lengths"): + tas.read_shard(bad, expected_sha256=_sha(bad)) + + +# --- §2.1 versioning ------------------------------------------------------------------- + + +def test_major_version_bump_rejected_in_shard_header(): + bad = _retouched_shard(**{"metadata.json": _header(contract_version="2.0")}) + with pytest.raises(tas.TrackASourceError, match="major"): + tas.read_shard(bad, expected_sha256=_sha(bad)) + + +def test_major_version_bump_rejected_in_manifest(): + with pytest.raises(tas.TrackASourceError, match="major"): + tas.read_manifest(json.dumps({**MANIFEST, "contract_version": "2.0"}).encode()) + + +def test_minor_version_drift_accepted(): + bumped = _retouched_shard(**{"metadata.json": _header(contract_version="1.9")}) + frame, header = tas.read_shard(bumped, expected_sha256=_sha(bumped)) + assert header["contract_version"] == "1.9" and frame.n_rows == 6 + + +# --- run assembly (§3.2 completeness; §3.3 manifest-is-identity) ---------------------- + + +def test_missing_shard_bytes_rejected(): + with pytest.raises(tas.TrackASourceError, match="not provided"): + tas.frames_for_target(MANIFEST, {}) + + +def test_shard_target_disagreeing_with_manifest_rejected(): + bad = _retouched_shard(**{"metadata.json": _header(target="lr_ged_ns")}) + manifest = {**MANIFEST, "shards": [{"name": SHARD_NAME, "sha256": _sha(bad)}]} + with pytest.raises(tas.TrackASourceError, match="target"): + tas.frames_for_target(manifest, {SHARD_NAME: bad}) + + +def test_wrong_month_coverage_rejected(): + manifest = {**MANIFEST, "expected_months": [543, 544]} + with pytest.raises(tas.TrackASourceError, match="not provided"): + # month 544's shard is absent entirely — caught at the bytes gate + tas.frames_for_target( + {**manifest, "shards": MANIFEST["shards"] + [{"name": "m544", "sha256": "0" * 64}]}, + {SHARD_NAME: SHARD}, + ) + bad = _retouched_shard(**{"metadata.json": _header(time_id=999)}) + manifest = {**MANIFEST, "shards": [{"name": SHARD_NAME, "sha256": _sha(bad)}]} + with pytest.raises(tas.TrackASourceError, match="months covered"): + tas.frames_for_target(manifest, {SHARD_NAME: bad}) + + +def test_wrong_cell_count_rejected(): + manifest = {**MANIFEST, "expected_cell_count": 7} + with pytest.raises(tas.TrackASourceError, match="cells"): + tas.frames_for_target(manifest, {SHARD_NAME: SHARD}) + + +def test_manifest_missing_required_field_rejected(): + truncated = {k: v for k, v in MANIFEST.items() if k != "expected_months"} + with pytest.raises(tas.TrackASourceError, match="expected_months"): + tas.read_manifest(json.dumps(truncated).encode()) diff --git a/views_postprocessing/unfao/track_a_source.py b/views_postprocessing/unfao/track_a_source.py new file mode 100644 index 0000000..8dc9204 --- /dev/null +++ b/views_postprocessing/unfao/track_a_source.py @@ -0,0 +1,146 @@ +"""Hop-A source adapter: verified Track-A archives → interior ``PredictionFrame``s +(ADR-013 §3 consumer side; the delivery *source* adapter required by the #88 review +amendment — the manager calls this, representation logic never lives inline in it). + +Pandas-free by construction: a Track-A archive is zip + npy + json; this module turns it +into declared numpy primitives and hands them to ``frames.build_prediction_frame``. The +new wire path never touches pandas. + +Declare-don't-guess, per hop: every check here is a **mechanism** guard — the artifact +must match what its own manifest and header *declare* (hashes, counts, shapes, versions). +The §6 *policy* gate (``delivery.draws``, S_min) is separate and runs at the FAO-facing +upload, not here. + +What this module verifies (fail loud, naming expected vs found): +- manifest: contract major-version compatible (§2.1 — MINOR additive, MAJOR breaking); +- shard bytes match the manifest's declared sha256 (§3.2: views-postprocessing verifies + Hop-A hashes on read); +- archive members are exactly the §3.1 triple; header major-version compatible; +- payload matches the header's own declaration: ``(N, sample_count)`` float32, identifier + arrays of length N (the per-hop ingest assert); +- a run's shards cover exactly the manifest's expected months and cell count (§4.2a + feeds on this when the manager awaits all targets). +""" + +from __future__ import annotations + +import hashlib +import io +import json +import zipfile + +import numpy as np +from views_frames import PredictionFrame + +from views_postprocessing.unfao.frames import build_prediction_frame + +_ARCHIVE_MEMBERS = {"y_pred.npy", "identifiers.npz", "metadata.json"} +_SUPPORTED_CONTRACT_MAJOR = "1" + + +class TrackASourceError(ValueError): + """A Track-A artifact does not match what its manifest or header declares.""" + + +def _require_contract_major(version: str, *, where: str) -> None: + major = str(version).split(".", 1)[0] + if major != _SUPPORTED_CONTRACT_MAJOR: + raise TrackASourceError( + f"{where}: contract_version {version!r} has major {major!r}; this consumer " + f"speaks major {_SUPPORTED_CONTRACT_MAJOR!r} (§2.1 — MAJOR bumps are breaking)." + ) + + +def read_manifest(manifest_bytes: bytes) -> dict: + """Parse and validate a Hop-A (run, target) manifest (§3.2).""" + manifest = json.loads(manifest_bytes) + _require_contract_major(manifest.get("contract_version", "?"), where="manifest") + for key in ("run_id", "target", "shards", "expected_months", "expected_cell_count"): + if key not in manifest: + raise TrackASourceError(f"manifest: required field {key!r} is missing.") + return manifest + + +def read_shard(shard_bytes: bytes, *, expected_sha256: str) -> tuple[PredictionFrame, dict]: + """One verified Track-A shard → ``(PredictionFrame, header)``. + + ``expected_sha256`` is the manifest's declaration for these bytes — the §3.2 + hash verification this repo owns on read. + """ + actual = hashlib.sha256(shard_bytes).hexdigest() + if actual != expected_sha256: + raise TrackASourceError( + f"shard: sha256 mismatch — manifest declares {expected_sha256}, " + f"bytes hash to {actual}. Refusing a tampered or torn artifact." + ) + with zipfile.ZipFile(io.BytesIO(shard_bytes)) as zf: + members = set(zf.namelist()) + if members != _ARCHIVE_MEMBERS: + raise TrackASourceError( + f"shard: archive members {sorted(members)} != required {sorted(_ARCHIVE_MEMBERS)} (§3.1)." + ) + header = json.loads(zf.read("metadata.json")) + values = np.load(io.BytesIO(zf.read("y_pred.npy"))) + with zipfile.ZipFile(io.BytesIO(zf.read("identifiers.npz"))) as ids: + time = np.load(io.BytesIO(ids.read("time.npy"))) + unit = np.load(io.BytesIO(ids.read("unit.npy"))) + + _require_contract_major(header.get("contract_version", "?"), where="shard header") + declared_s = header.get("sample_count") + if values.ndim != 2 or values.shape[1] != declared_s: + found = values.shape if values.ndim == 2 else f"{values.ndim}-D" + raise TrackASourceError( + f"shard: header declares sample_count={declared_s} but y_pred is {found} — " + f"the artifact lies about its own draws (per-hop ingest assert)." + ) + if str(values.dtype) != header.get("dtype"): + raise TrackASourceError( + f"shard: header declares dtype={header.get('dtype')!r}, y_pred is {values.dtype}." + ) + n = values.shape[0] + if time.shape != (n,) or unit.shape != (n,): + raise TrackASourceError( + f"shard: identifier lengths (time={time.shape}, unit={unit.shape}) != N={n}." + ) + return build_prediction_frame(values, time, unit), header + + +def frames_for_target(manifest: dict, shard_bytes_by_name: dict) -> PredictionFrame: + """A (run, target)'s verified shards → one interior ``PredictionFrame``. + + ``shard_bytes_by_name`` maps shard ``name`` → downloaded bytes; the manifest is the + only source of which shards exist (§3.3: names are locators, manifest content is + identity). Verifies run completeness against the manifest's own declarations — + months covered exactly, cell count per month — then stacks months into one frame. + """ + frames, months_seen = [], [] + for entry in manifest["shards"]: + name = entry["name"] + if name not in shard_bytes_by_name: + raise TrackASourceError( + f"run: manifest lists shard {name!r} but its bytes were not provided — " + f"an unmanifested or missing shard must not be silently skipped." + ) + frame, header = read_shard(shard_bytes_by_name[name], expected_sha256=entry["sha256"]) + if header.get("target") != manifest["target"]: + raise TrackASourceError( + f"run: shard header target {header.get('target')!r} != manifest target " + f"{manifest['target']!r}." + ) + if frame.n_rows != manifest["expected_cell_count"]: + raise TrackASourceError( + f"run: shard {name!r} carries {frame.n_rows} cells, manifest expects " + f"{manifest['expected_cell_count']}." + ) + frames.append(frame) + months_seen.append(int(header.get("time_id"))) + + if sorted(months_seen) != sorted(int(m) for m in manifest["expected_months"]): + raise TrackASourceError( + f"run: months covered {sorted(months_seen)} != manifest expected " + f"{sorted(manifest['expected_months'])} — a torn run must not be assembled." + ) + values = np.concatenate([f.values for f in frames], axis=0) + time = np.concatenate([np.asarray(f.index.time) for f in frames]) + unit = np.concatenate([np.asarray(f.index.unit) for f in frames]) + return build_prediction_frame(values, time, unit)