diff --git a/.github/workflows/rss_source_pipeline.yml b/.github/workflows/rss_source_pipeline.yml index 7bd252b..b72763b 100644 --- a/.github/workflows/rss_source_pipeline.yml +++ b/.github/workflows/rss_source_pipeline.yml @@ -23,6 +23,16 @@ on: required: false default: "50" type: string + period_start: + description: "Manual completed ISO-week Monday date; required for manual runs." + required: false + default: "" + type: string + as_of: + description: "Manual completed ISO-week Sunday date; required for manual runs." + required: false + default: "" + type: string commit_outputs: description: "Commit generated live CSV outputs back to data/live." required: false @@ -32,7 +42,7 @@ on: - "false" - "true" schedule: - - cron: "15 12 * * 6" + - cron: "15 12 * * 1" permissions: contents: write @@ -107,9 +117,42 @@ jobs: git commit -m "Update live RSS source events [skip ci]" git push fi + - name: Build completed weekly producer artifact + env: + PERIOD_START: ${{ github.event.inputs.period_start || '' }} + AS_OF: ${{ github.event.inputs.as_of || '' }} + WATCHLIST_PATH: ${{ github.event.inputs.watchlist_path || 'data/live/political_watchlist.csv' }} + run: | + set -euo pipefail + test "${GITHUB_RUN_ATTEMPT}" = "1" + mkdir -p data/output/political-event-weekly-v1 + ARGS=( + --generated-at "$(date -u +%Y-%m-%dT%H:%M:%S.000000Z)" + --workflow-ref "${GITHUB_WORKFLOW_REF}" + --source-run-id "${GITHUB_RUN_ID}" + --producer-ref "${GITHUB_SHA}" + --source-events data/output/rss_source_pipeline/source_events.csv + --watchlist "${WATCHLIST_PATH}" + --feed-status data/output/rss_source_pipeline/source_fetch_status.json + --output-dir data/output/political-event-weekly-v1 + ) + if [ "${GITHUB_EVENT_NAME}" = "schedule" ]; then + ARGS+=(--scheduled-today "$(date -u +%F)" --run-mode scheduled) + else + test -n "${PERIOD_START}" && test -n "${AS_OF}" + ARGS+=(--period-start "${PERIOD_START}" --as-of "${AS_OF}" --run-mode manual) + fi + python scripts/build_weekly_artifact.py "${ARGS[@]}" - name: Upload RSS source artifact uses: actions/upload-artifact@v7 with: name: rss-source-pipeline path: data/output/rss_source_pipeline/ if-no-files-found: error + - name: Upload political weekly artifact + uses: actions/upload-artifact@v7 + with: + name: political-event-weekly-v1 + path: data/output/political-event-weekly-v1/ + 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..982314b --- /dev/null +++ b/docs/weekly_producer_artifact_contract.md @@ -0,0 +1,40 @@ +# Political Event Weekly Producer Artifact + +`political-event-weekly-v1` is the producer-owned, complete-UTC-ISO-week artifact +emitted by the RSS source pipeline. + +## Fixed contract + +- Scheduled execution is Monday `12:15 UTC` and covers the immediately preceding + complete Monday–Sunday UTC ISO week. +- Manual execution must provide matching `period_start` (Monday) and `as_of` + (Sunday). No latest-source or wall-clock fallback is allowed for period identity. +- `generated_at` is the actual UTC build time and is separate from period identity. +- `GITHUB_RUN_ATTEMPT` must be exactly `1`; a rerun is fail-closed rather than a + new version of the same snapshot. +- Artifact retention is exactly 30 days. + +The dedicated artifact contains exactly: + +```text +period_lock.json +political_events.csv +political_watchlist.csv +political_event_weekly.json +weekly_manifest.json +``` + +The producer filters `political_events.csv` to the locked inclusive date range +before writing it; malformed `event_date` fails closed. The original source +events/watchlist input snapshot digest remains in `period_lock.json`, while the +filtered event bytes and row count are bound by the artifact manifest. + +The manifest binds the first four files by name, byte length and SHA-256. It also +records CSV headers/row counts, period/as-of/generated-at, producer SHA, +workflow ref, run id/attempt, source snapshot digest/provenance, and complete +feed counters. Any partial/failed/stale/missing feed, source mismatch, unsafe +wire, or readback mismatch prevents dedicated artifact upload. + +This contract is producer-side only. QAR consumer acquisition/readback is a +separate later slice; no legacy compatibility, identity store, Pages, publisher, +or workflow permission expansion is implied. diff --git a/scripts/build_weekly_artifact.py b/scripts/build_weekly_artifact.py new file mode 100644 index 0000000..efc62fa --- /dev/null +++ b/scripts/build_weekly_artifact.py @@ -0,0 +1,91 @@ +#!/usr/bin/env python3 +"""Build the dedicated producer-owned political-event weekly artifact.""" + +from __future__ import annotations + +import argparse +import json +import sys +from datetime import date, datetime, timezone +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) + +from political_event_tracking_research.weekly_artifact import ( # noqa: E402 + WeeklyArtifactError, + build_weekly_artifact, + completed_week_period, +) + + +def _date(value: str) -> date: + try: + return date.fromisoformat(value) + except ValueError: + raise argparse.ArgumentTypeError("invalid ISO date") from None + + +def _generated_at(value: str) -> datetime: + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + raise argparse.ArgumentTypeError("invalid UTC timestamp") from None + if parsed.tzinfo != timezone.utc: + raise argparse.ArgumentTypeError("generated_at must be UTC") + return parsed + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--period-start", type=_date) + parser.add_argument("--as-of", type=_date) + parser.add_argument("--scheduled-today", type=_date) + parser.add_argument("--generated-at", required=True, type=_generated_at) + parser.add_argument("--workflow-ref", required=True) + parser.add_argument("--source-run-id", required=True) + parser.add_argument("--producer-ref", required=True) + parser.add_argument("--source-events", required=True, type=Path) + parser.add_argument("--watchlist", required=True, type=Path) + parser.add_argument("--feed-status", required=True, type=Path) + parser.add_argument("--output-dir", required=True, type=Path) + parser.add_argument("--run-mode", choices=("scheduled", "manual"), required=True) + args = parser.parse_args() + + if args.scheduled_today is not None: + if args.run_mode != "scheduled" or args.period_start is not None or args.as_of is not None: + parser.error("scheduled mode requires only --scheduled-today") + period_start, period_end = completed_week_period(args.scheduled_today) + as_of = period_end - date.resolution + elif args.run_mode == "manual" and args.period_start is not None and args.as_of is not None: + period_start, as_of = args.period_start, args.as_of + else: + parser.error("manual mode requires --period-start and --as-of") + + try: + feed_status = json.loads(args.feed_status.read_text(encoding="utf-8")) + files = build_weekly_artifact( + period_start=period_start, + as_of=as_of, + generated_at=args.generated_at, + workflow_ref=args.workflow_ref, + source_run_id=args.source_run_id, + producer_ref=args.producer_ref, + source_events=args.source_events.read_bytes(), + watchlist=args.watchlist.read_bytes(), + feed_status=feed_status, + source_provenance="official_rss_source_pipeline_v1", + run_mode=args.run_mode, + ) + args.output_dir.mkdir(parents=True, exist_ok=True) + if any(args.output_dir.iterdir()): + raise WeeklyArtifactError("artifact_output_not_empty") + for name, content in files.items(): + (args.output_dir / name).write_bytes(content) + except WeeklyArtifactError as exc: + raise SystemExit(exc.code) from None + except (OSError, UnicodeError, json.JSONDecodeError, TypeError, ValueError): + raise SystemExit("weekly_artifact_input_invalid") from None + + +if __name__ == "__main__": + main() diff --git a/src/political_event_tracking_research/rss_source_fetch.py b/src/political_event_tracking_research/rss_source_fetch.py index 98e6b13..ca0504a 100644 --- a/src/political_event_tracking_research/rss_source_fetch.py +++ b/src/political_event_tracking_research/rss_source_fetch.py @@ -178,6 +178,7 @@ def write_fetch_status(path: str | Path, statuses: list[FeedFetchStatus], *, ite "feed_count": len(statuses), "successful_feed_count": sum(1 for item in statuses if item.ok), "failed_feed_count": sum(1 for item in statuses if not item.ok), + "complete": bool(statuses) and all(item.ok for item in statuses), "item_count": item_count, "feeds": [item.to_json() for item in statuses], } diff --git a/src/political_event_tracking_research/weekly_artifact.py b/src/political_event_tracking_research/weekly_artifact.py new file mode 100644 index 0000000..b3cae51 --- /dev/null +++ b/src/political_event_tracking_research/weekly_artifact.py @@ -0,0 +1,362 @@ +"""Concrete producer-owned weekly artifact contract for the PERT RSS pipeline.""" + +from __future__ import annotations + +import csv +import hashlib +import io +import json +from collections.abc import Mapping +from datetime import date, datetime, timedelta, timezone +from typing import Final + +from .weekly_contract import ( + WeeklyFeedStatus, + WeeklySourceArtifact, + WeeklySourceContract, + WeeklyContractError, + parse_weekly_contract, + serialize_weekly_contract, +) +from .weekly_period_lock import ( + PoliticalEventWeeklyPeriodLockV1, + PeriodLockError, + SourceSnapshotArtifact, + parse_period_lock_bytes, + serialize_period_lock, +) + +ARTIFACT_NAME: Final = "political-event-weekly-v1" +RETENTION_DAYS: Final = 30 +MANIFEST_VERSION: Final = "political_event_weekly_artifact_manifest.v1" +EVENTS_NAME: Final = "political_events.csv" +WATCHLIST_NAME: Final = "political_watchlist.csv" +PERIOD_LOCK_NAME: Final = "period_lock.json" +WEEKLY_NAME: Final = "political_event_weekly.json" +MANIFEST_NAME: Final = "weekly_manifest.json" +ARTIFACT_FILES: Final = (PERIOD_LOCK_NAME, EVENTS_NAME, WATCHLIST_NAME, WEEKLY_NAME, MANIFEST_NAME) +EVENT_HEADER: Final = ("event_id", "event_date", "symbol", "event_type", "direction", "confidence", "source_url", "notes") +WATCHLIST_HEADER: Final = ("symbol", "name", "bucket", "research_status", "thesis", "source_url") + + +class WeeklyArtifactError(ValueError): + """Stable, sanitized producer artifact contract error.""" + + def __init__(self, code: str) -> None: + self.code = code + super().__init__(code) + + +def _invalid(code: str) -> WeeklyArtifactError: + return WeeklyArtifactError(code) + + +def _sha256(value: bytes) -> str: + return hashlib.sha256(value).hexdigest() + + +def _snapshot_digest(events: bytes, watchlist: bytes) -> str: + digest = hashlib.sha256() + for value in (events, watchlist): + digest.update(len(value).to_bytes(8, "big")) + digest.update(value) + return digest.hexdigest() + + +def _canonical_json(value: object, code: str) -> bytes: + try: + return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False).encode("utf-8") + except (TypeError, ValueError, UnicodeError, OverflowError, RecursionError): + raise _invalid(code) from None + + +def _parse_canonical_json(wire: bytes, code: str) -> object: + if type(wire) is not bytes: + raise _invalid(code) + + def pairs(items: list[tuple[str, object]]) -> dict[str, object]: + result: dict[str, object] = {} + for key, item in items: + if key in result: + raise _invalid(f"{code}_duplicate_key") + result[key] = item + return result + + try: + value = json.loads(wire.decode("utf-8"), object_pairs_hook=pairs) + except WeeklyArtifactError: + raise + except (UnicodeError, json.JSONDecodeError, TypeError, ValueError, RecursionError): + raise _invalid(code) from None + if _canonical_json(value, code) != wire: + raise _invalid(f"{code}_noncanonical") + return value + + +def _csv_snapshot(value: object, expected_header: tuple[str, ...], code: str, *, allow_empty: bool = False) -> tuple[int, tuple[str, ...]]: + if type(value) is not bytes: + raise _invalid(code) + try: + text = value.decode("utf-8") + rows = list(csv.reader(io.StringIO(text, newline=""))) + except (UnicodeError, csv.Error, ValueError, RecursionError): + raise _invalid(code) from None + if not rows or tuple(rows[0]) != expected_header: + raise _invalid(f"{code}_header") + if (not allow_empty and len(rows) < 2) or any(len(row) != len(expected_header) for row in rows[1:]): + raise _invalid(f"{code}_rows") + return len(rows) - 1, expected_header + + +def _filter_events(value: object, period_start: date, as_of: date) -> bytes: + if type(value) is not bytes: + raise _invalid("events_csv_invalid") + try: + rows = list(csv.reader(io.StringIO(value.decode("utf-8"), newline=""))) + except (UnicodeError, csv.Error, ValueError, RecursionError): + raise _invalid("events_csv_invalid") from None + if not rows or tuple(rows[0]) != EVENT_HEADER: + raise _invalid("events_csv_header") + selected: list[list[str]] = [] + for row in rows[1:]: + if len(row) != len(EVENT_HEADER): + raise _invalid("events_csv_rows") + event_date = row[1] + if type(event_date) is not str or len(event_date) != 10: + raise _invalid("events_date_invalid") + try: + parsed_date = date.fromisoformat(event_date) + except ValueError: + raise _invalid("events_date_invalid") from None + if period_start <= parsed_date <= as_of: + selected.append(row) + output = io.StringIO(newline="") + writer = csv.writer(output, lineterminator="\n") + writer.writerow(EVENT_HEADER) + writer.writerows(selected) + return output.getvalue().encode("utf-8") + + +def _feed_status(value: object) -> WeeklyFeedStatus: + if not isinstance(value, Mapping): + raise _invalid("feed_status_invalid") + required = ("feed_count", "successful_feed_count", "failed_feed_count", "complete") + if any(key not in value for key in required): + raise _invalid("feed_status_invalid") + try: + status = WeeklyFeedStatus( + value["feed_count"], + value["successful_feed_count"], + value["failed_feed_count"], + value.get("stale_feed_count", 0), + value.get("missing_feed_count", 0), + value["complete"], + ) + except (WeeklyContractError, TypeError, ValueError): + raise _invalid("feed_status_incomplete") from None + return status + + +def _period_dates(period_start: object, as_of: object) -> tuple[date, date]: + if type(period_start) is not date or type(as_of) is not date: + raise _invalid("period_invalid") + try: + period_end = period_start + timedelta(days=7) + except OverflowError: + raise _invalid("period_invalid") from None + if period_start.weekday() != 0 or as_of != period_end - timedelta(days=1): + raise _invalid("period_mismatch") + return period_start, period_end + + +def completed_week_period(today: date) -> tuple[date, date]: + """Return the completed ISO week immediately before a scheduled Monday run.""" + if type(today) is not date or today.weekday() != 0: + raise _invalid("scheduled_period_invalid") + try: + period_start = today - timedelta(days=7) + except OverflowError: + raise _invalid("scheduled_period_invalid") from None + return period_start, period_start + timedelta(days=7) + + +def _timestamp(value: object) -> datetime: + if type(value) is not datetime or value.tzinfo != timezone.utc: + raise _invalid("generated_at_invalid") + return value + + +def _contract_payload(contract: WeeklySourceContract) -> dict[str, object]: + return json.loads(serialize_weekly_contract(contract)) + + +def _file_metadata(name: str, value: bytes, *, role: str, row_count: int | None = None, header: tuple[str, ...] | None = None) -> dict[str, object]: + return { + "name": name, + "role": role, + "length": len(value), + "sha256": _sha256(value), + "row_count": row_count, + "header": list(header) if header is not None else None, + } + + +def _manifest_payload(lock: PoliticalEventWeeklyPeriodLockV1, contract: WeeklySourceContract, files: Mapping[str, bytes]) -> dict[str, object]: + event_rows, event_header = _csv_snapshot(files[EVENTS_NAME], EVENT_HEADER, "events_csv_invalid", allow_empty=True) + watch_rows, watch_header = _csv_snapshot(files[WATCHLIST_NAME], WATCHLIST_HEADER, "watchlist_csv_invalid") + return { + "manifest_version": MANIFEST_VERSION, + "artifact_name": ARTIFACT_NAME, + "retention_days": RETENTION_DAYS, + "schema_version": "1", + "contract_version": "political_event_weekly.v1", + "cadence": "weekly", + "period_start": contract.period_start.isoformat(), + "period_end_exclusive": contract.period_end_exclusive.isoformat(), + "as_of": contract.as_of.isoformat(), + "generated_at": contract.generated_at.isoformat(timespec="microseconds").replace("+00:00", "Z"), + "workflow_ref": lock.workflow_ref, + "producer_ref": lock.producer_ref, + "source_run_id": lock.source_run_id, + "source_attempt": lock.source_attempt, + "source_snapshot_id": lock.source_snapshot_id, + "source_snapshot_digest": lock.source_snapshot_digest, + "source_provenance": lock.source_provenance, + "feed_status": _contract_payload(contract)["feed_status"], + "files": [ + _file_metadata(PERIOD_LOCK_NAME, files[PERIOD_LOCK_NAME], role="period_lock"), + _file_metadata(EVENTS_NAME, files[EVENTS_NAME], role="source_events", row_count=event_rows, header=event_header), + _file_metadata(WATCHLIST_NAME, files[WATCHLIST_NAME], role="source_watchlist", row_count=watch_rows, header=watch_header), + _file_metadata(WEEKLY_NAME, files[WEEKLY_NAME], role="weekly_contract"), + ], + } + + +def build_weekly_artifact( + *, + period_start: date, + as_of: date, + generated_at: datetime, + workflow_ref: str, + source_run_id: str, + producer_ref: str, + source_events: bytes, + watchlist: bytes, + feed_status: Mapping[str, object], + source_provenance: str, + run_mode: str, +) -> dict[str, bytes]: + period_start, period_end = _period_dates(period_start, as_of) + generated_at = _timestamp(generated_at) + if generated_at < datetime.combine(period_end, datetime.min.time(), timezone.utc): + raise _invalid("generated_at_invalid") + raw_snapshot_digest = _snapshot_digest(source_events, watchlist) + period_events = _filter_events(source_events, period_start, as_of) + event_rows, _ = _csv_snapshot(period_events, EVENT_HEADER, "events_csv_invalid", allow_empty=True) + watch_rows, _ = _csv_snapshot(watchlist, WATCHLIST_HEADER, "watchlist_csv_invalid") + status = _feed_status(feed_status) + if type(source_provenance) is not str or source_provenance != "official_rss_source_pipeline_v1": + raise _invalid("source_provenance_invalid") + if type(run_mode) is not str or run_mode not in {"scheduled", "manual"}: + raise _invalid("run_mode_invalid") + source_snapshot_digest = raw_snapshot_digest + source_snapshot_id = f"rss_source_snapshot_{as_of:%Y%m%d}_{source_run_id}" + try: + lock = PoliticalEventWeeklyPeriodLockV1( + period_start, + period_end, + as_of, + workflow_ref, + source_run_id, + 1, + producer_ref, + source_snapshot_id, + source_snapshot_digest, + source_provenance, + ( + SourceSnapshotArtifact(EVENTS_NAME, _sha256(period_events), event_rows), + SourceSnapshotArtifact(WATCHLIST_NAME, _sha256(watchlist), watch_rows), + ), + ) + contract = WeeklySourceContract( + as_of, + period_start, + period_end, + generated_at, + run_mode, + producer_ref, + source_provenance, + ( + WeeklySourceArtifact(EVENTS_NAME, _sha256(period_events), event_rows), + WeeklySourceArtifact(WATCHLIST_NAME, _sha256(watchlist), watch_rows), + ), + status, + ) + period_lock = serialize_period_lock(lock) + weekly = serialize_weekly_contract(contract) + except (PeriodLockError, WeeklyContractError, TypeError, ValueError, OverflowError): + raise _invalid("weekly_artifact_invalid") from None + files: dict[str, bytes] = { + PERIOD_LOCK_NAME: period_lock, + EVENTS_NAME: period_events, + WATCHLIST_NAME: watchlist, + WEEKLY_NAME: weekly, + } + files[MANIFEST_NAME] = _canonical_json(_manifest_payload(lock, contract, files), "manifest_serialization_invalid") + return parse_weekly_artifact(files) + + +def parse_weekly_artifact(files: Mapping[str, bytes]) -> dict[str, bytes]: + if not isinstance(files, Mapping) or set(files) != set(ARTIFACT_FILES): + raise _invalid("artifact_file_set_invalid") + if any(type(files[name]) is not bytes for name in ARTIFACT_FILES): + raise _invalid("artifact_member_invalid") + try: + lock = parse_period_lock_bytes(files[PERIOD_LOCK_NAME]) + except (PeriodLockError, TypeError, ValueError): + raise _invalid("period_lock_invalid") from None + weekly_payload = _parse_canonical_json(files[WEEKLY_NAME], "weekly_wire_invalid") + if not isinstance(weekly_payload, Mapping): + raise _invalid("weekly_shape_invalid") + try: + contract = parse_weekly_contract(weekly_payload) + except (WeeklyContractError, TypeError, ValueError): + raise _invalid("weekly_contract_invalid") from None + if serialize_weekly_contract(contract) != files[WEEKLY_NAME]: + raise _invalid("weekly_noncanonical") + expected_events = _filter_events(files[EVENTS_NAME], contract.period_start, contract.as_of) + if expected_events != files[EVENTS_NAME]: + raise _invalid("events_period_mismatch") + event_rows, _ = _csv_snapshot(files[EVENTS_NAME], EVENT_HEADER, "events_csv_invalid", allow_empty=True) + watch_rows, _ = _csv_snapshot(files[WATCHLIST_NAME], WATCHLIST_HEADER, "watchlist_csv_invalid") + expected_artifacts = ( + (EVENTS_NAME, _sha256(files[EVENTS_NAME]), event_rows), + (WATCHLIST_NAME, _sha256(files[WATCHLIST_NAME]), watch_rows), + ) + actual_lock_artifacts = tuple((item.path, item.sha256, item.row_count) for item in lock.source_artifacts) + actual_contract_artifacts = tuple((item.path, item.sha256, item.row_count) for item in contract.source_artifacts) + if actual_lock_artifacts != expected_artifacts or actual_contract_artifacts != expected_artifacts: + raise _invalid("source_artifact_mismatch") + expected_snapshot_id = f"rss_source_snapshot_{contract.as_of:%Y%m%d}_{lock.source_run_id}" + if ( + lock.period_start != contract.period_start + or lock.period_end_exclusive != contract.period_end_exclusive + or lock.as_of != contract.as_of + or lock.producer_ref != contract.producer_ref + or lock.source_provenance != contract.source_provenance + or lock.source_snapshot_id != expected_snapshot_id + or lock.source_attempt != 1 + ): + raise _invalid("period_contract_mismatch") + manifest_payload = _parse_canonical_json(files[MANIFEST_NAME], "manifest_wire_invalid") + if not isinstance(manifest_payload, Mapping): + raise _invalid("manifest_shape_invalid") + expected_manifest = _manifest_payload(lock, contract, files) + if manifest_payload != expected_manifest or _canonical_json(manifest_payload, "manifest_wire_invalid") != files[MANIFEST_NAME]: + raise _invalid("manifest_mismatch") + return {name: files[name] for name in ARTIFACT_FILES} + + +def serialize_weekly_artifact(files: Mapping[str, bytes]) -> dict[str, bytes]: + """Validate and return the exact fixed file set in canonical order.""" + return parse_weekly_artifact(files) diff --git a/tests/test_rss_source_fetch.py b/tests/test_rss_source_fetch.py index a883395..033c4e5 100644 --- a/tests/test_rss_source_fetch.py +++ b/tests/test_rss_source_fetch.py @@ -105,6 +105,7 @@ def fake_fetch(url: str) -> bytes: payload = json.loads(status.read_text(encoding="utf-8")) assert payload["successful_feed_count"] == 1 assert payload["failed_feed_count"] == 1 + assert payload["complete"] is False assert payload["feeds"][1]["feed_id"] == "bad" assert "RuntimeError" in payload["feeds"][1]["error"] diff --git a/tests/test_weekly_artifact.py b/tests/test_weekly_artifact.py new file mode 100644 index 0000000..99e978b --- /dev/null +++ b/tests/test_weekly_artifact.py @@ -0,0 +1,179 @@ +from __future__ import annotations + +import json +from datetime import date, datetime, timezone + +import pytest + +from political_event_tracking_research.weekly_artifact import ( + ARTIFACT_NAME, + RETENTION_DAYS, + WeeklyArtifactError, + build_weekly_artifact, + completed_week_period, + parse_weekly_artifact, + serialize_weekly_artifact, +) + + +EVENTS = b"event_id,event_date,symbol,event_type,direction,confidence,source_url,notes\ne1,2026-07-08,AAPL,public_mention,neutral,high,https://example.test/e1,note\n" +WATCHLIST = b"symbol,name,bucket,research_status,thesis,source_url\nAAPL,Apple,named_mentioned,watchlist,thesis,https://example.test/aapl\n" + + +def feed_status(**overrides: object) -> dict[str, object]: + value: dict[str, object] = { + "feed_count": 2, + "successful_feed_count": 2, + "failed_feed_count": 0, + "stale_feed_count": 0, + "missing_feed_count": 0, + "complete": True, + } + value.update(overrides) + return value + + +def build(**overrides: object) -> dict[str, bytes]: + value: dict[str, object] = { + "period_start": date(2026, 7, 6), + "as_of": date(2026, 7, 12), + "generated_at": datetime(2026, 7, 13, 12, 45, tzinfo=timezone.utc), + "workflow_ref": "QuantStrategyLab/PoliticalEventTrackingResearch/.github/workflows/rss_source_pipeline.yml@refs/heads/main", + "source_run_id": "12345", + "producer_ref": "a" * 40, + "source_events": EVENTS, + "watchlist": WATCHLIST, + "feed_status": feed_status(), + "source_provenance": "official_rss_source_pipeline_v1", + "run_mode": "scheduled", + } + value.update(overrides) + return build_weekly_artifact(**value) + + +def test_build_is_exact_five_files_and_round_trips() -> None: + files = build() + assert tuple(files) == ( + "period_lock.json", + "political_events.csv", + "political_watchlist.csv", + "political_event_weekly.json", + "weekly_manifest.json", + ) + assert parse_weekly_artifact(files) == files + assert serialize_weekly_artifact(files) == files + + manifest = json.loads(files["weekly_manifest.json"]) + assert manifest["artifact_name"] == ARTIFACT_NAME + assert manifest["retention_days"] == RETENTION_DAYS + assert {item["name"] for item in manifest["files"]} == set(files) - {"weekly_manifest.json"} + + +def test_build_is_deterministic_for_same_inputs() -> None: + assert build() == build() + + +@pytest.mark.parametrize( + ("field", "value"), + [ + ("as_of", date(2026, 7, 11)), + ("generated_at", datetime(2026, 7, 12, 23, 0, tzinfo=timezone.utc)), + ], +) +def test_period_and_provenance_mismatch_fail_closed(field: str, value: object) -> None: + with pytest.raises(WeeklyArtifactError): + build(**{field: value}) + + +@pytest.mark.parametrize( + "status", + [ + feed_status(failed_feed_count=1, successful_feed_count=1), + feed_status(stale_feed_count=1), + feed_status(missing_feed_count=1), + {"feed_count": 0, "successful_feed_count": 0, "failed_feed_count": 0}, + ], +) +def test_incomplete_feed_status_does_not_build(status: dict[str, object]) -> None: + with pytest.raises(WeeklyArtifactError): + build(feed_status=status) + + +@pytest.mark.parametrize("field", ["source_events", "watchlist"]) +def test_missing_or_malformed_csv_does_not_build(field: str) -> None: + with pytest.raises(WeeklyArtifactError): + build(**{field: b"not,the,approved,header\n"}) + + +def test_events_are_filtered_to_locked_period_deterministically() -> None: + source = EVENTS + b"old,2026-07-05,MSFT,public_mention,neutral,high,https://example.test/old,note\n" + files = build(source_events=source) + assert files["political_events.csv"] == EVENTS + base_lock = json.loads(build()["period_lock.json"]) + filtered_lock = json.loads(files["period_lock.json"]) + assert filtered_lock["source_snapshot_digest"] != base_lock["source_snapshot_digest"] + manifest = json.loads(files["weekly_manifest.json"]) + event_file = next(item for item in manifest["files"] if item["name"] == "political_events.csv") + assert event_file["row_count"] == 1 + + +def test_event_date_outside_period_is_filtered_including_next_monday() -> None: + source = EVENTS + b"next,2026-07-13,MSFT,public_mention,neutral,high,https://example.test/next,note\n" + files = build(source_events=source) + assert files["political_events.csv"] == EVENTS + + +def test_malformed_event_date_fails_closed() -> None: + source = EVENTS + b"bad,not-a-date,MSFT,public_mention,neutral,high,https://example.test/bad,note\n" + with pytest.raises(WeeklyArtifactError) as error: + build(source_events=source) + assert error.value.code == "events_date_invalid" + + +def test_zero_event_week_is_allowed() -> None: + files = build(source_events=b"event_id,event_date,symbol,event_type,direction,confidence,source_url,notes\n") + manifest = json.loads(files["weekly_manifest.json"]) + event_file = next(item for item in manifest["files"] if item["name"] == "political_events.csv") + assert event_file["row_count"] == 0 + assert files["political_events.csv"].endswith(b"\n") + + +def test_incoming_complete_false_is_not_overridden() -> None: + with pytest.raises(WeeklyArtifactError): + build(feed_status=feed_status(complete=False)) + + +def test_manifest_file_tamper_is_rejected() -> None: + files = build() + tampered = dict(files) + tampered["political_events.csv"] += b"e2,2026-07-09,MSFT,public_mention,neutral,high,https://example.test/e2,note\n" + with pytest.raises(WeeklyArtifactError): + parse_weekly_artifact(tampered) + + +def test_extra_or_missing_file_is_rejected() -> None: + files = build() + with pytest.raises(WeeklyArtifactError): + parse_weekly_artifact({key: value for key, value in files.items() if key != "weekly_manifest.json"}) + with pytest.raises(WeeklyArtifactError): + parse_weekly_artifact({**files, "extra.txt": b"x"}) + + +def test_malformed_period_lock_is_sanitized() -> None: + files = build() + tampered = dict(files) + tampered["period_lock.json"] = b"{}" + with pytest.raises(WeeklyArtifactError) as error: + parse_weekly_artifact(tampered) + assert error.value.code == "period_lock_invalid" + + +def test_manifest_is_not_self_hash_bound() -> None: + manifest = json.loads(build()["weekly_manifest.json"]) + assert "weekly_manifest.json" not in {item["name"] for item in manifest["files"]} + + +def test_scheduled_period_is_previous_complete_week() -> None: + assert completed_week_period(date(2026, 7, 13)) == (date(2026, 7, 6), date(2026, 7, 13)) + with pytest.raises(WeeklyArtifactError): + completed_week_period(date(2026, 7, 12)) diff --git a/tests/test_weekly_producer_workflow.py b/tests/test_weekly_producer_workflow.py new file mode 100644 index 0000000..1bfc71e --- /dev/null +++ b/tests/test_weekly_producer_workflow.py @@ -0,0 +1,16 @@ +from pathlib import Path + + +WORKFLOW = Path(__file__).parents[1] / ".github/workflows/rss_source_pipeline.yml" + + +def test_legacy_live_publication_precedes_weekly_artifact_failure_boundary() -> None: + text = WORKFLOW.read_text(encoding="utf-8") + assert text.index("Publish live CSV outputs to repository") < text.index("Build completed weekly producer artifact") + + +def test_dedicated_artifact_is_separate_from_legacy_source_artifact() -> None: + text = WORKFLOW.read_text(encoding="utf-8") + assert "path: data/output/rss_source_pipeline/" in text + assert "path: data/output/political-event-weekly-v1/" in text + assert text.index("Upload RSS source artifact") < text.index("Upload political weekly artifact")