From f0be26b8ec75e12233a6a8f9d1700b67c3853a55 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Wed, 15 Jul 2026 14:18:07 +0800 Subject: [PATCH 1/2] feat: add weekly source contract foundation Co-Authored-By: Codex --- .../weekly_contract.py | 159 ++++++++++++++++++ tests/test_weekly_contract.py | 101 +++++++++++ 2 files changed, 260 insertions(+) create mode 100644 src/political_event_tracking_research/weekly_contract.py create mode 100644 tests/test_weekly_contract.py diff --git a/src/political_event_tracking_research/weekly_contract.py b/src/political_event_tracking_research/weekly_contract.py new file mode 100644 index 0000000..970780e --- /dev/null +++ b/src/political_event_tracking_research/weekly_contract.py @@ -0,0 +1,159 @@ +"""Pure producer-owned contract for complete official PERT weekly inputs.""" +from __future__ import annotations + +import hashlib +import json +import re +from collections.abc import Mapping +from dataclasses import dataclass +from datetime import date, datetime, timedelta, timezone +from pathlib import PurePosixPath + +SCHEMA_VERSION = "1" +CONTRACT_VERSION = "political_event_weekly.v1" +CADENCE = "weekly" +_DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$") +_TIMESTAMP_RE = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d{1,6})?Z$") +_SHA256_RE = re.compile(r"^[0-9a-f]{64}$") +_SHA1_RE = re.compile(r"^[0-9a-f]{40}$") +_KEYS = frozenset({ + "schema_version", "contract_version", "cadence", "as_of", "period_start", "period_end_exclusive", + "generated_at", "run_mode", "producer_ref", "source_provenance", "source_artifacts", "feed_status", +}) +_FEED_KEYS = frozenset({"feed_count", "successful_feed_count", "failed_feed_count", "stale_feed_count", "missing_feed_count", "complete"}) +_ARTIFACT_KEYS = frozenset({"path", "sha256", "row_count"}) + + +class WeeklyContractError(ValueError): + def __init__(self, code: str) -> None: + self.code = code + super().__init__(code) + + +@dataclass(frozen=True, slots=True) +class WeeklySourceArtifact: + path: str + sha256: str + row_count: int + + +@dataclass(frozen=True, slots=True) +class WeeklyFeedStatus: + feed_count: int + successful_feed_count: int + failed_feed_count: int + stale_feed_count: int + missing_feed_count: int + + +@dataclass(frozen=True, slots=True) +class WeeklySourceContract: + as_of: date + period_start: date + period_end_exclusive: date + generated_at: datetime + run_mode: str + producer_ref: str + source_provenance: str + source_artifacts: tuple[WeeklySourceArtifact, ...] + feed_status: WeeklyFeedStatus + + +def _invalid(code: str) -> WeeklyContractError: + return WeeklyContractError(code) + + +def _date(value: object) -> date: + if type(value) is not str or not _DATE_RE.fullmatch(value): + raise _invalid("date_invalid") + try: + return date.fromisoformat(value) + except ValueError: + raise _invalid("date_invalid") from None + + +def _generated_at(value: object) -> datetime: + if type(value) is not str or not _TIMESTAMP_RE.fullmatch(value): + raise _invalid("generated_at_invalid") + try: + parsed = datetime.fromisoformat(value[:-1] + "+00:00") + except ValueError: + raise _invalid("generated_at_invalid") from None + if parsed.tzinfo != timezone.utc: + raise _invalid("generated_at_invalid") + return parsed + + +def _artifact(value: object) -> WeeklySourceArtifact: + if not isinstance(value, Mapping) or set(value) != _ARTIFACT_KEYS: + raise _invalid("source_artifact_invalid") + path = value["path"] + if type(path) is not str or not path or PurePosixPath(path).is_absolute() or ".." in PurePosixPath(path).parts or "\\" in path: + raise _invalid("source_artifact_invalid") + digest = value["sha256"] + row_count = value["row_count"] + if type(digest) is not str or not _SHA256_RE.fullmatch(digest) or type(row_count) is not int or row_count < 0: + raise _invalid("source_artifact_invalid") + return WeeklySourceArtifact(path, digest, row_count) + + +def _feed_status(value: object) -> WeeklyFeedStatus: + if not isinstance(value, Mapping) or set(value) != _FEED_KEYS or value.get("complete") is not True: + raise _invalid("feed_status_invalid") + values = [value[key] for key in _FEED_KEYS if key != "complete"] + if any(type(item) is not int or item < 0 for item in values): + raise _invalid("feed_status_invalid") + status = WeeklyFeedStatus(*(value[key] for key in ("feed_count", "successful_feed_count", "failed_feed_count", "stale_feed_count", "missing_feed_count"))) + if status.feed_count <= 0 or status.successful_feed_count != status.feed_count or any((status.failed_feed_count, status.stale_feed_count, status.missing_feed_count)): + raise _invalid("feed_status_incomplete") + return status + + +def parse_weekly_contract(value: Mapping[str, object]) -> WeeklySourceContract: + if not isinstance(value, Mapping) or set(value) != _KEYS: + raise _invalid("contract_shape_invalid") + if value["schema_version"] != SCHEMA_VERSION or value["contract_version"] != CONTRACT_VERSION or value["cadence"] != CADENCE: + raise _invalid("contract_version_invalid") + if type(value["run_mode"]) is not str or value["run_mode"] not in {"scheduled", "manual"}: + raise _invalid("run_mode_invalid") + producer_ref = value["producer_ref"] + provenance = value["source_provenance"] + if type(producer_ref) is not str or not _SHA1_RE.fullmatch(producer_ref) or type(provenance) is not str or not provenance: + raise _invalid("provenance_invalid") + as_of = _date(value["as_of"]) + start = _date(value["period_start"]) + end = _date(value["period_end_exclusive"]) + if start.weekday() != 0 or end != start + timedelta(days=7) or as_of != end - timedelta(days=1): + raise _invalid("period_invalid") + generated_at = _generated_at(value["generated_at"]) + if generated_at < datetime.combine(end, datetime.min.time(), timezone.utc): + raise _invalid("generated_at_before_period_end") + artifacts_value = value["source_artifacts"] + if not isinstance(artifacts_value, list) or not artifacts_value: + raise _invalid("source_artifact_invalid") + artifacts = tuple(sorted((_artifact(item) for item in artifacts_value), key=lambda item: item.path)) + if len({item.path for item in artifacts}) != len(artifacts): + raise _invalid("source_artifact_duplicate") + return WeeklySourceContract(as_of, start, end, generated_at, value["run_mode"], producer_ref, provenance, artifacts, _feed_status(value["feed_status"])) + + +def serialize_weekly_contract(contract: WeeklySourceContract) -> bytes: + if not isinstance(contract, WeeklySourceContract): + raise _invalid("contract_type_invalid") + try: + payload = { + "schema_version": SCHEMA_VERSION, "contract_version": CONTRACT_VERSION, "cadence": CADENCE, + "as_of": contract.as_of.isoformat(), "period_start": contract.period_start.isoformat(), + "period_end_exclusive": contract.period_end_exclusive.isoformat(), + "generated_at": contract.generated_at.isoformat(timespec="microseconds").replace("+00:00", "Z"), + "run_mode": contract.run_mode, "producer_ref": contract.producer_ref, "source_provenance": contract.source_provenance, + "source_artifacts": [{"path": item.path, "sha256": item.sha256, "row_count": item.row_count} for item in contract.source_artifacts], + "feed_status": {"feed_count": contract.feed_status.feed_count, "successful_feed_count": contract.feed_status.successful_feed_count, "failed_feed_count": 0, "stale_feed_count": 0, "missing_feed_count": 0, "complete": True}, + } + except (AttributeError, TypeError, ValueError, OverflowError): + raise _invalid("contract_invalid") from None + parse_weekly_contract(payload) + try: + return json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8") + except (TypeError, UnicodeError, ValueError): + raise _invalid("serialization_invalid") from None diff --git a/tests/test_weekly_contract.py b/tests/test_weekly_contract.py new file mode 100644 index 0000000..5cbaa42 --- /dev/null +++ b/tests/test_weekly_contract.py @@ -0,0 +1,101 @@ +from __future__ import annotations + +from datetime import date + +import pytest + +from political_event_tracking_research.weekly_contract import ( + WeeklyContractError, + parse_weekly_contract, + serialize_weekly_contract, +) + + +def 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", + "source_artifacts": [ + {"path": "data/live/political_events.csv", "sha256": "b" * 64, "row_count": 11}, + ], + "feed_status": { + "feed_count": 9, + "successful_feed_count": 9, + "failed_feed_count": 0, + "stale_feed_count": 0, + "missing_feed_count": 0, + "complete": True, + }, + } + value.update(overrides) + return value + + +def test_valid_weekly_contract_round_trips_deterministically(): + contract = parse_weekly_contract(payload()) + encoded = serialize_weekly_contract(contract) + assert encoded == serialize_weekly_contract(parse_weekly_contract(__import__("json").loads(encoded))) + assert contract.as_of == date(2026, 7, 12) + + +@pytest.mark.parametrize("field,value", [ + ("schema_version", "2"), + ("contract_version", "political_event_weekly.v2"), + ("cadence", "daily"), + ("as_of", "2026-07-13"), + ("period_start", "2026-07-07"), + ("period_end_exclusive", "2026-07-14"), + ("generated_at", "2026-07-12T23:59:59Z"), + ("generated_at", "2026-07-13T00:00:00+08:00"), +]) +def test_period_and_time_contract_fail_closed(field, value): + with pytest.raises(WeeklyContractError): + parse_weekly_contract(payload(**{field: value})) + + +def test_scheduled_and_manual_are_explicit(): + assert parse_weekly_contract(payload(run_mode="scheduled")).run_mode == "scheduled" + with pytest.raises(WeeklyContractError): + parse_weekly_contract(payload(run_mode="")) + with pytest.raises(WeeklyContractError): + parse_weekly_contract({key: value for key, value in payload().items() if key != "as_of"}) + + +@pytest.mark.parametrize("status", [ + {"feed_count": 9, "successful_feed_count": 8, "failed_feed_count": 1, "stale_feed_count": 0, "missing_feed_count": 0, "complete": True}, + {"feed_count": 9, "successful_feed_count": 9, "failed_feed_count": 0, "stale_feed_count": 1, "missing_feed_count": 0, "complete": True}, + {"feed_count": 9, "successful_feed_count": 9, "failed_feed_count": 0, "stale_feed_count": 0, "missing_feed_count": 0, "complete": False}, +]) +def test_partial_stale_or_incomplete_feed_status_fails_closed(status): + with pytest.raises(WeeklyContractError, match="feed_status"): + parse_weekly_contract(payload(feed_status=status)) + + +def test_artifacts_are_sorted_and_duplicate_or_unsafe_inputs_fail_closed(): + items = [ + {"path": "z.csv", "sha256": "c" * 64, "row_count": 1}, + {"path": "a.csv", "sha256": "d" * 64, "row_count": 2}, + ] + contract = parse_weekly_contract(payload(source_artifacts=items)) + assert [item.path for item in contract.source_artifacts] == ["a.csv", "z.csv"] + with pytest.raises(WeeklyContractError): + parse_weekly_contract(payload(source_artifacts=[items[0], items[0]])) + with pytest.raises(WeeklyContractError): + parse_weekly_contract(payload(source_artifacts=[{"path": "../secret", "sha256": "c" * 64, "row_count": 1}])) + + +def test_unknown_fields_and_wire_type_confusion_fail_closed(): + with pytest.raises(WeeklyContractError): + parse_weekly_contract(payload(extra=True)) + with pytest.raises(WeeklyContractError): + parse_weekly_contract(payload(schema_version=1)) + with pytest.raises(WeeklyContractError): + parse_weekly_contract(payload(feed_status={**payload()["feed_status"], "feed_count": True})) From 315a389a031e3d6d8832bad2460015e1dbb19a7b Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Wed, 15 Jul 2026 14:31:48 +0800 Subject: [PATCH 2/2] fix: tighten weekly source contract invariants Co-Authored-By: Codex --- .../weekly_contract.py | 22 ++++++++++----- tests/test_weekly_contract.py | 27 +++++++++++++++++++ 2 files changed, 42 insertions(+), 7 deletions(-) diff --git a/src/political_event_tracking_research/weekly_contract.py b/src/political_event_tracking_research/weekly_contract.py index 970780e..7ae0ba5 100644 --- a/src/political_event_tracking_research/weekly_contract.py +++ b/src/political_event_tracking_research/weekly_contract.py @@ -4,6 +4,7 @@ import hashlib import json import re +import unicodedata from collections.abc import Mapping from dataclasses import dataclass from datetime import date, datetime, timedelta, timezone @@ -12,6 +13,7 @@ SCHEMA_VERSION = "1" CONTRACT_VERSION = "political_event_weekly.v1" CADENCE = "weekly" +MAX_SAFE_JSON_INTEGER = 2**53 - 1 _DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$") _TIMESTAMP_RE = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d{1,6})?Z$") _SHA256_RE = re.compile(r"^[0-9a-f]{64}$") @@ -44,6 +46,7 @@ class WeeklyFeedStatus: failed_feed_count: int stale_feed_count: int missing_feed_count: int + complete: bool @dataclass(frozen=True, slots=True) @@ -88,23 +91,28 @@ def _artifact(value: object) -> WeeklySourceArtifact: if not isinstance(value, Mapping) or set(value) != _ARTIFACT_KEYS: raise _invalid("source_artifact_invalid") path = value["path"] - if type(path) is not str or not path or PurePosixPath(path).is_absolute() or ".." in PurePosixPath(path).parts or "\\" in path: + canonical = PurePosixPath(path) if type(path) is str else None + if ( + type(path) is not str or not path or not path.isascii() or unicodedata.normalize("NFC", path) != path + or canonical is None or canonical.is_absolute() or path != str(canonical) or path.endswith("/") + or "//" in path or any(part in {"", ".", ".."} for part in canonical.parts) or "\\" in path + ): raise _invalid("source_artifact_invalid") digest = value["sha256"] row_count = value["row_count"] - if type(digest) is not str or not _SHA256_RE.fullmatch(digest) or type(row_count) is not int or row_count < 0: + if type(digest) is not str or not _SHA256_RE.fullmatch(digest) or type(row_count) is not int or not 0 <= row_count <= MAX_SAFE_JSON_INTEGER: raise _invalid("source_artifact_invalid") return WeeklySourceArtifact(path, digest, row_count) def _feed_status(value: object) -> WeeklyFeedStatus: - if not isinstance(value, Mapping) or set(value) != _FEED_KEYS or value.get("complete") is not True: + if not isinstance(value, Mapping) or set(value) != _FEED_KEYS or type(value.get("complete")) is not bool: raise _invalid("feed_status_invalid") values = [value[key] for key in _FEED_KEYS if key != "complete"] - if any(type(item) is not int or item < 0 for item in values): + if any(type(item) is not int or not 0 <= item <= MAX_SAFE_JSON_INTEGER for item in values): raise _invalid("feed_status_invalid") - status = WeeklyFeedStatus(*(value[key] for key in ("feed_count", "successful_feed_count", "failed_feed_count", "stale_feed_count", "missing_feed_count"))) - if status.feed_count <= 0 or status.successful_feed_count != status.feed_count or any((status.failed_feed_count, status.stale_feed_count, status.missing_feed_count)): + status = WeeklyFeedStatus(*(value[key] for key in ("feed_count", "successful_feed_count", "failed_feed_count", "stale_feed_count", "missing_feed_count")), value["complete"]) + if status.feed_count <= 0 or status.successful_feed_count != status.feed_count or any((status.failed_feed_count, status.stale_feed_count, status.missing_feed_count)) or not status.complete: raise _invalid("feed_status_incomplete") return status @@ -148,7 +156,7 @@ def serialize_weekly_contract(contract: WeeklySourceContract) -> bytes: "generated_at": contract.generated_at.isoformat(timespec="microseconds").replace("+00:00", "Z"), "run_mode": contract.run_mode, "producer_ref": contract.producer_ref, "source_provenance": contract.source_provenance, "source_artifacts": [{"path": item.path, "sha256": item.sha256, "row_count": item.row_count} for item in contract.source_artifacts], - "feed_status": {"feed_count": contract.feed_status.feed_count, "successful_feed_count": contract.feed_status.successful_feed_count, "failed_feed_count": 0, "stale_feed_count": 0, "missing_feed_count": 0, "complete": True}, + "feed_status": {"feed_count": contract.feed_status.feed_count, "successful_feed_count": contract.feed_status.successful_feed_count, "failed_feed_count": contract.feed_status.failed_feed_count, "stale_feed_count": contract.feed_status.stale_feed_count, "missing_feed_count": contract.feed_status.missing_feed_count, "complete": contract.feed_status.complete}, } except (AttributeError, TypeError, ValueError, OverflowError): raise _invalid("contract_invalid") from None diff --git a/tests/test_weekly_contract.py b/tests/test_weekly_contract.py index 5cbaa42..6c2d5d8 100644 --- a/tests/test_weekly_contract.py +++ b/tests/test_weekly_contract.py @@ -69,6 +69,14 @@ def test_scheduled_and_manual_are_explicit(): parse_weekly_contract({key: value for key, value in payload().items() if key != "as_of"}) +def test_real_feed_counters_round_trip_without_serializer_defaults(): + status = {"feed_count": 12, "successful_feed_count": 12, "failed_feed_count": 0, "stale_feed_count": 0, "missing_feed_count": 0, "complete": True} + contract = parse_weekly_contract(payload(feed_status=status)) + encoded = serialize_weekly_contract(contract) + assert b'"feed_count":12' in encoded + assert parse_weekly_contract(__import__("json").loads(encoded)).feed_status == contract.feed_status + + @pytest.mark.parametrize("status", [ {"feed_count": 9, "successful_feed_count": 8, "failed_feed_count": 1, "stale_feed_count": 0, "missing_feed_count": 0, "complete": True}, {"feed_count": 9, "successful_feed_count": 9, "failed_feed_count": 0, "stale_feed_count": 1, "missing_feed_count": 0, "complete": True}, @@ -90,6 +98,25 @@ def test_artifacts_are_sorted_and_duplicate_or_unsafe_inputs_fail_closed(): parse_weekly_contract(payload(source_artifacts=[items[0], items[0]])) with pytest.raises(WeeklyContractError): parse_weekly_contract(payload(source_artifacts=[{"path": "../secret", "sha256": "c" * 64, "row_count": 1}])) + for alias in ("a//b.csv", "a/b.csv/", "./a.csv", "a/./b.csv", "a\\b.csv", "é.csv"): + with pytest.raises(WeeklyContractError): + parse_weekly_contract(payload(source_artifacts=[{"path": alias, "sha256": "c" * 64, "row_count": 1}])) + + +@pytest.mark.parametrize("field", ["row_count", "feed_count", "successful_feed_count"]) +def test_wire_integers_use_safe_json_range_and_reject_bool(field): + artifact = {"path": "data/live/political_events.csv", "sha256": "b" * 64, "row_count": 11} + status = payload()["feed_status"] + if field == "row_count": + with pytest.raises(WeeklyContractError): + parse_weekly_contract(payload(source_artifacts=[{**artifact, "row_count": 2**53}])) + with pytest.raises(WeeklyContractError): + parse_weekly_contract(payload(source_artifacts=[{**artifact, "row_count": True}])) + else: + with pytest.raises(WeeklyContractError): + parse_weekly_contract(payload(feed_status={**status, field: 2**53})) + with pytest.raises(WeeklyContractError): + parse_weekly_contract(payload(feed_status={**status, field: True})) def test_unknown_fields_and_wire_type_confusion_fail_closed():