diff --git a/tests/test_track_a_source.py b/tests/test_track_a_source.py index 16fe0e8..a40551c 100644 --- a/tests/test_track_a_source.py +++ b/tests/test_track_a_source.py @@ -68,8 +68,11 @@ def test_read_shard_round_trips_the_fixture(): def test_frames_for_target_assembles_the_run(): - frame = tas.frames_for_target(MANIFEST, {SHARD_NAME: SHARD}) + frame, headers = tas.frames_for_target(MANIFEST, {SHARD_NAME: SHARD}) assert frame.n_rows == 6 and frame.sample_count == 4 + # headers ride along in manifest shard order (provenance pass-through, §10.2) + assert [h["time_id"] for h in headers] == [543] + assert headers[0]["provenance"]["ensemble"] == "fixture_ensemble" # --- §3.2 hash verification ----------------------------------------------------------- diff --git a/tests/test_wire_source_selection.py b/tests/test_wire_source_selection.py new file mode 100644 index 0000000..9545881 --- /dev/null +++ b/tests/test_wire_source_selection.py @@ -0,0 +1,127 @@ +"""Contract-mode Hop-A inbound (wire/source_selection.py) — dict-backed fake store +serving the golden fixture; selection policy, await-all-targets, declared identity.""" + +import json +from pathlib import Path + +import pytest + +from views_postprocessing.unfao.wire import source_selection as sel + +_FIX = Path(__file__).resolve().parent / "fixtures" / "wire_contract" +_MANIFEST_NAME = "fixture_run_0__lr_ged_sb__manifest.json" +_SHARD_NAME = "fixture_run_0__lr_ged_sb__m000543.tap.zip" + + +class FakeStore: + """The three-method store port over a dict of (metadata, bytes) records.""" + + def __init__(self, records): + # records: list of (file_id, metadata_dict, payload_bytes); newest LAST. + self._records = records + self.calls = [] + + def latest_file_id(self, filters): + self.calls.append(("latest_file_id", dict(filters))) + for file_id, meta, _ in reversed(self._records): + if all(meta.get(k) == v for k, v in filters.items()): + return file_id + return None + + def file_metadata(self, file_id): + return next(m for i, m, _ in self._records if i == file_id) + + def download(self, file_id): + return next(b for i, _, b in self._records if i == file_id) + + +def _fixture_store() -> FakeStore: + return FakeStore( + [ + ( + "f-shard", + {**sel.HOP_A_SHARD_FILTERS, "name": _SHARD_NAME}, + (_FIX / _SHARD_NAME).read_bytes(), + ), + ( + "f-manifest", + {**sel.HOP_A_MANIFEST_FILTERS, "name": _MANIFEST_NAME}, + (_FIX / _MANIFEST_NAME).read_bytes(), + ), + ] + ) + + +def test_happy_path_assembles_the_fixture_run(): + result = sel.fetch_run( + _fixture_store(), + expected_targets=["lr_ged_sb"], + expected_ensemble="fixture_ensemble", + ) + frame, headers = result["lr_ged_sb"] + assert frame.n_rows == 6 and frame.sample_count == 4 + assert headers[0]["run_id"] == "fixture_run_0" + + +def test_empty_store_refuses_loud(): + with pytest.raises(sel.SourceSelectionError, match="nothing has been published"): + sel.fetch_run( + FakeStore([]), expected_targets=["lr_ged_sb"], expected_ensemble="e" + ) + + +def test_missing_declared_target_refuses_loud(): + # The store has lr_ged_sb only; the delivery declares two targets (§4.2a). + with pytest.raises(sel.SourceSelectionError, match="lr_ged_ns.*awaiting all"): + sel.fetch_run( + _fixture_store(), + expected_targets=["lr_ged_sb", "lr_ged_ns"], + expected_ensemble="fixture_ensemble", + ) + + +def test_wrong_declared_ensemble_refuses_loud(): + with pytest.raises(sel.SourceSelectionError, match="rusty_bucket.*never falls back"): + sel.fetch_run( + _fixture_store(), + expected_targets=["lr_ged_sb"], + expected_ensemble="rusty_bucket", + ) + + +def test_manifested_shard_missing_from_store_refuses_loud(): + store = _fixture_store() + store._records = [r for r in store._records if r[0] != "f-shard"] + with pytest.raises(sel.SourceSelectionError, match="torn run"): + sel.fetch_run( + store, expected_targets=["lr_ged_sb"], expected_ensemble="fixture_ensemble" + ) + + +def test_manifest_name_content_mismatch_refuses_loud(): + # A second target's manifest name serving OTHER content: names are locators, + # content is identity (§3.3). + manifest = json.loads((_FIX / _MANIFEST_NAME).read_text()) + imposter_name = "fixture_run_0__lr_ged_ns__manifest.json" + store = _fixture_store() + store._records.insert( + 0, + ( + "f-imposter", + {**sel.HOP_A_MANIFEST_FILTERS, "name": imposter_name}, + json.dumps(manifest).encode(), # still claims target lr_ged_sb + ), + ) + with pytest.raises(sel.SourceSelectionError, match="content is identity"): + sel.fetch_run( + store, + expected_targets=["lr_ged_sb", "lr_ged_ns"], + expected_ensemble="fixture_ensemble", + ) + + +def test_selection_filters_are_golden_strings(): + # Pinned to pipeline-core's shipped publisher constants (re-verified 2026-07-20). + assert sel.HOP_A_SHARD_FILTERS == {"category": "forecast", "type": "sampled_forecast_shard"} + assert sel.HOP_A_MANIFEST_FILTERS == {"category": "forecast", "type": "sampled_forecast_manifest"} + assert sel.HOP_A_MANIFEST_NAME_TEMPLATE == "{run_id}__{target}__manifest.json" diff --git a/views_postprocessing/unfao/track_a_source.py b/views_postprocessing/unfao/track_a_source.py index 8dc9204..94a4e4d 100644 --- a/views_postprocessing/unfao/track_a_source.py +++ b/views_postprocessing/unfao/track_a_source.py @@ -105,15 +105,20 @@ def read_shard(shard_bytes: bytes, *, expected_sha256: str) -> tuple[PredictionF 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``. +def frames_for_target( + manifest: dict, shard_bytes_by_name: dict +) -> tuple[PredictionFrame, list[dict]]: + """A (run, target)'s verified shards → ``(PredictionFrame, headers)``. ``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. + The returned ``headers`` (manifest shard order) carry the producer-minted + provenance the sink passes through untouched (§10.2 — nothing is minted + downstream). """ - frames, months_seen = [], [] + frames, months_seen, headers = [], [], [] for entry in manifest["shards"]: name = entry["name"] if name not in shard_bytes_by_name: @@ -133,6 +138,7 @@ def frames_for_target(manifest: dict, shard_bytes_by_name: dict) -> PredictionFr f"{manifest['expected_cell_count']}." ) frames.append(frame) + headers.append(header) months_seen.append(int(header.get("time_id"))) if sorted(months_seen) != sorted(int(m) for m in manifest["expected_months"]): @@ -143,4 +149,4 @@ def frames_for_target(manifest: dict, shard_bytes_by_name: dict) -> PredictionFr 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) + return build_prediction_frame(values, time, unit), headers diff --git a/views_postprocessing/unfao/wire/source_selection.py b/views_postprocessing/unfao/wire/source_selection.py new file mode 100644 index 0000000..67c5487 --- /dev/null +++ b/views_postprocessing/unfao/wire/source_selection.py @@ -0,0 +1,102 @@ +"""Contract-mode Hop-A inbound: select and assemble the newest manifested run +(ADR-013 §3.2/§4.2a). + +The store is a **port** (DIP): any object with the three-method surface below — +production passes the manager's ``DatastoreModule``-backed wrapper, tests pass a +dict-backed fake. This module owns the *selection policy*; byte verification and +assembly stay in ``track_a_source`` (mechanism), the §6 policy gate stays at the +sink. + +Selection filters are pinned to pipeline-core's shipped publisher constants +(``sampled_forecast_publisher.py``: ``SHARD_TYPE``/``MANIFEST_TYPE``/ +``FORECAST_CATEGORY``/``MANIFEST_NAME_TEMPLATE`` — re-verified 2026-07-20) and +golden-string tested here; a drift on either side fails loud in tests, not in +production. + +Await-all-targets (§4.2a): the newest manifest names the run; the run is usable +only when EVERY declared target has a manifested leg. A missing target refuses +loud — a partial run is never assembled. Declared identity (§4.2a): each shard +header's ``provenance.ensemble`` must equal the launch declaration; mismatches +fail loud, never fall back. + +Store port surface:: + + latest_file_id(filters: dict) -> str | None + file_metadata(file_id: str) -> dict # carries at least "name" + download(file_id: str) -> bytes +""" + +from __future__ import annotations + +from views_postprocessing.unfao import track_a_source + +# pipeline-core's shipped vocabulary (§3.1/§3.2/§3.3) — golden-string tested. +HOP_A_SHARD_FILTERS = {"category": "forecast", "type": "sampled_forecast_shard"} +HOP_A_MANIFEST_FILTERS = {"category": "forecast", "type": "sampled_forecast_manifest"} +HOP_A_MANIFEST_NAME_TEMPLATE = "{run_id}__{target}__manifest.json" + + +class SourceSelectionError(ValueError): + """The store's contents do not satisfy the declared selection.""" + + +def fetch_run(store, *, expected_targets, expected_ensemble: str) -> dict: + """Assemble the newest fully-manifested run matching the declaration. + + Returns ``{target: (frame, headers)}`` for exactly ``expected_targets``. + """ + newest_id = store.latest_file_id(dict(HOP_A_MANIFEST_FILTERS)) + if newest_id is None: + raise SourceSelectionError( + f"no Hop-A manifest in the store (filters={HOP_A_MANIFEST_FILTERS}) — " + f"nothing has been published under the contract." + ) + newest = track_a_source.read_manifest(store.download(newest_id)) + run_id = newest["run_id"] + + per_target: dict = {} + for target in expected_targets: + manifest = _manifest_for(store, run_id=run_id, target=target, newest=newest) + shard_bytes = {} + for entry in manifest["shards"]: + shard_id = store.latest_file_id( + {**HOP_A_SHARD_FILTERS, "name": entry["name"]} + ) + if shard_id is None: + raise SourceSelectionError( + f"run {run_id!r}: manifested shard {entry['name']!r} not found in " + f"the store — a torn run must not be assembled." + ) + shard_bytes[entry["name"]] = store.download(shard_id) + frame, headers = track_a_source.frames_for_target(manifest, shard_bytes) + for header in headers: + found = header.get("provenance", {}).get("ensemble") + if found != expected_ensemble: + raise SourceSelectionError( + f"run {run_id!r}, target {target!r}: header declares ensemble " + f"{found!r}, the delivery was launched for {expected_ensemble!r} — " + f"identity mismatch fails loud, never falls back (§4.2a)." + ) + per_target[target] = (frame, headers) + return per_target + + +def _manifest_for(store, *, run_id: str, target: str, newest: dict) -> dict: + """The (run, target) manifest — the newest one if it already is that pair.""" + if newest["target"] == target: + return newest + name = HOP_A_MANIFEST_NAME_TEMPLATE.format(run_id=run_id, target=target) + file_id = store.latest_file_id({**HOP_A_MANIFEST_FILTERS, "name": name}) + if file_id is None: + raise SourceSelectionError( + f"run {run_id!r}: no manifest for declared target {target!r} — the run is " + f"incomplete for this delivery; awaiting all of its targets (§4.2a)." + ) + manifest = track_a_source.read_manifest(store.download(file_id)) + if manifest["run_id"] != run_id or manifest["target"] != target: + raise SourceSelectionError( + f"manifest named {name!r} declares (run={manifest['run_id']!r}, " + f"target={manifest['target']!r}) — names are locators, content is identity " + f"(§3.3); refusing the mismatch." + ) + return manifest