diff --git a/src/political_event_tracking_research/weekly_manifest.py b/src/political_event_tracking_research/weekly_manifest.py new file mode 100644 index 0000000..4078bea --- /dev/null +++ b/src/political_event_tracking_research/weekly_manifest.py @@ -0,0 +1,86 @@ +"""Pure deterministic manifest integration for ``political_event_weekly.v1``.""" +from __future__ import annotations + +import json +from collections.abc import Mapping +from pathlib import Path + +from .weekly_contract import WeeklyContractError, WeeklySourceContract, parse_weekly_contract, serialize_weekly_contract + +MANIFEST_TYPE = "political_event_weekly_manifest" +_MANIFEST_KEYS = frozenset({"manifest_type", "contract"}) + + +def build_weekly_manifest(contract: WeeklySourceContract) -> dict[str, object]: + if not isinstance(contract, WeeklySourceContract): + raise WeeklyContractError("manifest_contract_invalid") + try: + contract_payload = json.loads(serialize_weekly_contract(contract)) + except (TypeError, ValueError, UnicodeError): + raise WeeklyContractError("manifest_contract_invalid") from None + return {"manifest_type": MANIFEST_TYPE, "contract": contract_payload} + + +def parse_weekly_manifest(value: Mapping[str, object]) -> WeeklySourceContract: + if not isinstance(value, Mapping) or set(value) != _MANIFEST_KEYS or value.get("manifest_type") != MANIFEST_TYPE: + raise WeeklyContractError("manifest_shape_invalid") + contract = value.get("contract") + if not isinstance(contract, Mapping): + raise WeeklyContractError("manifest_contract_invalid") + return parse_weekly_contract(contract) + + +def parse_weekly_manifest_bytes(wire: bytes) -> WeeklySourceContract: + if type(wire) is not bytes: + raise WeeklyContractError("manifest_wire_invalid") + + def pairs(items: list[tuple[str, object]]) -> dict[str, object]: + result: dict[str, object] = {} + for key, item in items: + if key in result: + raise WeeklyContractError("manifest_duplicate_key") + result[key] = item + return result + + try: + value = json.loads(wire.decode("utf-8"), object_pairs_hook=pairs) + except WeeklyContractError: + raise + except (UnicodeError, json.JSONDecodeError, TypeError, ValueError, RecursionError): + raise WeeklyContractError("manifest_wire_invalid") from None + if not isinstance(value, Mapping): + raise WeeklyContractError("manifest_shape_invalid") + contract = parse_weekly_manifest(value) + if serialize_weekly_manifest(contract) != wire: + raise WeeklyContractError("manifest_noncanonical") + return contract + + +def validate_weekly_manifest(value: Mapping[str, object], expected: WeeklySourceContract) -> WeeklySourceContract: + if not isinstance(expected, WeeklySourceContract): + raise WeeklyContractError("manifest_expected_invalid") + parsed = parse_weekly_manifest(value) + if parsed != expected: + raise WeeklyContractError("manifest_contract_mismatch") + if build_weekly_manifest(parsed) != dict(value): + raise WeeklyContractError("manifest_noncanonical") + return parsed + + +def serialize_weekly_manifest(contract: WeeklySourceContract) -> bytes: + payload = build_weekly_manifest(contract) + try: + return json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8") + except (TypeError, ValueError, UnicodeError): + raise WeeklyContractError("manifest_serialization_invalid") from None + + +def write_weekly_manifest(contract: WeeklySourceContract, output_path: str | Path) -> Path: + content = serialize_weekly_manifest(contract) + output = Path(output_path) + try: + output.parent.mkdir(parents=True, exist_ok=True) + output.write_bytes(content) + except (OSError, TypeError, ValueError): + raise WeeklyContractError("manifest_write_invalid") from None + return output diff --git a/tests/test_weekly_manifest.py b/tests/test_weekly_manifest.py new file mode 100644 index 0000000..bc6478e --- /dev/null +++ b/tests/test_weekly_manifest.py @@ -0,0 +1,99 @@ +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from political_event_tracking_research.weekly_contract import WeeklyContractError, parse_weekly_contract +from political_event_tracking_research.weekly_manifest import ( + MANIFEST_TYPE, + build_weekly_manifest, + parse_weekly_manifest, + parse_weekly_manifest_bytes, + serialize_weekly_manifest, + validate_weekly_manifest, + write_weekly_manifest, +) + + +def contract_payload(**overrides: object) -> dict[str, object]: + value: dict[str, object] = { + "schema_version": "1", "contract_version": "political_event_weekly.v1", "cadence": "weekly", + "as_of": "2026-07-12", "period_start": "2026-07-06", "period_end_exclusive": "2026-07-13", + "generated_at": "2026-07-13T00:00:00.123456Z", "run_mode": "manual", "producer_ref": "a" * 40, + "source_provenance": "official_political_event_tracking_research_v1", + "source_artifacts": [{"path": "data/live/political_events.csv", "sha256": "b" * 64, "row_count": 11}], + "feed_status": {"feed_count": 12, "successful_feed_count": 12, "failed_feed_count": 0, "stale_feed_count": 0, "missing_feed_count": 0, "complete": True}, + } + value.update(overrides) + return value + + +def test_weekly_manifest_is_deterministic_and_round_trips(): + contract = parse_weekly_contract(contract_payload()) + manifest = build_weekly_manifest(contract) + assert manifest["manifest_type"] == MANIFEST_TYPE + encoded = serialize_weekly_manifest(contract) + assert encoded == serialize_weekly_manifest(parse_weekly_manifest(json.loads(encoded))) + assert parse_weekly_manifest(manifest) == contract + assert parse_weekly_manifest_bytes(encoded) == contract + + +@pytest.mark.parametrize("wire", [ + lambda encoded: b" " + encoded, + lambda encoded: encoded.replace(b'"contract":', b'"contract" :'), + lambda encoded: encoded.replace(b'"manifest_type":', b'"manifest_type":"x","manifest_type":'), +]) +def test_manifest_wire_must_be_exact_canonical_bytes(wire): + contract = parse_weekly_contract(contract_payload()) + encoded = serialize_weekly_manifest(contract) + with pytest.raises(WeeklyContractError): + parse_weekly_manifest_bytes(wire(encoded)) + + +@pytest.mark.parametrize("field,value", [ + ("as_of", "2026-07-13"), + ("generated_at", "2026-07-12T23:59:59Z"), + ("producer_ref", "c" * 40), + ("source_provenance", "official_other_v1"), + ("source_artifacts", [{"path": "data/live/political_events.csv", "sha256": "c" * 64, "row_count": 12}]), +]) +def test_manifest_contract_tamper_is_rejected(field, value): + contract = parse_weekly_contract(contract_payload()) + manifest = build_weekly_manifest(contract) + manifest["contract"][field] = value + with pytest.raises(WeeklyContractError): + validate_weekly_manifest(manifest, contract) + + +def test_mapping_with_alias_or_unknown_shape_fails_closed(): + contract = parse_weekly_contract(contract_payload()) + manifest = build_weekly_manifest(contract) + manifest["contract"]["generatedAt"] = manifest["contract"].pop("generated_at") + with pytest.raises(WeeklyContractError): + validate_weekly_manifest(manifest, contract) + + +def test_manifest_feed_partial_and_shape_tamper_fail_closed(): + contract = parse_weekly_contract(contract_payload()) + manifest = build_weekly_manifest(contract) + manifest["contract"]["feed_status"]["failed_feed_count"] = 1 + with pytest.raises(WeeklyContractError): + parse_weekly_manifest(manifest) + manifest = build_weekly_manifest(contract) + manifest["extra"] = True + with pytest.raises(WeeklyContractError): + parse_weekly_manifest(manifest) + + +def test_write_validates_before_creating_output(tmp_path: Path): + contract = parse_weekly_contract(contract_payload()) + output = tmp_path / "manifest.json" + write_weekly_manifest(contract, output) + assert json.loads(output.read_text(encoding="utf-8"))["manifest_type"] == MANIFEST_TYPE + + invalid = tmp_path / "invalid.json" + with pytest.raises(WeeklyContractError): + write_weekly_manifest(object(), invalid) # type: ignore[arg-type] + assert not invalid.exists()