feat(wire): S5/#110 — §4.2 run manifest builder (commit marker, byte-stable)

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()

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()