From 38376f5c694536df9d30e06154479ee1c5368bd9 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Wed, 15 Jul 2026 15:31:30 +0800 Subject: [PATCH 1/2] feat: publish weekly producer contract artifact Co-Authored-By: Codex --- .github/workflows/rss_source_pipeline.yml | 64 +++- docs/weekly_producer_artifact_contract.md | 35 +++ scripts/write_weekly_artifact.py | 57 ++++ .../weekly_producer.py | 294 ++++++++++++++++++ tests/test_weekly_producer.py | 161 ++++++++++ tests/test_weekly_producer_workflow.py | 21 ++ 6 files changed, 628 insertions(+), 4 deletions(-) create mode 100644 docs/weekly_producer_artifact_contract.md create mode 100644 scripts/write_weekly_artifact.py create mode 100644 src/political_event_tracking_research/weekly_producer.py create mode 100644 tests/test_weekly_producer.py create mode 100644 tests/test_weekly_producer_workflow.py diff --git a/.github/workflows/rss_source_pipeline.yml b/.github/workflows/rss_source_pipeline.yml index 7bd252b..7ac3c29 100644 --- a/.github/workflows/rss_source_pipeline.yml +++ b/.github/workflows/rss_source_pipeline.yml @@ -23,6 +23,14 @@ on: required: false default: "50" type: string + period_start: + description: "Manual weekly period Monday UTC (YYYY-MM-DD)." + required: true + type: string + as_of: + description: "Manual weekly as_of Sunday UTC (YYYY-MM-DD)." + required: true + type: string commit_outputs: description: "Commit generated live CSV outputs back to data/live." required: false @@ -32,7 +40,7 @@ on: - "false" - "true" schedule: - - cron: "15 12 * * 6" + - cron: "15 0 * * 1" permissions: contents: write @@ -46,8 +54,8 @@ jobs: runs-on: ubuntu-latest timeout-minutes: 30 steps: - - uses: actions/checkout@v6 - - uses: actions/setup-python@v6 + - uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6 + - uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6 with: python-version: "3.11" - name: Install package @@ -75,6 +83,47 @@ jobs: --watchlist "${WATCHLIST_PATH}" \ --events data/output/rss_source_pipeline/source_events.csv \ --output data/output/rss_source_pipeline/source_tracker.csv + - name: Build weekly contract artifact + env: + INPUT_PERIOD_START: ${{ inputs.period_start || '' }} + INPUT_AS_OF: ${{ inputs.as_of || '' }} + PRODUCER_REF: ${{ github.sha }} + run: | + set -euo pipefail + if [ "${GITHUB_EVENT_NAME}" = "workflow_dispatch" ]; then + test -n "${INPUT_PERIOD_START}" + test -n "${INPUT_AS_OF}" + PERIOD_START="${INPUT_PERIOD_START}" + AS_OF="${INPUT_AS_OF}" + RUN_MODE=manual + else + RUN_DATE="$(date -u +%F)" + read -r PERIOD_START AS_OF < <(RUN_DATE="${RUN_DATE}" python3 - <<'PY' + import os + from datetime import date, timedelta + + today = date.fromisoformat(os.environ["RUN_DATE"]) + start = today - timedelta(days=today.weekday() + 7) + print(start.isoformat(), (start + timedelta(days=6)).isoformat()) + PY + ) + RUN_MODE=scheduled + fi + GENERATED_AT="$(date -u +%Y-%m-%dT%H:%M:%S.%6NZ)" + python scripts/write_weekly_artifact.py \ + --base-dir . \ + --output-dir data/output/rss_source_pipeline/weekly-artifact \ + --feed-status data/output/rss_source_pipeline/source_fetch_status.json \ + --period-start "${PERIOD_START}" \ + --as-of "${AS_OF}" \ + --generated-at "${GENERATED_AT}" \ + --run-mode "${RUN_MODE}" \ + --producer-ref "${PRODUCER_REF}" \ + --source-provenance official_political_event_tracking_research_v1 \ + data/output/rss_source_pipeline/source_items.csv \ + data/output/rss_source_pipeline/source_events.csv \ + data/output/rss_source_pipeline/source_tracker.csv \ + data/output/rss_source_pipeline/source_fetch_status.json - name: Publish live CSV outputs to repository env: COMMIT_OUTPUTS: ${{ github.event_name == 'schedule' && 'true' || github.event.inputs.commit_outputs || 'false' }} @@ -108,8 +157,15 @@ jobs: git push fi - name: Upload RSS source artifact - uses: actions/upload-artifact@v7 + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7 with: name: rss-source-pipeline path: data/output/rss_source_pipeline/ if-no-files-found: error + - name: Upload weekly contract artifact + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7 + with: + name: political-event-weekly-v1 + path: data/output/rss_source_pipeline/weekly-artifact/weekly_manifest.json + if-no-files-found: error + retention-days: 30 diff --git a/docs/weekly_producer_artifact_contract.md b/docs/weekly_producer_artifact_contract.md new file mode 100644 index 0000000..e9b5edd --- /dev/null +++ b/docs/weekly_producer_artifact_contract.md @@ -0,0 +1,35 @@ +# Weekly producer artifact contract + +This is the producer-side boundary for `political_event_weekly.v1`. It is a +pure local adapter: callers provide the completed UTC week, `as_of`, actual +UTC `generated_at`, producer full SHA, provenance identifier, input files, and +feed status. The adapter never infers dates, reads the wall clock, fetches +feeds, or uploads files. + +## Artifact + +- Name: `political-event-weekly-v1` +- File set: exactly `weekly_manifest.json` +- Retention: 30 days, configured by the consuming artifact workflow; it is not + inferred from report content. +- Bytes: canonical `political_event_weekly.v1` manifest bytes from the merged + weekly manifest serializer. + +The input file paths are repository-relative POSIX paths under the trusted +`base_dir`. They must be regular, non-symlink files. Their SHA-256 and CSV row +counts are read from those exact files. The feed status file must be one of +the declared inputs, and its counters must agree with every feed entry. Any +failed, stale, missing, partial, malformed, or mismatched input fails before +the output directory is created. + +The CLI requires all period, timestamp, producer, provenance, and path inputs +explicitly. Upload availability is an external workflow gate: callers must +only upload the returned artifact after successful local readback; an upload +step that cannot accept the exact single-file artifact must fail the workflow, +not publish a success signal. + +The RSS source workflow now supplies this boundary without changing fetching: +scheduled runs execute Monday UTC after the previous ISO week is complete and +derive that week using the documented producer rule; manual runs require both +`period_start` and `as_of`. It uploads the single manifest with a 30-day +retention setting. Other workflows and downstream consumers are unchanged. diff --git a/scripts/write_weekly_artifact.py b/scripts/write_weekly_artifact.py new file mode 100644 index 0000000..b93caae --- /dev/null +++ b/scripts/write_weekly_artifact.py @@ -0,0 +1,57 @@ +#!/usr/bin/env python3 +"""Write a producer-owned ``political_event_weekly.v1`` manifest artifact.""" + +from __future__ import annotations + +import argparse +from datetime import datetime +from pathlib import Path +import sys + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) + +from political_event_tracking_research.weekly_contract import WeeklyContractError +from political_event_tracking_research.weekly_producer import write_weekly_artifact_from_files + + +def _generated_at(value: str) -> datetime: + if not value.endswith("Z"): + raise argparse.ArgumentTypeError("generated_at_invalid") + try: + return datetime.fromisoformat(value[:-1] + "+00:00") + except ValueError as error: + raise argparse.ArgumentTypeError("generated_at_invalid") from error + + +def main() -> None: + parser = argparse.ArgumentParser(description="Write a validated weekly producer artifact.") + parser.add_argument("paths", nargs="+", help="Explicit input files, relative to --base-dir.") + parser.add_argument("--base-dir", required=True) + parser.add_argument("--output-dir", required=True) + parser.add_argument("--feed-status", required=True) + parser.add_argument("--period-start", required=True) + parser.add_argument("--as-of", required=True) + parser.add_argument("--generated-at", required=True, type=_generated_at) + parser.add_argument("--run-mode", required=True, choices=("scheduled", "manual")) + parser.add_argument("--producer-ref", required=True) + parser.add_argument("--source-provenance", required=True) + args = parser.parse_args() + try: + write_weekly_artifact_from_files( + args.paths, + base_dir=args.base_dir, + output_dir=args.output_dir, + feed_status_path=args.feed_status, + period_start=args.period_start, + as_of=args.as_of, + generated_at=args.generated_at, + run_mode=args.run_mode, + producer_ref=args.producer_ref, + source_provenance=args.source_provenance, + ) + except WeeklyContractError as error: + raise SystemExit(error.code) from None + + +if __name__ == "__main__": + main() diff --git a/src/political_event_tracking_research/weekly_producer.py b/src/political_event_tracking_research/weekly_producer.py new file mode 100644 index 0000000..7dcf440 --- /dev/null +++ b/src/political_event_tracking_research/weekly_producer.py @@ -0,0 +1,294 @@ +"""Producer-side integration for a validated ``political_event_weekly.v1`` artifact. + +This module only materializes a manifest from explicitly supplied local inputs. It +does not fetch feeds, infer a period, or publish/upload an artifact. +""" +from __future__ import annotations + +import csv +import hashlib +import json +import shutil +import stat +from collections.abc import Mapping, Sequence +from datetime import date, datetime, timedelta +from pathlib import Path +from typing import Any + +from .weekly_contract import ( + MAX_SAFE_JSON_INTEGER, + WeeklyContractError, + WeeklyFeedStatus, + WeeklySourceArtifact, + WeeklySourceContract, + _date, +) +from .weekly_manifest import ( + parse_weekly_manifest, + serialize_weekly_manifest, +) + +WEEKLY_ARTIFACT_NAME = "political-event-weekly-v1" +WEEKLY_ARTIFACT_FILE = "weekly_manifest.json" +WEEKLY_ARTIFACT_FILES = (WEEKLY_ARTIFACT_FILE,) +WEEKLY_ARTIFACT_RETENTION_DAYS = 30 + + +def _invalid(code: str) -> WeeklyContractError: + return WeeklyContractError(code) + + +def _checked_base(path: str | Path) -> Path: + try: + base = Path(path) + except (TypeError, ValueError): + raise _invalid("producer_base_invalid") from None + try: + base_stat = base.lstat() + except OSError: + raise _invalid("producer_base_invalid") from None + if stat.S_ISLNK(base_stat.st_mode) or not stat.S_ISDIR(base_stat.st_mode): + raise _invalid("producer_base_invalid") + return base + + +def _relative_input(path: str | Path, base: Path) -> tuple[str, Path]: + try: + candidate = Path(path) + except (TypeError, ValueError): + raise _invalid("producer_input_path_invalid") from None + if not candidate.is_absolute(): + candidate = base / candidate + try: + relative = candidate.relative_to(base) + except ValueError: + raise _invalid("producer_input_path_invalid") from None + if not relative.parts or any(part in {"", ".", ".."} for part in relative.parts) or "\\" in relative.as_posix(): + raise _invalid("producer_input_path_invalid") + + current = base + try: + for index, part in enumerate(relative.parts): + current = current / part + item_stat = current.lstat() + is_final = index == len(relative.parts) - 1 + if stat.S_ISLNK(item_stat.st_mode) or (is_final and not stat.S_ISREG(item_stat.st_mode)) or ( + not is_final and not stat.S_ISDIR(item_stat.st_mode) + ): + raise _invalid("producer_input_file_invalid") + except WeeklyContractError: + raise + except OSError: + raise _invalid("producer_input_file_invalid") from None + return relative.as_posix(), current + + +def _sha256(path: Path) -> str: + digest = hashlib.sha256() + try: + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + except OSError: + raise _invalid("producer_input_file_invalid") from None + return digest.hexdigest() + + +def _row_count(path: Path) -> int: + if path.suffix.lower() != ".csv": + return 0 + try: + with path.open(newline="", encoding="utf-8") as handle: + return sum(1 for _ in csv.DictReader(handle)) + except (OSError, UnicodeError, csv.Error): + raise _invalid("producer_input_file_invalid") from None + + +def _safe_counter(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 _feed_status(path: Path) -> WeeklyFeedStatus: + try: + value = json.loads(path.read_text(encoding="utf-8")) + except (OSError, UnicodeError, json.JSONDecodeError, TypeError, ValueError, RecursionError): + raise _invalid("feed_status_invalid") from None + if not isinstance(value, Mapping): + raise _invalid("feed_status_invalid") + required = {"feed_count", "successful_feed_count", "failed_feed_count", "feeds"} + if not required.issubset(value) or not isinstance(value["feeds"], list) or not value["feeds"]: + raise _invalid("feed_status_invalid") + + feed_count = _safe_counter(value["feed_count"], "feed_status_invalid") + successful = _safe_counter(value["successful_feed_count"], "feed_status_invalid") + failed = _safe_counter(value["failed_feed_count"], "feed_status_invalid") + stale = _safe_counter(value.get("stale_feed_count", 0), "feed_status_invalid") + missing = _safe_counter(value.get("missing_feed_count", 0), "feed_status_invalid") + calculated_successful = 0 + calculated_failed = 0 + calculated_stale = 0 + calculated_missing = 0 + for item in value["feeds"]: + if not isinstance(item, Mapping) or type(item.get("ok")) is not bool: + raise _invalid("feed_status_invalid") + if item["ok"]: + calculated_successful += 1 + else: + calculated_failed += 1 + for key, counter in (("stale", "stale_feed_count"), ("missing", "missing_feed_count")): + if key in item and type(item[key]) is not bool: + raise _invalid("feed_status_invalid") + if item.get(key) is True: + if counter == "stale_feed_count": + calculated_stale += 1 + else: + calculated_missing += 1 + if (feed_count, successful, failed, stale, missing) != ( + len(value["feeds"]), + calculated_successful, + calculated_failed, + calculated_stale, + calculated_missing, + ): + raise _invalid("feed_status_mismatch") + try: + return WeeklyFeedStatus(feed_count, successful, failed, stale, missing, True) + except WeeklyContractError: + raise + + +def _contract_from_files( + artifact_paths: Sequence[str | Path], + *, + feed_status_path: str | Path, + base_dir: str | Path, + period_start: date, + as_of: date, + generated_at: datetime, + run_mode: str, + producer_ref: str, + source_provenance: str, +) -> WeeklySourceContract: + if isinstance(artifact_paths, (str, bytes)) or not isinstance(artifact_paths, Sequence) or not artifact_paths: + raise _invalid("producer_inputs_invalid") + if type(period_start) is not date or type(as_of) is not date or type(generated_at) is not datetime: + raise _invalid("producer_inputs_invalid") + base = _checked_base(base_dir) + seen: set[str] = set() + artifacts: list[WeeklySourceArtifact] = [] + status_key: str | None = None + status_file: Path | None = None + for raw_path in artifact_paths: + key, path = _relative_input(raw_path, base) + if key in seen: + raise _invalid("producer_input_duplicate") + seen.add(key) + artifacts.append(WeeklySourceArtifact(key, _sha256(path), _row_count(path))) + status_key, status_file = _relative_input(feed_status_path, base) + if status_key not in seen or status_file is None: + raise _invalid("feed_status_missing") + status = _feed_status(status_file) + try: + period_end_exclusive = period_start + timedelta(days=7) + except OverflowError: + raise _invalid("period_invalid") from None + try: + return WeeklySourceContract( + as_of, + period_start, + period_end_exclusive, + generated_at, + run_mode, + producer_ref, + source_provenance, + tuple(artifacts), + status, + ) + except WeeklyContractError: + raise + + +def build_weekly_manifest_from_files( + artifact_paths: Sequence[str | Path], + *, + feed_status_path: str | Path, + base_dir: str | Path, + period_start: date | str, + as_of: date | str, + generated_at: datetime, + run_mode: str, + producer_ref: str, + source_provenance: str, + expected_manifest: Mapping[str, object] | None = None, +) -> bytes: + """Build canonical manifest bytes from explicitly supplied, local inputs. + + Date strings are accepted only for CLI-friendly construction and are parsed + strictly by the merged weekly contract. No date or timestamp is inferred. + """ + parsed_period_start = _date(period_start) if isinstance(period_start, str) else period_start + parsed_as_of = _date(as_of) if isinstance(as_of, str) else as_of + contract = _contract_from_files( + artifact_paths, + feed_status_path=feed_status_path, + base_dir=base_dir, + period_start=parsed_period_start, + as_of=parsed_as_of, + generated_at=generated_at, + run_mode=run_mode, + producer_ref=producer_ref, + source_provenance=source_provenance, + ) + content = serialize_weekly_manifest(contract) + if expected_manifest is not None: + try: + expected = parse_weekly_manifest(expected_manifest) + if serialize_weekly_manifest(expected) != content: + raise _invalid("manifest_input_mismatch") + except WeeklyContractError: + raise + except (TypeError, ValueError, UnicodeError, RecursionError): + raise _invalid("manifest_input_mismatch") from None + return content + + +def write_weekly_artifact_from_files( + artifact_paths: Sequence[str | Path], + *, + output_dir: str | Path, + expected_manifest: Mapping[str, object] | None = None, + **kwargs: Any, +) -> Path: + """Write one validated manifest file without overwriting an existing artifact.""" + content = build_weekly_manifest_from_files(artifact_paths, expected_manifest=expected_manifest, **kwargs) + output = Path(output_dir) + created_output = False + try: + if output.is_symlink(): + raise _invalid("artifact_output_invalid") + if output.exists(): + if not output.is_dir() or any(output.iterdir()): + raise _invalid("artifact_output_exists") + else: + output.mkdir(parents=True) + created_output = True + target = output / WEEKLY_ARTIFACT_FILE + with target.open("xb") as handle: + handle.write(content) + members = list(output.iterdir()) + if [member.name for member in members] != [WEEKLY_ARTIFACT_FILE] or not stat.S_ISREG(target.lstat().st_mode): + raise _invalid("artifact_output_invalid") + from .weekly_manifest import parse_weekly_manifest_bytes # noqa: PLC0415 - avoid import cycle at module load + + parse_weekly_manifest_bytes(target.read_bytes()) + except WeeklyContractError: + if created_output: + shutil.rmtree(output, ignore_errors=True) + raise + except (OSError, TypeError, ValueError): + if created_output: + shutil.rmtree(output, ignore_errors=True) + raise _invalid("artifact_write_invalid") from None + return target diff --git a/tests/test_weekly_producer.py b/tests/test_weekly_producer.py new file mode 100644 index 0000000..d68f56c --- /dev/null +++ b/tests/test_weekly_producer.py @@ -0,0 +1,161 @@ +from __future__ import annotations + +import json +from datetime import date, datetime, timezone +from pathlib import Path + +import pytest + +from political_event_tracking_research.weekly_contract import WeeklyContractError +from political_event_tracking_research.weekly_manifest import parse_weekly_manifest_bytes +from political_event_tracking_research.weekly_producer import ( + WEEKLY_ARTIFACT_FILE, + WEEKLY_ARTIFACT_NAME, + WEEKLY_ARTIFACT_RETENTION_DAYS, + build_weekly_manifest_from_files, + write_weekly_artifact_from_files, +) + + +def _write_inputs(root: Path, *, failed: bool = False) -> tuple[Path, Path]: + events = root / "data" / "live" / "source_events.csv" + status = root / "data" / "live" / "source_fetch_status.json" + events.parent.mkdir(parents=True) + events.write_text("event_id,symbol\ne1,MU\ne2,INTC\n", encoding="utf-8") + status.write_text( + json.dumps( + { + "feed_count": 2, + "successful_feed_count": 1 if failed else 2, + "failed_feed_count": 1 if failed else 0, + "feeds": [ + {"feed_id": "a", "ok": not failed, "item_count": 1}, + {"feed_id": "b", "ok": True, "item_count": 1}, + ], + } + ), + encoding="utf-8", + ) + return events, status + + +def _kwargs(root: Path, events: Path, status: Path) -> dict[str, object]: + return { + "artifact_paths": [events, status], + "feed_status_path": status, + "base_dir": root, + "period_start": "2026-07-06", + "as_of": "2026-07-12", + "generated_at": datetime(2026, 7, 13, tzinfo=timezone.utc), + "run_mode": "manual", + "producer_ref": "a" * 40, + "source_provenance": "official_political_event_tracking_research_v1", + } + + +def test_builds_manifest_from_real_input_files_and_is_deterministic(tmp_path: Path) -> None: + events, status = _write_inputs(tmp_path) + first = build_weekly_manifest_from_files(**_kwargs(tmp_path, events, status)) + second_kwargs = _kwargs(tmp_path, events, status) + second_kwargs["artifact_paths"] = [status, events] + second = build_weekly_manifest_from_files(**second_kwargs) + + assert first == second + contract = parse_weekly_manifest_bytes(first) + assert contract.feed_status.feed_count == 2 + assert {item.path for item in contract.source_artifacts} == { + "data/live/source_events.csv", + "data/live/source_fetch_status.json", + } + + +def test_writes_the_declared_single_file_artifact(tmp_path: Path) -> None: + events, status = _write_inputs(tmp_path) + output = tmp_path / "artifact" + target = write_weekly_artifact_from_files(**_kwargs(tmp_path, events, status), output_dir=output) + + assert WEEKLY_ARTIFACT_NAME == "political-event-weekly-v1" + assert WEEKLY_ARTIFACT_RETENTION_DAYS > 0 + assert target == output / WEEKLY_ARTIFACT_FILE + assert [path.name for path in output.iterdir()] == [WEEKLY_ARTIFACT_FILE] + parse_weekly_manifest_bytes(target.read_bytes()) + + +@pytest.mark.parametrize("field", ["period_start", "as_of", "generated_at", "producer_ref"]) +def test_requires_explicit_valid_producer_inputs(tmp_path: Path, field: str) -> None: + events, status = _write_inputs(tmp_path) + kwargs = _kwargs(tmp_path, events, status) + kwargs[field] = "" if field != "generated_at" else datetime(2026, 7, 12, tzinfo=timezone.utc) + with pytest.raises(WeeklyContractError): + build_weekly_manifest_from_files(**kwargs) + + +def test_partial_feed_fails_before_artifact_creation(tmp_path: Path) -> None: + events, status = _write_inputs(tmp_path, failed=True) + output = tmp_path / "artifact" + with pytest.raises(WeeklyContractError): + write_weekly_artifact_from_files(**_kwargs(tmp_path, events, status), output_dir=output) + assert not output.exists() + + +def test_mismatched_expected_manifest_fails_before_write(tmp_path: Path) -> None: + events, status = _write_inputs(tmp_path) + output = tmp_path / "artifact" + expected = {"manifest_type": "political_event_weekly_manifest", "contract": {}} + with pytest.raises(WeeklyContractError): + write_weekly_artifact_from_files(**_kwargs(tmp_path, events, status), output_dir=output, expected_manifest=expected) + assert not output.exists() + + +def test_existing_nonempty_artifact_directory_is_not_overwritten(tmp_path: Path) -> None: + events, status = _write_inputs(tmp_path) + output = tmp_path / "artifact" + output.mkdir() + sentinel = output / "sentinel" + sentinel.write_text("keep", encoding="utf-8") + with pytest.raises(WeeklyContractError): + write_weekly_artifact_from_files(**_kwargs(tmp_path, events, status), output_dir=output) + assert sentinel.read_text(encoding="utf-8") == "keep" + + +def test_relative_inputs_are_resolved_under_base_dir(tmp_path: Path) -> None: + events, status = _write_inputs(tmp_path) + kwargs = _kwargs(tmp_path, events, status) + kwargs["artifact_paths"] = ["data/live/source_events.csv", "data/live/source_fetch_status.json"] + kwargs["feed_status_path"] = "data/live/source_fetch_status.json" + assert parse_weekly_manifest_bytes(build_weekly_manifest_from_files(**kwargs)).feed_status.complete + + +def test_symlink_input_fails_closed(tmp_path: Path) -> None: + events, status = _write_inputs(tmp_path) + link = tmp_path / "data" / "live" / "link.csv" + link.symlink_to(events) + kwargs = _kwargs(tmp_path, events, status) + kwargs["artifact_paths"] = [link, status] + with pytest.raises(WeeklyContractError): + build_weekly_manifest_from_files(**kwargs) + + +def test_stale_or_missing_feed_status_fails_closed(tmp_path: Path) -> None: + events, status = _write_inputs(tmp_path) + payload = json.loads(status.read_text(encoding="utf-8")) + payload["feeds"][0]["stale"] = True + status.write_text(json.dumps(payload), encoding="utf-8") + with pytest.raises(WeeklyContractError): + build_weekly_manifest_from_files(**_kwargs(tmp_path, events, status)) + + +def test_generated_at_before_completed_week_fails_closed(tmp_path: Path) -> None: + events, status = _write_inputs(tmp_path) + kwargs = _kwargs(tmp_path, events, status) + kwargs["generated_at"] = datetime(2026, 7, 12, 23, 59, 59, tzinfo=timezone.utc) + with pytest.raises(WeeklyContractError): + build_weekly_manifest_from_files(**kwargs) + + +def test_unrepresentable_period_boundary_fails_closed(tmp_path: Path) -> None: + events, status = _write_inputs(tmp_path) + kwargs = _kwargs(tmp_path, events, status) + kwargs["period_start"] = date.max + with pytest.raises(WeeklyContractError): + build_weekly_manifest_from_files(**kwargs) diff --git a/tests/test_weekly_producer_workflow.py b/tests/test_weekly_producer_workflow.py new file mode 100644 index 0000000..bdfba77 --- /dev/null +++ b/tests/test_weekly_producer_workflow.py @@ -0,0 +1,21 @@ +from pathlib import Path + + +WORKFLOW = Path(__file__).parents[1] / ".github" / "workflows" / "rss_source_pipeline.yml" + + +def test_weekly_workflow_uses_pinned_actions_and_explicit_contract_inputs() -> None: + text = WORKFLOW.read_text(encoding="utf-8") + assert "actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10" in text + assert "actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1" in text + assert text.count("actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a") == 2 + assert "period_start:" in text and "as_of:" in text + assert "--period-start \"${PERIOD_START}\"" in text + assert "--as-of \"${AS_OF}\"" in text + assert "--generated-at \"${GENERATED_AT}\"" in text + assert "retention-days: 30" in text + assert "if-no-files-found: error" in text + assert 'cron: "15 0 * * 1"' in text + for line in text.splitlines(): + if "uses: actions/" in line: + assert len(line.split("@", 1)[1].split()[0]) == 40 From 1b8e1503f3bc1f7446dd19aa3231614f8ea46ac6 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Wed, 15 Jul 2026 15:46:30 +0800 Subject: [PATCH 2/2] fix: isolate weekly manifest artifact upload Co-Authored-By: Codex --- .github/workflows/rss_source_pipeline.yml | 6 +++++- tests/test_weekly_producer_workflow.py | 12 ++++++++++++ 2 files changed, 17 insertions(+), 1 deletion(-) diff --git a/.github/workflows/rss_source_pipeline.yml b/.github/workflows/rss_source_pipeline.yml index 7ac3c29..e088a1b 100644 --- a/.github/workflows/rss_source_pipeline.yml +++ b/.github/workflows/rss_source_pipeline.yml @@ -160,7 +160,11 @@ jobs: uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7 with: name: rss-source-pipeline - path: data/output/rss_source_pipeline/ + path: | + data/output/rss_source_pipeline/source_items.csv + data/output/rss_source_pipeline/source_events.csv + data/output/rss_source_pipeline/source_tracker.csv + data/output/rss_source_pipeline/source_fetch_status.json if-no-files-found: error - name: Upload weekly contract artifact uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7 diff --git a/tests/test_weekly_producer_workflow.py b/tests/test_weekly_producer_workflow.py index bdfba77..f1e74c6 100644 --- a/tests/test_weekly_producer_workflow.py +++ b/tests/test_weekly_producer_workflow.py @@ -1,3 +1,4 @@ +import re from pathlib import Path @@ -16,6 +17,17 @@ def test_weekly_workflow_uses_pinned_actions_and_explicit_contract_inputs() -> N assert "retention-days: 30" in text assert "if-no-files-found: error" in text assert 'cron: "15 0 * * 1"' in text + upload_steps = re.findall(r"^ - name: Upload.*?(?=^ - name:|\Z)", text, re.MULTILINE | re.DOTALL) + assert len(upload_steps) == 2 + broad, dedicated = upload_steps + assert "weekly_manifest.json" not in broad + assert "source_items.csv" in broad + assert "source_events.csv" in broad + assert "source_tracker.csv" in broad + assert "source_fetch_status.json" in broad + assert dedicated.count("weekly_manifest.json") == 1 + assert "name: political-event-weekly-v1" in dedicated + assert "path: data/output/rss_source_pipeline/weekly-artifact/weekly_manifest.json" in dedicated for line in text.splitlines(): if "uses: actions/" in line: assert len(line.split("@", 1)[1].split()[0]) == 40