feat: reslice weekly contract path invariants

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 json
+import re
+import unicodedata
+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"
+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}$")
+_SHA1_RE = re.compile(r"^[0-9a-f]{40}$")
+_PATH_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/-]*$")
+_PROVENANCE_RE = re.compile(r"^[a-z][a-z0-9]*(?:_[a-z0-9]+)*$")
+_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)
+
+
+def _invalid(code: str) -> WeeklyContractError:
+    return WeeklyContractError(code)
+
+
+def _safe_int(value: object, code: str) -> int:
+    if type(value) is not int or not 0 <= value <= MAX_SAFE_JSON_INTEGER:
+        raise _invalid(code)
+    return value
+
+
+def _canonical_path(value: object) -> str:
+    if type(value) is not str or not value or not value.isascii() or unicodedata.normalize("NFC", value) != value or not _PATH_RE.fullmatch(value):
+        raise _invalid("source_artifact_path_invalid")
+    path = PurePosixPath(value)
+    if path.is_absolute() or value != str(path) or value.endswith("/") or "//" in value or any(part in {"", ".", ".."} for part in path.parts):
+        raise _invalid("source_artifact_path_invalid")
+    return value
+
+
+@dataclass(frozen=True, slots=True)
+class WeeklySourceArtifact:
+    path: str
+    sha256: str
+    row_count: int
+
+    def __post_init__(self) -> None:
+        _canonical_path(self.path)
+        if type(self.sha256) is not str or not _SHA256_RE.fullmatch(self.sha256):
+            raise _invalid("source_artifact_invalid")
+        _safe_int(self.row_count, "source_artifact_invalid")
+
+
+@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
+    complete: bool
+
+    def __post_init__(self) -> None:
+        counts = (self.feed_count, self.successful_feed_count, self.failed_feed_count, self.stale_feed_count, self.missing_feed_count)
+        for count in counts:
+            _safe_int(count, "feed_status_invalid")
+        if type(self.complete) is not bool:
+            raise _invalid("feed_status_invalid")
+        if self.feed_count <= 0 or self.successful_feed_count != self.feed_count or any((self.failed_feed_count, self.stale_feed_count, self.missing_feed_count)) or not self.complete:
+            raise _invalid("feed_status_incomplete")
+
+
+@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 __post_init__(self) -> None:
+        if type(self.as_of) is not date or type(self.period_start) is not date or type(self.period_end_exclusive) is not date or self.period_start.weekday() != 0 or self.period_end_exclusive != self.period_start + timedelta(days=7) or self.as_of != self.period_end_exclusive - timedelta(days=1):
+            raise _invalid("period_invalid")
+        if type(self.generated_at) is not datetime or self.generated_at.tzinfo != timezone.utc or self.generated_at < datetime.combine(self.period_end_exclusive, datetime.min.time(), timezone.utc):
+            raise _invalid("generated_at_invalid")
+        if self.run_mode not in {"scheduled", "manual"} or type(self.producer_ref) is not str or not _SHA1_RE.fullmatch(self.producer_ref) or type(self.source_provenance) is not str or not _PROVENANCE_RE.fullmatch(self.source_provenance):
+            raise _invalid("provenance_invalid")
+        if not isinstance(self.source_artifacts, tuple) or not self.source_artifacts or any(not isinstance(item, WeeklySourceArtifact) for item in self.source_artifacts):
+            raise _invalid("source_artifact_invalid")
+        ordered = tuple(sorted(self.source_artifacts, key=lambda item: item.path))
+        if len({item.path for item in ordered}) != len(ordered):
+            raise _invalid("source_artifact_duplicate")
+        object.__setattr__(self, "source_artifacts", ordered)
+        if not isinstance(self.feed_status, WeeklyFeedStatus):
+            raise _invalid("feed_status_invalid")
+
+
+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:
+        return datetime.fromisoformat(value[:-1] + "+00:00")
+    except ValueError:
+        raise _invalid("generated_at_invalid") from None
+
+
+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")
+    source_artifacts = value["source_artifacts"]
+    if not isinstance(source_artifacts, list) or not source_artifacts:
+        raise _invalid("source_artifact_invalid")
+    artifacts_list: list[WeeklySourceArtifact] = []
+    for item in source_artifacts:
+        if not isinstance(item, Mapping) or set(item) != _ARTIFACT_KEYS:
+            raise _invalid("source_artifact_invalid")
+        artifacts_list.append(WeeklySourceArtifact(item["path"], item["sha256"], item["row_count"]))
+    artifacts = tuple(artifacts_list)
+    status_value = value["feed_status"]
+    if not isinstance(status_value, Mapping) or set(status_value) != _FEED_KEYS:
+        raise _invalid("feed_status_invalid")
+    status = WeeklyFeedStatus(status_value["feed_count"], status_value["successful_feed_count"], status_value["failed_feed_count"], status_value["stale_feed_count"], status_value["missing_feed_count"], status_value["complete"])
+    return WeeklySourceContract(_date(value["as_of"]), _date(value["period_start"]), _date(value["period_end_exclusive"]), _generated_at(value["generated_at"]), value["run_mode"], value["producer_ref"], value["source_provenance"], artifacts, status)
+
+
+def serialize_weekly_contract(contract: WeeklySourceContract) -> bytes:
+    if not isinstance(contract, WeeklySourceContract):
+        raise _invalid("contract_type_invalid")
+    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": 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}}
+    parse_weekly_contract(payload)
+    return json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")

tests/test_weekly_contract.py

@@ -0,0 +1,86 @@
+from __future__ import annotations
+
+import json
+from datetime import date, datetime, timezone
+
+import pytest
+
+from political_event_tracking_research.weekly_contract import (
+    MAX_SAFE_JSON_INTEGER,
+    WeeklyContractError,
+    WeeklyFeedStatus,
+    WeeklySourceArtifact,
+    WeeklySourceContract,
+    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_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_parser_and_serializer_are_permutation_stable():
+    items = [
+        {"path": "z.csv", "sha256": "c" * 64, "row_count": 2},
+        {"path": "a.csv", "sha256": "d" * 64, "row_count": 1},
+    ]
+    first = parse_weekly_contract(payload(source_artifacts=items))
+    second = parse_weekly_contract(payload(source_artifacts=list(reversed(items))))
+    assert first == second
+    assert serialize_weekly_contract(first) == serialize_weekly_contract(second)
+
+
+def test_direct_value_objects_canonicalize_and_round_trip():
+    z = WeeklySourceArtifact("z.csv", "c" * 64, 2)
+    a = WeeklySourceArtifact("a.csv", "d" * 64, 1)
+    contract = WeeklySourceContract(
+        date(2026, 7, 12), date(2026, 7, 6), date(2026, 7, 13), datetime(2026, 7, 13, tzinfo=timezone.utc),
+        "manual", "a" * 40, "official_political_event_tracking_research_v1", (z, a), WeeklyFeedStatus(12, 12, 0, 0, 0, True),
+    )
+    assert [item.path for item in contract.source_artifacts] == ["a.csv", "z.csv"]
+    assert parse_weekly_contract(json.loads(serialize_weekly_contract(contract))) == contract
+
+
+@pytest.mark.parametrize("path", ["", "/absolute", "./a.csv", "a/../b.csv", "a//b.csv", "a/b.csv/", "a\\b.csv", "a\tb.csv", "a\nb.csv", "é.csv", "a\x00b.csv"])
+def test_artifact_paths_use_ascii_safe_canonical_posix_allowlist(path):
+    with pytest.raises(WeeklyContractError):
+        parse_weekly_contract(payload(source_artifacts=[{"path": path, "sha256": "c" * 64, "row_count": 1}]))
+    with pytest.raises(WeeklyContractError):
+        WeeklySourceArtifact(path, "c" * 64, 1)
+
+
+@pytest.mark.parametrize("provenance", ["", "Official_PERT_v1", "official-pert-v1", "official_pert_é", "official.pert"])
+def test_provenance_is_canonical_ascii_identifier(provenance):
+    with pytest.raises(WeeklyContractError):
+        parse_weekly_contract(payload(source_provenance=provenance))
+
+
+def test_safe_integer_boundaries_and_bool_rejection():
+    assert WeeklySourceArtifact("a.csv", "a" * 64, MAX_SAFE_JSON_INTEGER).row_count == MAX_SAFE_JSON_INTEGER
+    for value in (-1, MAX_SAFE_JSON_INTEGER + 1, True):
+        with pytest.raises(WeeklyContractError):
+            WeeklySourceArtifact("a.csv", "a" * 64, value)
+        with pytest.raises(WeeklyContractError):
+            parse_weekly_contract(payload(feed_status={"feed_count": value, "successful_feed_count": value, "failed_feed_count": 0, "stale_feed_count": 0, "missing_feed_count": 0, "complete": True}))
+
+
+def test_feed_status_is_typed_and_serializer_preserves_counters():
+    status = WeeklyFeedStatus(12, 12, 0, 0, 0, True)
+    assert parse_weekly_contract(json.loads(serialize_weekly_contract(parse_weekly_contract(payload(feed_status={"feed_count": 12, "successful_feed_count": 12, "failed_feed_count": 0, "stale_feed_count": 0, "missing_feed_count": 0, "complete": True}))))) .feed_status == status
+
+
+def test_unknown_fields_and_contract_or_period_mismatch_fail_closed():
+    with pytest.raises(WeeklyContractError):
+        parse_weekly_contract(payload(extra=True))
+    with pytest.raises(WeeklyContractError):
+        parse_weekly_contract(payload(as_of="2026-07-13"))