diff --git a/tests/test_wire_run_manifest.py b/tests/test_wire_run_manifest.py new file mode 100644 index 0000000..d8d3d89 --- /dev/null +++ b/tests/test_wire_run_manifest.py @@ -0,0 +1,61 @@ +"""§4.2 run manifest builder (wire/run_manifest.py) — byte parity with the fixture.""" + +import hashlib +from pathlib import Path + +import pytest + +from views_postprocessing.unfao.wire import run_manifest + +_FIX = Path(__file__).resolve().parent / "fixtures" / "wire_contract" + + +def _sha(name: str) -> str: + return hashlib.sha256((_FIX / name).read_bytes()).hexdigest() + + +def _fixture_kwargs() -> dict: + return dict( + run_id="fixture_run_0", + targets=["lr_ged_sb"], + shard_records=[ + { + "name": "fixture_run_0__lr_ged_sb__m000543.arrow.parquet", + "target": "lr_ged_sb", + "time_id": 543, + "sha256": _sha("fixture_run_0__lr_ged_sb__m000543.arrow.parquet"), + } + ], + expected_months=[543], + expected_cell_count=6, + sidecar_record={ + "name": "fixture_run_0__sidecar.parquet", + "sha256": _sha("fixture_run_0__sidecar.parquet"), + }, + ) + + +def test_byte_parity_with_the_fixture_run_manifest(): + built = run_manifest.build_run_manifest(**_fixture_kwargs()) + assert built == (_FIX / "fixture_run_0__manifest.json").read_bytes() + + +def test_shard_record_with_extra_key_raises(): + kwargs = _fixture_kwargs() + kwargs["shard_records"][0]["file_id"] = "nope" # the field §3.2 corrected away + with pytest.raises(run_manifest.RunManifestError, match="file_id"): + run_manifest.build_run_manifest(**kwargs) + + +def test_sidecar_record_missing_hash_raises(): + kwargs = _fixture_kwargs() + kwargs["sidecar_record"] = {"name": "x"} + with pytest.raises(run_manifest.RunManifestError, match="sha256"): + run_manifest.build_run_manifest(**kwargs) + + +def test_shard_record_key_order_is_pinned_regardless_of_input_order(): + kwargs = _fixture_kwargs() + kwargs["shard_records"][0] = dict(reversed(list(kwargs["shard_records"][0].items()))) + built = run_manifest.build_run_manifest(**kwargs) + assert built == (_FIX / "fixture_run_0__manifest.json").read_bytes() diff --git a/views_postprocessing/unfao/wire/run_manifest.py b/views_postprocessing/unfao/wire/run_manifest.py new file mode 100644 index 0000000..5e8c541 --- /dev/null +++ b/views_postprocessing/unfao/wire/run_manifest.py @@ -0,0 +1,58 @@ +"""The §4.2 run manifest builder — the run's single commit marker (ADR-013). + +ONE manifest per run, spanning all targets, serialized exactly as the §10 fixture +pins it (key order + ``indent=2``): uploaded LAST, after every shard AND the +sidecar, so a run is either whole or invisible. The fields are the §4.2 table: +shard entries carry ``name``/``target``/``time_id``/``sha256`` (hash of the +complete file bytes); ``expected_cell_count`` is per-shard N (§3.2's scope +ruling); the sidecar object is the Erratum-E1 home of the sidecar hash. + +Builders declare, they never inspect: every value here arrives from the caller +(the sink), which in turn passed it through from Hop-A headers or computed it +from bytes it just wrote. +""" + +from __future__ import annotations + +import json + +from views_postprocessing.unfao.wire.header import CONTRACT_VERSION + + +class RunManifestError(ValueError): + """The declared manifest inputs violate the §4.2 schema.""" + + +_SHARD_KEYS = ("name", "target", "time_id", "sha256") +_SIDECAR_KEYS = ("name", "sha256") + + +def build_run_manifest( + *, + run_id: str, + targets: list[str], + shard_records: list[dict], + expected_months: list[int], + expected_cell_count: int, + sidecar_record: dict, +) -> bytes: + """§4.2 run-manifest JSON bytes, byte-stable against the §10 fixture.""" + for record in shard_records: + if set(record) != set(_SHARD_KEYS): + raise RunManifestError( + f"shard record keys {sorted(record)} != required {sorted(_SHARD_KEYS)} (§4.2)." + ) + if set(sidecar_record) != set(_SIDECAR_KEYS): + raise RunManifestError( + f"sidecar record keys {sorted(sidecar_record)} != required {sorted(_SIDECAR_KEYS)} (§4.2/E1)." + ) + manifest = { + "contract_version": CONTRACT_VERSION, + "run_id": run_id, + "targets": list(targets), + "shards": [{key: record[key] for key in _SHARD_KEYS} for record in shard_records], + "expected_months": list(expected_months), + "expected_cell_count": expected_cell_count, + "sidecar": {key: sidecar_record[key] for key in _SIDECAR_KEYS}, + } + return json.dumps(manifest, indent=2).encode()