diff --git a/src/political_event_tracking_research/feed_primitives.py b/src/political_event_tracking_research/feed_primitives.py new file mode 100644 index 0000000..64cd32c --- /dev/null +++ b/src/political_event_tracking_research/feed_primitives.py @@ -0,0 +1,297 @@ +from __future__ import annotations + +import hashlib +import json +import re +from collections.abc import Iterable, Mapping +from dataclasses import dataclass + + +STATUS_VERSION = "pert.feed_primitives.v1" +MAX_SAFE_JSON_INTEGER = 2**53 - 1 +MAX_ROWS_PER_FEED = 10_000 +_DIGEST_RE = re.compile(r"^[0-9a-f]{64}$") +_ERROR_RE = re.compile(r"^[a-z][a-z0-9_]{0,63}$") +_ROW_KEYS = frozenset({"item_id", "published_at", "source_type", "source_url", "author", "text"}) +_FEED_KEYS = frozenset({"feed_id", "feed_url", "kind", "state", "rows", "error_code"}) +_WIRE_KEYS = frozenset( + { + "status_version", + "configured_feed_count", + "feed_count", + "successful_feed_count", + "failed_feed_count", + "quarantined_feed_count", + "accepted_row_count", + "rejected_row_count", + "publication_complete", + "eligible_for_live_publication", + "aggregate_row_digest", + "feeds", + } +) +_FEED_WIRE_KEYS = frozenset( + { + "feed_id", + "feed_url", + "kind", + "state", + "accepted_row_count", + "rejected_row_count", + "row_digest", + "error_code", + } +) + + +class PrimitiveStatusError(ValueError): + def __init__(self, code: str): + super().__init__(code) + self.code = code + + +def _fail(code: str) -> None: + raise PrimitiveStatusError(code) + + +def _exact_str(value: object, code: str) -> str: + if type(value) is not str or not value or any(ord(char) < 0x20 for char in value): + _fail(code) + return value + + +def _safe_int(value: object, code: str) -> int: + if type(value) is not int or value < 0 or value > MAX_SAFE_JSON_INTEGER: + _fail(code) + return value + + +def _mapping(value: object, keys: frozenset[str], code: str) -> dict[str, object]: + if not isinstance(value, Mapping): + _fail(code) + try: + result = dict(value) + except (AttributeError, KeyError, OverflowError, TypeError, ValueError, RuntimeError, UnicodeError): + _fail(code) + if set(result) != keys or any(type(key) is not str for key in result): + _fail(code) + return result + + +@dataclass(frozen=True) +class PrimitiveRow: + item_id: str + published_at: str + source_type: str + source_url: str + author: str + text: str + + @classmethod + def from_mapping(cls, value: object) -> PrimitiveRow: + data = _mapping(value, _ROW_KEYS, "row_invalid") + return cls(*(_exact_str(data[key], "row_invalid") for key in _ROW_KEYS_ORDER)) + + def to_mapping(self) -> dict[str, str]: + return {key: getattr(self, key) for key in _ROW_KEYS_ORDER} + + +_ROW_KEYS_ORDER = ("item_id", "published_at", "source_type", "source_url", "author", "text") + + +def _row_sort_key(row: PrimitiveRow) -> tuple[str, str]: + return row.published_at, row.item_id + + +def _canonical_bytes(value: object, code: str = "status_invalid") -> bytes: + try: + return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False).encode( + "utf-8" + ) + except (TypeError, ValueError, UnicodeError, RecursionError): + _fail(code) + + +def _digest(rows: list[PrimitiveRow]) -> str: + return hashlib.sha256(_canonical_bytes([row.to_mapping() for row in rows])).hexdigest() + + +def _validate_feed_record(value: object) -> tuple[dict[str, object], list[PrimitiveRow]]: + data = _mapping(value, _FEED_KEYS, "feed_invalid") + feed_id = _exact_str(data["feed_id"], "feed_invalid") + feed_url = _exact_str(data["feed_url"], "feed_invalid") + kind = data["kind"] + if type(kind) is not str or kind not in {"rss2", "atom", "unknown"}: + _fail("feed_kind_invalid") + state = data["state"] + if type(state) is not str or state not in {"accepted", "failed", "quarantined"}: + _fail("feed_state_invalid") + rows_value = data["rows"] + if not isinstance(rows_value, (list, tuple)): + _fail("feed_rows_invalid") + if len(rows_value) > MAX_ROWS_PER_FEED: + _fail("feed_rows_invalid") + rows = [PrimitiveRow.from_mapping(row) for row in rows_value] + error = data["error_code"] + if error is not None and (type(error) is not str or not _ERROR_RE.fullmatch(error)): + _fail("feed_error_invalid") + if state == "accepted" and (not rows or error is not None or kind == "unknown"): + _fail("feed_state_invalid") + if state in {"failed", "quarantined"} and (rows or error is None or kind == "unknown" and state == "quarantined"): + _fail("feed_state_invalid") + if state == "failed" and kind not in {"rss2", "atom", "unknown"}: + _fail("feed_state_invalid") + return {"feed_id": feed_id, "feed_url": feed_url, "kind": kind, "state": state, "error_code": error}, rows + + +def _feed_wire(data: dict[str, object], rows: list[PrimitiveRow]) -> dict[str, object]: + accepted = data["state"] == "accepted" + ordered = sorted(rows, key=_row_sort_key) if accepted else [] + return { + "feed_id": data["feed_id"], + "feed_url": data["feed_url"], + "kind": data["kind"], + "state": data["state"], + "accepted_row_count": len(ordered), + "rejected_row_count": 0, + "row_digest": _digest(ordered) if accepted else hashlib.sha256(b"[]").hexdigest(), + "error_code": data["error_code"], + } + + +def build_status(feed_records: Iterable[Mapping[str, object]]) -> dict[str, object]: + try: + records = list(feed_records) + except (TypeError, ValueError, RuntimeError): + _fail("status_invalid") + parsed = [_validate_feed_record(record) for record in records] + if len({item[0]["feed_id"] for item in parsed}) != len(parsed): + _fail("feed_duplicate") + parsed.sort(key=lambda item: (item[0]["feed_id"], item[0]["feed_url"])) + wires = [_feed_wire(data, rows) for data, rows in parsed] + accepted_rows = sorted( + [row for data, rows in parsed if data["state"] == "accepted" for row in rows], key=_row_sort_key + ) + failed_count = sum(data["state"] == "failed" for data, _ in parsed) + quarantined_count = sum(data["state"] == "quarantined" for data, _ in parsed) + complete = bool(parsed) and failed_count == 0 and quarantined_count == 0 and all( + data["state"] == "accepted" for data, _ in parsed + ) + return { + "status_version": STATUS_VERSION, + "configured_feed_count": len(parsed), + "feed_count": len(parsed), + "successful_feed_count": len(parsed) - failed_count - quarantined_count, + "failed_feed_count": failed_count, + "quarantined_feed_count": quarantined_count, + "accepted_row_count": len(accepted_rows), + "rejected_row_count": 0, + "publication_complete": complete, + "eligible_for_live_publication": complete, + "aggregate_row_digest": _digest(accepted_rows), + "feeds": wires, + } + + +def _validate_wire(value: object) -> dict[str, object]: + data = _mapping(value, _WIRE_KEYS, "status_invalid") + if data["status_version"] != STATUS_VERSION: + _fail("status_version_invalid") + for key in _WIRE_KEYS - { + "status_version", + "publication_complete", + "eligible_for_live_publication", + "aggregate_row_digest", + "feeds", + }: + _safe_int(data[key], "status_counter_invalid") + for key in ("publication_complete", "eligible_for_live_publication"): + if type(data[key]) is not bool: + _fail("status_flag_invalid") + digest = _exact_str(data["aggregate_row_digest"], "status_digest_invalid") + if not _DIGEST_RE.fullmatch(digest): + _fail("status_digest_invalid") + feeds = data["feeds"] + if not isinstance(feeds, list): + _fail("feed_invalid") + previous: tuple[str, str] | None = None + feed_ids: set[str] = set() + for feed in feeds: + item = _mapping(feed, _FEED_WIRE_KEYS, "feed_invalid") + key = (_exact_str(item["feed_id"], "feed_invalid"), _exact_str(item["feed_url"], "feed_invalid")) + if key[0] in feed_ids: + _fail("feed_duplicate") + feed_ids.add(key[0]) + if previous is not None and key <= previous: + _fail("feed_order_invalid") + previous = key + if type(item["kind"]) is not str or item["kind"] not in {"rss2", "atom", "unknown"}: + _fail("feed_kind_invalid") + if type(item["state"]) is not str or item["state"] not in {"accepted", "failed", "quarantined"}: + _fail("feed_state_invalid") + _safe_int(item["accepted_row_count"], "feed_counter_invalid") + _safe_int(item["rejected_row_count"], "feed_counter_invalid") + row_digest = _exact_str(item["row_digest"], "status_digest_invalid") + if not _DIGEST_RE.fullmatch(row_digest): + _fail("status_digest_invalid") + error = item["error_code"] + if error is not None and (type(error) is not str or not _ERROR_RE.fullmatch(error)): + _fail("feed_error_invalid") + if data["configured_feed_count"] != len(feeds) or data["feed_count"] != len(feeds): + _fail("status_counter_invalid") + accepted = sum(item["state"] == "accepted" for item in feeds) + failed = sum(item["state"] == "failed" for item in feeds) + quarantined = sum(item["state"] == "quarantined" for item in feeds) + accepted_rows = sum(item["accepted_row_count"] for item in feeds) + rejected_rows = sum(item["rejected_row_count"] for item in feeds) + if data["successful_feed_count"] != accepted or data["failed_feed_count"] != failed: + _fail("status_counter_invalid") + if data["quarantined_feed_count"] != quarantined: + _fail("status_counter_invalid") + if data["accepted_row_count"] != accepted_rows or data["rejected_row_count"] != rejected_rows: + _fail("status_counter_invalid") + for item in feeds: + if item["state"] == "accepted": + if item["accepted_row_count"] == 0 or item["error_code"] is not None: + _fail("feed_state_invalid") + elif item["accepted_row_count"] != 0 or item["error_code"] is None: + _fail("feed_state_invalid") + complete = bool(feeds) and failed == 0 and quarantined == 0 and accepted == len(feeds) + if data["publication_complete"] != complete or data["eligible_for_live_publication"] != complete: + _fail("status_flag_invalid") + return data + + +def serialize_status(status: Mapping[str, object]) -> bytes: + data = _validate_wire(status) + return _canonical_bytes(data) + + +def parse_status_bytes(payload: bytes) -> dict[str, object]: + if type(payload) is not bytes: + _fail("status_bytes_invalid") + try: + value = json.loads(payload.decode("utf-8"), object_pairs_hook=_reject_duplicate_keys) + except (UnicodeError, json.JSONDecodeError, RecursionError): + _fail("status_bytes_invalid") + data = _validate_wire(value) + if serialize_status(data) != payload: + _fail("status_noncanonical") + return data + + +def _reject_duplicate_keys(pairs: list[tuple[str, object]]) -> dict[str, object]: + result: dict[str, object] = {} + for key, value in pairs: + if key in result: + _fail("status_duplicate_key") + result[key] = value + return result + + +def status_for_rows(payload: bytes, feed_records: Iterable[Mapping[str, object]]) -> dict[str, object]: + actual = parse_status_bytes(payload) + expected = build_status(feed_records) + if actual != expected: + _fail("status_integrity_mismatch") + return actual diff --git a/src/political_event_tracking_research/rss_source_fetch.py b/src/political_event_tracking_research/rss_source_fetch.py index 0884243..fdb7834 100644 --- a/src/political_event_tracking_research/rss_source_fetch.py +++ b/src/political_event_tracking_research/rss_source_fetch.py @@ -5,17 +5,18 @@ import email.utils import hashlib import html -import json import re import urllib.request from collections.abc import Callable from dataclasses import dataclass +from types import MappingProxyType from pathlib import Path import defusedxml.ElementTree as ET from defusedxml.common import DefusedXmlException from .csv_utils import read_csv_rows, write_csv_rows +from .feed_primitives import PrimitiveRow, build_status, serialize_status USER_AGENT = ( @@ -55,6 +56,18 @@ class FeedXmlError(ValueError): """Sanitized producer-boundary XML failure.""" +def _is_foreign_tag(tag: object, namespace: str | None) -> bool: + if not isinstance(tag, str) or not tag.startswith("{"): + return False + return namespace is None or not tag.startswith("{" + namespace + "}") + + +@dataclass(frozen=True) +class ParsedFeed: + feed_kind: str + entries: tuple[MappingProxyType, ...] + + def load_feed_config(path: str | Path) -> list[FeedConfig]: feeds: list[FeedConfig] = [] for row in read_csv_rows(path): @@ -131,7 +144,7 @@ def stable_item_id(feed_id: str, link: str, title: str) -> str: return f"{feed_id}-{digest}" -def parse_feed_items(feed_bytes: bytes, feed: FeedConfig, *, max_items: int = 25) -> list[dict[str, str]]: +def parse_feed_snapshot(feed_bytes: bytes, feed: FeedConfig, *, max_items: int = 25) -> ParsedFeed: if type(feed_bytes) is not bytes: raise FeedXmlError("feed_xml_invalid") if len(feed_bytes) > MAX_XML_BYTES: @@ -140,16 +153,32 @@ def parse_feed_items(feed_bytes: bytes, feed: FeedConfig, *, max_items: int = 25 root = ET.fromstring(feed_bytes, forbid_dtd=True, forbid_entities=True, forbid_external=True) except (DefusedXmlException, ET.ParseError, LookupError, UnicodeError, ValueError, RecursionError): raise FeedXmlError("feed_xml_invalid") from None - rows: list[dict[str, str]] = [] - - rss_items = root.findall("./channel/item") - if rss_items: - for item in rss_items[:max_items]: - title = child_text(item, ("title",)) - link = rss_item_link(item) - published = child_text(item, ("pubDate", "{http://purl.org/dc/elements/1.1/}date")) - description = child_text(item, ("description", "{http://purl.org/rss/1.0/modules/content/}encoded")) - text = " ".join(part for part in (title, strip_html(description)) if part) + def parse_rows(items: list[ET.Element], kind: str) -> ParsedFeed: + rows: list[dict[str, str]] = [] + for item in items[:max_items]: + if kind == "rss2": + title = child_text(item, ("title",)) + link = rss_item_link(item) + published = child_text(item, ("pubDate", "{http://purl.org/dc/elements/1.1/}date")) + description = child_text(item, ("description", "{http://purl.org/rss/1.0/modules/content/}encoded")) + text = " ".join(part for part in (title, strip_html(description)) if part) + else: + title = child_text(item, ("{http://www.w3.org/2005/Atom}title", "title")) + link = atom_entry_link(item) + published = child_text( + item, + ( + "{http://www.w3.org/2005/Atom}published", + "{http://www.w3.org/2005/Atom}updated", + "published", + "updated", + ), + ) + summary = child_text( + item, + ("{http://www.w3.org/2005/Atom}summary", "{http://www.w3.org/2005/Atom}content"), + ) + text = " ".join(part for part in (title, strip_html(summary)) if part) rows.append( { "item_id": stable_item_id(feed.feed_id, link, title), @@ -160,47 +189,92 @@ def parse_feed_items(feed_bytes: bytes, feed: FeedConfig, *, max_items: int = 25 "text": text, } ) - return rows - - atom_entries = root.findall("{http://www.w3.org/2005/Atom}entry") - for entry in atom_entries[:max_items]: - title = child_text(entry, ("{http://www.w3.org/2005/Atom}title", "title")) - link = atom_entry_link(entry) - published = child_text( - entry, - ("{http://www.w3.org/2005/Atom}published", "{http://www.w3.org/2005/Atom}updated", "published", "updated"), - ) - summary = child_text(entry, ("{http://www.w3.org/2005/Atom}summary", "{http://www.w3.org/2005/Atom}content")) - text = " ".join(part for part in (title, strip_html(summary)) if part) - rows.append( - { - "item_id": stable_item_id(feed.feed_id, link, title), - "published_at": parse_datetime(published), - "source_type": feed.source_type, - "source_url": link, - "author": feed.author, - "text": text, - } - ) - return rows + return ParsedFeed(kind, tuple(MappingProxyType(row) for row in rows)) + + if root.tag == "rss": + if root.attrib.get("version") != "2.0": + raise FeedXmlError("feed_schema_invalid") + channels = root.findall("./channel") + if len(channels) != 1: + raise FeedXmlError("feed_schema_invalid") + channel = channels[0] + rss_metadata = { + "title", + "link", + "description", + "language", + "copyright", + "managingEditor", + "webMaster", + "pubDate", + "lastBuildDate", + "category", + "generator", + "docs", + "cloud", + "ttl", + "image", + "rating", + "textInput", + "skipHours", + "skipDays", + } + if any( + child.tag != "item" + and child.tag not in rss_metadata + and not _is_foreign_tag(child.tag, None) + for child in channel + ): + raise FeedXmlError("feed_schema_invalid") + try: + return parse_rows([child for child in channel if child.tag == "item"], "rss2") + except (TypeError, ValueError, UnicodeError, RecursionError): + raise FeedXmlError("feed_xml_invalid") from None + + if root.tag == "{http://www.w3.org/2005/Atom}feed": + atom_entry = "{http://www.w3.org/2005/Atom}entry" + atom_metadata = { + "{http://www.w3.org/2005/Atom}id", + "{http://www.w3.org/2005/Atom}title", + "{http://www.w3.org/2005/Atom}updated", + "{http://www.w3.org/2005/Atom}subtitle", + "{http://www.w3.org/2005/Atom}link", + "{http://www.w3.org/2005/Atom}author", + "{http://www.w3.org/2005/Atom}category", + "{http://www.w3.org/2005/Atom}contributor", + "{http://www.w3.org/2005/Atom}generator", + "{http://www.w3.org/2005/Atom}icon", + "{http://www.w3.org/2005/Atom}logo", + "{http://www.w3.org/2005/Atom}rights", + } + if any( + child.tag != atom_entry + and child.tag not in atom_metadata + and not _is_foreign_tag(child.tag, "http://www.w3.org/2005/Atom") + for child in root + ): + raise FeedXmlError("feed_schema_invalid") + try: + return parse_rows([child for child in root if child.tag == atom_entry], "atom") + except (TypeError, ValueError, UnicodeError, RecursionError): + raise FeedXmlError("feed_xml_invalid") from None + + raise FeedXmlError("feed_schema_invalid") + + +def parse_feed_items(feed_bytes: bytes, feed: FeedConfig, *, max_items: int = 25) -> list[dict[str, str]]: + return [dict(row) for row in parse_feed_snapshot(feed_bytes, feed, max_items=max_items).entries] def utc_now_iso() -> str: return dt.datetime.now(dt.UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z") -def write_fetch_status(path: str | Path, statuses: list[FeedFetchStatus], *, item_count: int) -> None: - payload = { - "generated_at": utc_now_iso(), - "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), - "item_count": item_count, - "feeds": [item.to_json() for item in statuses], - } +def write_fetch_status(path: str | Path, feed_records: list[dict[str, object]]) -> None: + payload = serialize_status(build_status(feed_records)) output_path = Path(path) output_path.parent.mkdir(parents=True, exist_ok=True) - output_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8") + output_path.write_bytes(payload) def fetch_rss_sources( @@ -213,40 +287,62 @@ def fetch_rss_sources( fetcher: Callable[[str], bytes] = fetch_url, ) -> list[dict[str, str]]: rows: list[dict[str, str]] = [] - statuses: list[FeedFetchStatus] = [] + feed_records: list[dict[str, object]] = [] for feed in load_feed_config(feeds_path): try: - feed_rows = parse_feed_items(fetcher(feed.feed_url), feed, max_items=max_items_per_feed) - except Exception as exc: - statuses.append( - FeedFetchStatus( - feed_id=feed.feed_id, - feed_url=feed.feed_url, - ok=False, - item_count=0, - error=f"{type(exc).__name__}: {exc}", - ) + parsed = parse_feed_snapshot(fetcher(feed.feed_url), feed, max_items=max_items_per_feed) + feed_rows = [PrimitiveRow.from_mapping(row).to_mapping() for row in parsed.entries] + state = "accepted" if feed_rows else "quarantined" + error_code = None if feed_rows else "zero_entries" + feed_records.append( + { + "feed_id": feed.feed_id, + "feed_url": feed.feed_url, + "kind": parsed.feed_kind, + "state": state, + "rows": feed_rows, + "error_code": error_code, + } + ) + if feed_rows: + rows.extend(feed_rows) + except ( + DefusedXmlException, + ET.ParseError, + LookupError, + OSError, + RuntimeError, + TypeError, + ValueError, + UnicodeError, + ) as exc: + code = getattr(exc, "code", "fetch_failed") + if type(code) is not str or not re.fullmatch(r"[a-z][a-z0-9_]{0,63}", code): + code = "fetch_failed" + feed_records.append( + { + "feed_id": feed.feed_id, + "feed_url": feed.feed_url, + "kind": "unknown", + "state": "failed", + "rows": [], + "error_code": code, + } ) if not continue_on_feed_error: raise - continue - rows.extend(feed_rows) - statuses.append( - FeedFetchStatus( - feed_id=feed.feed_id, - feed_url=feed.feed_url, - ok=True, - item_count=len(feed_rows), - ) - ) - if statuses and not any(item.ok for item in statuses): + status = build_status(feed_records) + if feed_records and not any(record["state"] == "accepted" for record in feed_records): if status_output: - write_fetch_status(status_output, statuses, item_count=0) - raise RuntimeError("all configured RSS/Atom feeds failed") + write_fetch_status(status_output, feed_records) + if any(record["state"] == "failed" for record in feed_records): + raise RuntimeError("all configured RSS/Atom feeds failed") + write_csv_rows(output_path, ["item_id", "published_at", "source_type", "source_url", "author", "text"], []) + return [] rows.sort(key=lambda row: (row["published_at"], row["item_id"])) write_csv_rows(output_path, ["item_id", "published_at", "source_type", "source_url", "author", "text"], rows) if status_output: - write_fetch_status(status_output, statuses, item_count=len(rows)) + write_fetch_status(status_output, feed_records) return rows diff --git a/tests/test_feed_primitives.py b/tests/test_feed_primitives.py new file mode 100644 index 0000000..1c51b00 --- /dev/null +++ b/tests/test_feed_primitives.py @@ -0,0 +1,97 @@ +from __future__ import annotations + +import json + +import pytest + +from political_event_tracking_research.feed_primitives import ( + MAX_SAFE_JSON_INTEGER, + PrimitiveStatusError, + build_status, + parse_status_bytes, + serialize_status, + status_for_rows, +) + + +ROW = { + "item_id": "feed-a-1", + "published_at": "2026-05-01T12:30:00Z", + "source_type": "official_remarks", + "source_url": "https://example.test/item/1", + "author": "Example", + "text": "Policy mention", +} + + +def feed( + feed_id: str, + *, + state: str = "accepted", + rows: list[dict[str, str]] | None = None, + error_code: str | None = None, +) -> dict[str, object]: + return { + "feed_id": feed_id, + "feed_url": f"https://example.test/{feed_id}", + "kind": "rss2", + "state": state, + "rows": rows if rows is not None else [ROW], + "error_code": error_code, + } + + +def test_quarantine_has_zero_rows_and_no_accepted_digest_contribution() -> None: + empty = feed("empty", state="quarantined", rows=[], error_code="zero_entries") + status = build_status([feed("good"), empty]) + good_only = build_status([feed("good")]) + assert status["accepted_row_count"] == good_only["accepted_row_count"] + assert status["aggregate_row_digest"] == good_only["aggregate_row_digest"] + assert status["publication_complete"] is False + assert status["eligible_for_live_publication"] is False + + +def test_quarantine_with_rows_and_failed_rows_are_rejected() -> None: + with pytest.raises(PrimitiveStatusError, match="feed_state_invalid"): + build_status([feed("empty", state="quarantined", error_code="zero_entries")]) + status = build_status([feed("bad", state="failed", rows=[], error_code="fetch_failed")]) + assert status["accepted_row_count"] == 0 + assert status["publication_complete"] is False + + +def test_status_roundtrip_and_feed_order_are_canonical() -> None: + status = build_status([feed("b"), feed("a")]) + wire = serialize_status(status) + assert parse_status_bytes(wire) == status + payload = json.loads(wire) + payload["feeds"] = list(reversed(payload["feeds"])) + with pytest.raises(PrimitiveStatusError, match="feed_order_invalid"): + serialize_status(payload) + + +def test_status_digest_binds_one_snapshot() -> None: + status = build_status([feed("a")]) + assert status_for_rows(serialize_status(status), [feed("a")]) == status + changed = {**ROW, "item_id": "changed"} + with pytest.raises(PrimitiveStatusError, match="status_integrity_mismatch"): + status_for_rows(serialize_status(status), [feed("a", rows=[changed])]) + + +def test_safe_integer_and_bool_are_rejected() -> None: + payload = json.loads(serialize_status(build_status([feed("a")]))) + payload["feed_count"] = MAX_SAFE_JSON_INTEGER + 1 + with pytest.raises(PrimitiveStatusError, match="status_counter_invalid"): + serialize_status(payload) + + +def test_wire_rejects_duplicate_feed_id_with_different_urls() -> None: + status = build_status([feed("a"), feed("b")]) + payload = json.loads(serialize_status(status)) + payload["feeds"][1]["feed_id"] = "a" + payload["feeds"][1]["feed_url"] = "https://example.test/other" + payload["feeds"] = sorted(payload["feeds"], key=lambda item: (item["feed_id"], item["feed_url"])) + with pytest.raises(PrimitiveStatusError, match="feed_duplicate"): + serialize_status(payload) + payload["feed_count"] = True + with pytest.raises(PrimitiveStatusError, match="status_counter_invalid"): + serialize_status(payload) diff --git a/tests/test_rss_source_fetch.py b/tests/test_rss_source_fetch.py index d1f324e..93eb91d 100644 --- a/tests/test_rss_source_fetch.py +++ b/tests/test_rss_source_fetch.py @@ -7,7 +7,12 @@ import pytest from political_event_tracking_research import rss_source_fetch -from political_event_tracking_research.rss_source_fetch import FeedConfig, fetch_rss_sources, parse_feed_items +from political_event_tracking_research.rss_source_fetch import ( + FeedConfig, + fetch_rss_sources, + parse_feed_items, + parse_feed_snapshot, +) def test_parse_rss_feed_items_to_source_items() -> None: @@ -70,6 +75,49 @@ def test_parse_atom_feed_items_to_source_items() -> None: assert "EVT2" in rows[0]["text"] +@pytest.mark.parametrize( + ("payload", "expected_kind"), + [ + (b"", "rss2"), + (b"", "atom"), + ], +) +def test_empty_rss_and_atom_preserve_explicit_kind(payload: bytes, expected_kind: str) -> None: + parsed = parse_feed_snapshot(payload, FeedConfig("x", "https://example.test", "official", "")) + assert parsed.feed_kind == expected_kind + assert parsed.entries == () + assert parse_feed_items(payload, FeedConfig("x", "https://example.test", "official", "")) == [] + + +@pytest.mark.parametrize( + "payload", + [ + b"", + b"", + b"", + b"", + ], +) +def test_near_match_roots_and_containers_fail_closed(payload: bytes) -> None: + with pytest.raises(ValueError, match="feed_schema_invalid"): + parse_feed_snapshot(payload, FeedConfig("x", "https://example.test", "official", "")) + + +def test_foreign_namespaced_metadata_is_allowed_without_relaxing_container_grammar() -> None: + rss = b"" + atom = b"" + feed = FeedConfig("x", "https://example.test", "official", "") + assert parse_feed_snapshot(rss, feed).feed_kind == "rss2" + assert parse_feed_snapshot(atom, feed).feed_kind == "atom" + + +def test_parsed_entries_are_shallow_immutable() -> None: + payload = b"x" + parsed = parse_feed_snapshot(payload, FeedConfig("x", "https://example.test", "official", "")) + with pytest.raises(TypeError): + parsed.entries[0]["text"] = "changed" + + @pytest.mark.parametrize( "payload", [ @@ -171,8 +219,8 @@ 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["feeds"][1]["feed_id"] == "bad" - assert "RuntimeError" in payload["feeds"][1]["error"] + assert payload["feeds"][0]["feed_id"] == "bad" + assert payload["feeds"][0]["error_code"] == "fetch_failed" def test_fetch_rss_sources_fails_when_all_feeds_fail(tmp_path: Path) -> None: @@ -194,3 +242,31 @@ def fake_fetch(_url: str) -> bytes: status_output=tmp_path / "status.json", fetcher=fake_fetch, ) + + +def test_fetch_rss_sources_preserves_empty_feed_kind_and_quarantines(tmp_path: Path) -> None: + feeds_path = tmp_path / "feeds.csv" + feeds_path.write_text( + "feed_id,feed_url,source_type,author\n" + "atom,https://example.invalid/atom,official,Example\n" + "rss,https://example.invalid/rss,official,Example\n", + encoding="utf-8", + ) + payloads = { + "https://example.invalid/atom": b"", + "https://example.invalid/rss": b"Empty", + } + + output = tmp_path / "source_items.csv" + status = tmp_path / "status.json" + rows = fetch_rss_sources(feeds_path, output, status_output=status, fetcher=payloads.__getitem__) + + assert rows == [] + assert output.read_text(encoding="utf-8") == "item_id,published_at,source_type,source_url,author,text\n" + payload = json.loads(status.read_text(encoding="utf-8")) + assert [(item["feed_id"], item["kind"], item["state"]) for item in payload["feeds"]] == [ + ("atom", "atom", "quarantined"), + ("rss", "rss2", "quarantined"), + ] + assert payload["accepted_row_count"] == 0 + assert payload["publication_complete"] is False