diff --git a/src/govoplan_connectors/backend/feeds.py b/src/govoplan_connectors/backend/feeds.py new file mode 100644 index 0000000..1711983 --- /dev/null +++ b/src/govoplan_connectors/backend/feeds.py @@ -0,0 +1,413 @@ +from __future__ import annotations + +import hashlib +import re +import xml.etree.ElementTree as ET +from collections.abc import Mapping +from dataclasses import replace +from datetime import datetime, timedelta, timezone +from email.utils import format_datetime, parsedate_to_datetime + +from defusedxml import ElementTree as SafeET +from defusedxml.common import DefusedXmlException + +from govoplan_core.core.feeds import ( + FeedCapabilityError, + FeedDocument, + FeedEntry, + FeedProvider, + FeedRenderRequest, + FeedRenderResult, +) +from govoplan_core.security.http_fetch import fetch_http + + +MAX_FEED_BYTES = 5_000_000 +ATOM_NS = "http://www.w3.org/2005/Atom" + + +class ConnectorFeedProvider(FeedProvider): + def fetch( + self, + url: str, + *, + timeout: float = 15, + max_entries: int = 2_000, + ) -> FeedDocument: + try: + response = fetch_http( + url, + timeout=timeout, + label="RSS/Atom feed URL", + headers={ + "Accept": ( + "application/atom+xml, application/rss+xml, " + "application/xml;q=0.9, text/xml;q=0.8" + ) + }, + max_bytes=MAX_FEED_BYTES, + ) + except Exception as exc: + raise FeedCapabilityError(f"Feed acquisition failed: {exc}") from exc + if response.status < 200 or response.status >= 300: + raise FeedCapabilityError( + f"Feed acquisition returned HTTP {response.status}." + ) + content_type = _header(response.headers, "content-type") + document = self.parse( + response.body, + source_url=url, + content_type=content_type, + max_entries=max_entries, + ) + acquired_at = datetime.now(timezone.utc) + return replace( + document, + acquired_at=acquired_at, + fresh_until=_fresh_until(response.headers, acquired_at), + etag=_header(response.headers, "etag"), + last_modified=_header(response.headers, "last-modified"), + metadata={ + **dict(document.metadata), + "http_status": response.status, + "byte_count": len(response.body), + }, + ) + + def parse( + self, + content: bytes, + *, + source_url: str, + content_type: str | None = None, + max_entries: int = 2_000, + ) -> FeedDocument: + if not content: + raise FeedCapabilityError("Feed content is empty.") + if len(content) > MAX_FEED_BYTES: + raise FeedCapabilityError( + f"Feeds are limited to {MAX_FEED_BYTES // 1_000_000} MB." + ) + try: + root = SafeET.fromstring(content) + except (ET.ParseError, DefusedXmlException) as exc: + raise FeedCapabilityError(f"Feed XML is not safe or valid: {exc}") from exc + local_name = _local_name(root.tag) + if local_name == "rss": + document = _parse_rss(root, source_url=source_url, max_entries=max_entries) + elif local_name == "feed": + document = _parse_atom(root, source_url=source_url, max_entries=max_entries) + else: + raise FeedCapabilityError("The document is neither an RSS nor an Atom feed.") + return replace( + document, + content_type=content_type, + sha256=hashlib.sha256(content).hexdigest(), + ) + + def render(self, request: FeedRenderRequest) -> FeedRenderResult: + if not request.title.strip() or not request.feed_url.strip(): + raise FeedCapabilityError("Feed title and feed URL are required.") + entries = tuple( + entry + for entry in request.entries + if entry.visibility in request.allowed_visibilities + ) + root = ( + _render_rss(request, entries) + if request.format == "rss" + else _render_atom(request, entries) + ) + body = ET.tostring(root, encoding="utf-8", xml_declaration=True) + return FeedRenderResult( + format=request.format, + content_type=( + "application/rss+xml; charset=utf-8" + if request.format == "rss" + else "application/atom+xml; charset=utf-8" + ), + body=body, + included_entries=len(entries), + excluded_entries=len(request.entries) - len(entries), + ) + + +def feed_rows(document: FeedDocument) -> tuple[Mapping[str, object], ...]: + """Map feed entries to the connector tabular shape used by Datasources.""" + + return tuple( + { + "id": entry.id, + "title": entry.title, + "url": entry.url, + "summary": entry.summary, + "content": entry.content, + "author": entry.author, + "published_at": ( + entry.published_at.isoformat() if entry.published_at else None + ), + "updated_at": entry.updated_at.isoformat() if entry.updated_at else None, + "categories": list(entry.categories), + "enclosures": [dict(item) for item in entry.enclosures], + } + for entry in document.entries + ) + + +def _parse_rss(root: ET.Element, *, source_url: str, max_entries: int) -> FeedDocument: + channel = _first_child(root, "channel") + if channel is None: + raise FeedCapabilityError("RSS feed is missing its channel element.") + entries: list[FeedEntry] = [] + for item in _children(channel, "item"): + if len(entries) >= max_entries: + raise FeedCapabilityError(f"Feeds are limited to {max_entries:,} entries.") + url = _text(item, "link") + identifier = _text(item, "guid") or url or _entry_fallback_id(item) + entries.append( + FeedEntry( + id=identifier, + title=_text(item, "title") or "(Untitled)", + url=url, + summary=_text(item, "description"), + content=_text(item, "encoded"), + author=_text(item, "author") or _text(item, "creator"), + published_at=_parse_date(_text(item, "pubDate")), + categories=tuple( + value for child in _children(item, "category") + if (value := (child.text or "").strip()) + ), + enclosures=tuple( + { + "url": child.attrib.get("url"), + "media_type": child.attrib.get("type"), + "size_bytes": _integer(child.attrib.get("length")), + } + for child in _children(item, "enclosure") + ), + ) + ) + return FeedDocument( + format="rss", + title=_text(channel, "title") or "Untitled feed", + source_url=source_url, + entries=tuple(entries), + description=_text(channel, "description"), + home_url=_text(channel, "link"), + language=_text(channel, "language"), + updated_at=_parse_date( + _text(channel, "lastBuildDate") or _text(channel, "pubDate") + ), + ) + + +def _parse_atom(root: ET.Element, *, source_url: str, max_entries: int) -> FeedDocument: + entries: list[FeedEntry] = [] + for item in _children(root, "entry"): + if len(entries) >= max_entries: + raise FeedCapabilityError(f"Feeds are limited to {max_entries:,} entries.") + alternate = _atom_link(item, "alternate") + identifier = _text(item, "id") or alternate or _entry_fallback_id(item) + author = _first_child(item, "author") + entries.append( + FeedEntry( + id=identifier, + title=_text(item, "title") or "(Untitled)", + url=alternate, + summary=_text(item, "summary"), + content=_text(item, "content"), + author=_text(author, "name") if author is not None else None, + published_at=_parse_date(_text(item, "published")), + updated_at=_parse_date(_text(item, "updated")), + categories=tuple( + value for child in _children(item, "category") + if (value := (child.attrib.get("term") or "").strip()) + ), + enclosures=tuple( + { + "url": child.attrib.get("href"), + "media_type": child.attrib.get("type"), + "size_bytes": _integer(child.attrib.get("length")), + } + for child in _children(item, "link") + if child.attrib.get("rel") == "enclosure" + ), + ) + ) + return FeedDocument( + format="atom", + title=_text(root, "title") or "Untitled feed", + source_url=source_url, + entries=tuple(entries), + description=_text(root, "subtitle"), + home_url=_atom_link(root, "alternate"), + updated_at=_parse_date(_text(root, "updated")), + ) + + +def _render_rss(request: FeedRenderRequest, entries: tuple[FeedEntry, ...]) -> ET.Element: + ET.register_namespace("atom", ATOM_NS) + root = ET.Element("rss", {"version": "2.0"}) + channel = ET.SubElement(root, "channel") + _element(channel, "title", request.title) + _element(channel, "link", request.home_url) + _element(channel, "description", request.description or request.title) + _element(channel, f"{{{ATOM_NS}}}link", None, { + "href": request.feed_url, + "rel": "self", + "type": "application/rss+xml", + }) + if request.language: + _element(channel, "language", request.language) + for entry in entries: + item = ET.SubElement(channel, "item") + _element(item, "guid", entry.id, {"isPermaLink": "false"}) + _element(item, "title", entry.title) + if entry.url: + _element(item, "link", entry.url) + if entry.summary or entry.content: + _element(item, "description", entry.summary or entry.content) + if entry.author: + _element(item, "author", entry.author) + date = entry.published_at or entry.updated_at + if date: + _element(item, "pubDate", format_datetime(_utc(date))) + for category in entry.categories: + _element(item, "category", category) + for enclosure in entry.enclosures: + attributes = { + "url": str(enclosure.get("url") or ""), + "type": str(enclosure.get("media_type") or "application/octet-stream"), + "length": str(enclosure.get("size_bytes") or 0), + } + if attributes["url"]: + _element(item, "enclosure", None, attributes) + return root + + +def _render_atom(request: FeedRenderRequest, entries: tuple[FeedEntry, ...]) -> ET.Element: + ET.register_namespace("", ATOM_NS) + root = ET.Element(f"{{{ATOM_NS}}}feed") + _element(root, f"{{{ATOM_NS}}}id", request.feed_url) + _element(root, f"{{{ATOM_NS}}}title", request.title) + _element(root, f"{{{ATOM_NS}}}link", None, {"href": request.home_url}) + _element( + root, + f"{{{ATOM_NS}}}link", + None, + {"href": request.feed_url, "rel": "self", "type": "application/atom+xml"}, + ) + latest = max( + (date for entry in entries for date in (entry.updated_at, entry.published_at) if date), + default=datetime.now(timezone.utc), + ) + _element(root, f"{{{ATOM_NS}}}updated", _utc(latest).isoformat().replace("+00:00", "Z")) + if request.description: + _element(root, f"{{{ATOM_NS}}}subtitle", request.description) + for value in entries: + entry = ET.SubElement(root, f"{{{ATOM_NS}}}entry") + _element(entry, f"{{{ATOM_NS}}}id", value.id) + _element(entry, f"{{{ATOM_NS}}}title", value.title) + if value.url: + _element(entry, f"{{{ATOM_NS}}}link", None, {"href": value.url}) + if value.summary: + _element(entry, f"{{{ATOM_NS}}}summary", value.summary) + if value.content: + _element(entry, f"{{{ATOM_NS}}}content", value.content, {"type": "html"}) + updated = value.updated_at or value.published_at or latest + _element(entry, f"{{{ATOM_NS}}}updated", _utc(updated).isoformat().replace("+00:00", "Z")) + if value.published_at: + _element(entry, f"{{{ATOM_NS}}}published", _utc(value.published_at).isoformat().replace("+00:00", "Z")) + if value.author: + author = ET.SubElement(entry, f"{{{ATOM_NS}}}author") + _element(author, f"{{{ATOM_NS}}}name", value.author) + for category in value.categories: + _element(entry, f"{{{ATOM_NS}}}category", None, {"term": category}) + return root + + +def _children(element: ET.Element, name: str) -> tuple[ET.Element, ...]: + return tuple(child for child in element if _local_name(child.tag) == name) + + +def _first_child(element: ET.Element, name: str) -> ET.Element | None: + return next((child for child in element if _local_name(child.tag) == name), None) + + +def _text(element: ET.Element | None, name: str) -> str | None: + if element is None: + return None + child = _first_child(element, name) + if child is None: + return None + value = "".join(child.itertext()).strip() + return value or None + + +def _local_name(tag: str) -> str: + return tag.rsplit("}", 1)[-1].split(":", 1)[-1] + + +def _atom_link(element: ET.Element, relation: str) -> str | None: + for child in _children(element, "link"): + if (child.attrib.get("rel") or "alternate") == relation: + return child.attrib.get("href") + return None + + +def _parse_date(value: str | None) -> datetime | None: + if not value: + return None + try: + parsed = parsedate_to_datetime(value) + except (TypeError, ValueError, OverflowError): + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + return None + return _utc(parsed) + + +def _utc(value: datetime) -> datetime: + if value.tzinfo is None: + return value.replace(tzinfo=timezone.utc) + return value.astimezone(timezone.utc) + + +def _integer(value: str | None) -> int | None: + try: + return int(value) if value is not None else None + except ValueError: + return None + + +def _entry_fallback_id(element: ET.Element) -> str: + body = ET.tostring(element, encoding="utf-8") + return f"urn:sha256:{hashlib.sha256(body).hexdigest()}" + + +def _header(headers: Mapping[str, str], name: str) -> str | None: + lowered = name.casefold() + return next((value for key, value in headers.items() if key.casefold() == lowered), None) + + +def _fresh_until(headers: Mapping[str, str], acquired_at: datetime) -> datetime | None: + cache_control = _header(headers, "cache-control") or "" + match = re.search(r"(?:^|,)\s*max-age\s*=\s*(\d+)", cache_control, re.IGNORECASE) + if match: + return acquired_at + timedelta(seconds=int(match.group(1))) + return _parse_date(_header(headers, "expires")) + + +def _element( + parent: ET.Element, + tag: str, + text: str | None, + attributes: Mapping[str, str] | None = None, +) -> ET.Element: + child = ET.SubElement(parent, tag, dict(attributes or {})) + child.text = text + return child + + +__all__ = ["ConnectorFeedProvider", "MAX_FEED_BYTES", "feed_rows"] diff --git a/src/govoplan_connectors/backend/manifest.py b/src/govoplan_connectors/backend/manifest.py index d811ed8..3d34f4f 100644 --- a/src/govoplan_connectors/backend/manifest.py +++ b/src/govoplan_connectors/backend/manifest.py @@ -11,6 +11,7 @@ from govoplan_core.core.module_guards import ( persistent_table_uninstall_guard, ) from govoplan_core.core.datasources import CAPABILITY_DATASOURCE_ORIGINS +from govoplan_core.core.feeds import CAPABILITY_CONNECTORS_FEEDS from govoplan_core.core.modules import ( DocumentationTopic, MigrationSpec, @@ -46,6 +47,7 @@ from govoplan_connectors.backend.tabular_sources import ( from govoplan_connectors.backend.datasource_origins import ( ConnectorDatasourceOriginProvider, ) +from govoplan_connectors.backend.feeds import ConnectorFeedProvider MODULE_ID = "connectors" @@ -53,6 +55,7 @@ MODULE_VERSION = "0.1.14" TABULAR_SOURCE_INTERFACE_VERSION = "0.1.0" DATASOURCE_ORIGIN_INTERFACE_VERSION = "0.1.0" SANCTIONS_SNAPSHOT_INTERFACE_VERSION = "1.0.0" +FEED_INTERFACE_VERSION = "0.1.0" def _permission(scope: str, label: str, description: str) -> PermissionDefinition: @@ -138,6 +141,10 @@ def _sanctions_snapshot_provider( return SqlSanctionsSnapshotProvider() +def _feed_provider(_context) -> ConnectorFeedProvider: + return ConnectorFeedProvider() + + def _tenant_summary(session, tenant_id: str) -> dict[str, int]: return { "connector_tabular_sources": ( @@ -175,6 +182,9 @@ manifest = ModuleManifest( "audit", "files", "policy", + "datasources", + "portal", + "reporting", "risk_compliance", ), required_capabilities=( @@ -198,6 +208,10 @@ manifest = ModuleManifest( name="connectors.sanctions_snapshots", version=SANCTIONS_SNAPSHOT_INTERFACE_VERSION, ), + ModuleInterfaceProvider( + name="connectors.feeds", + version=FEED_INTERFACE_VERSION, + ), ), permissions=PERMISSIONS, role_templates=ROLE_TEMPLATES, @@ -209,6 +223,7 @@ manifest = ModuleManifest( CAPABILITY_CONNECTORS_SANCTIONS_SNAPSHOTS: ( _sanctions_snapshot_provider ), + CAPABILITY_CONNECTORS_FEEDS: _feed_provider, }, tenant_summary_providers=(_tenant_summary,), migration_spec=MigrationSpec( @@ -254,6 +269,25 @@ manifest = ModuleManifest( related_modules=("dataflow", "files", "reporting", "risk_compliance"), order=40, ), + DocumentationTopic( + id="connectors.rss-atom", + title="RSS and Atom feeds", + summary="Import governed feed snapshots and emit visibility-filtered feeds.", + body=( + "Connectors owns bounded, SSRF-protected RSS/Atom transport and XML " + "parsing. Imported entries become immutable tabular snapshots exposed " + "through Datasources, including acquisition, freshness, ETag, content " + "digest, and source provenance. Portal or Reporting owns publication " + "routes and must pass the allowed visibility set when rendering output. " + "A separate RSS module is only warranted if GovOPlaN later needs a " + "dedicated feed-reader product surface." + ), + layer="available", + documentation_types=("admin", "user"), + audience=("operator", "module_admin", "power_user"), + related_modules=("datasources", "dataflow", "portal", "reporting"), + order=42, + ), DocumentationTopic( id="connectors.sanctions-snapshots", title="Sanctions source snapshots", diff --git a/src/govoplan_connectors/backend/router.py b/src/govoplan_connectors/backend/router.py index a16259c..4ab0bd0 100644 --- a/src/govoplan_connectors/backend/router.py +++ b/src/govoplan_connectors/backend/router.py @@ -1,6 +1,8 @@ from __future__ import annotations -from fastapi import APIRouter, Depends, HTTPException, Query, status +from dataclasses import asdict + +from fastapi import APIRouter, Depends, HTTPException, Query, Response, status from sqlalchemy.orm import Session from govoplan_core.audit.logging import audit_event @@ -13,9 +15,18 @@ from govoplan_core.core.tabular_sources import ( TabularSourceError, TabularSourceNotFoundError, ) +from govoplan_core.core.feeds import ( + FeedCapabilityError, + FeedEntry, + FeedRenderRequest, +) from govoplan_core.core.sanctions import SanctionsSnapshotReference from govoplan_core.db.session import get_session from govoplan_connectors.backend.schemas import ( + FeedAcquireRequest, + FeedDocumentResponse, + FeedImportRequest, + FeedRenderPayload, SanctionsAcquisitionRunListResponse, SanctionsAcquisitionRunResponse, SanctionsRefreshResponse, @@ -30,6 +41,7 @@ from govoplan_connectors.backend.schemas import ( TabularSourcePreviewResponse, TabularSourceResponse, ) +from govoplan_connectors.backend.feeds import ConnectorFeedProvider, feed_rows from govoplan_connectors.backend.sanctions_sources import ( SANCTIONS_READ_SCOPE, SANCTIONS_REFRESH_SCOPE, @@ -50,6 +62,7 @@ from govoplan_connectors.backend.tabular_sources import ( router = APIRouter(prefix="/connectors", tags=["connectors"]) provider = SqlTabularSourceProvider() sanctions_provider = SqlSanctionsSnapshotProvider() +feed_transport = ConnectorFeedProvider() def _require_any_scope(principal: ApiPrincipal, *scopes: str) -> None: @@ -88,6 +101,132 @@ def _sanctions_http_error( ) +def _feed_http_error(exc: FeedCapabilityError) -> HTTPException: + return HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, + detail=str(exc), + ) + + +@router.post("/feeds/preview", response_model=FeedDocumentResponse) +def api_preview_feed( + payload: FeedAcquireRequest, + principal: ApiPrincipal = Depends(get_api_principal), +) -> FeedDocumentResponse: + _require_any_scope(principal, READ_SCOPE, ADMIN_SCOPE) + try: + document = feed_transport.fetch( + payload.url, + max_entries=payload.max_entries, + ) + except FeedCapabilityError as exc: + raise _feed_http_error(exc) from exc + return FeedDocumentResponse.model_validate(asdict(document)) + + +@router.post( + "/feeds/import", + response_model=TabularSourceResponse, + status_code=status.HTTP_201_CREATED, +) +def api_import_feed_snapshot( + payload: FeedImportRequest, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> TabularSourceResponse: + _require_any_scope(principal, WRITE_SCOPE, ADMIN_SCOPE) + try: + document = feed_transport.fetch( + payload.url, + max_entries=payload.max_entries, + ) + source = provider.create_snapshot( + session, + principal, + snapshot=TabularSnapshotInput( + name=payload.name, + source_name=payload.source_name, + description=payload.description or document.description, + rows=feed_rows(document), + metadata={ + "import_format": document.format, + "feed": { + "source_url": document.source_url, + "home_url": document.home_url, + "acquired_at": ( + document.acquired_at.isoformat() + if document.acquired_at + else None + ), + "fresh_until": ( + document.fresh_until.isoformat() + if document.fresh_until + else None + ), + "etag": document.etag, + "last_modified": document.last_modified, + "content_type": document.content_type, + "sha256": document.sha256, + }, + }, + ), + ) + except (FeedCapabilityError, TabularSourceError) as exc: + if isinstance(exc, FeedCapabilityError): + raise _feed_http_error(exc) from exc + raise _http_error(exc) from exc + audit_event( + session, + tenant_id=principal.tenant_id, + user_id=getattr(principal.user, "id", None), + api_key_id=principal.api_key_id, + action="connectors.feed_snapshot.created", + object_type="connector_tabular_source", + object_id=source.ref, + details={ + "source_url": document.source_url, + "format": document.format, + "sha256": document.sha256, + "row_count": source.row_count, + }, + ) + session.commit() + return _source_response(source) + + +@router.post("/feeds/render") +def api_render_feed( + payload: FeedRenderPayload, + principal: ApiPrincipal = Depends(get_api_principal), +) -> Response: + _require_any_scope(principal, READ_SCOPE, ADMIN_SCOPE) + try: + rendered = feed_transport.render( + FeedRenderRequest( + format=payload.format, + title=payload.title, + feed_url=payload.feed_url, + home_url=payload.home_url, + description=payload.description, + language=payload.language, + entries=tuple( + FeedEntry(**item.model_dump()) for item in payload.entries + ), + allowed_visibilities=frozenset(payload.allowed_visibilities), + ) + ) + except FeedCapabilityError as exc: + raise _feed_http_error(exc) from exc + return Response( + content=rendered.body, + media_type=rendered.content_type, + headers={ + "X-GovOPlaN-Feed-Included": str(rendered.included_entries), + "X-GovOPlaN-Feed-Excluded": str(rendered.excluded_entries), + }, + ) + + @router.get("/tabular-sources", response_model=TabularSourceListResponse) def api_list_tabular_sources( query: str = Query(default="", max_length=200), diff --git a/src/govoplan_connectors/backend/schemas.py b/src/govoplan_connectors/backend/schemas.py index 1ebd26d..c22676a 100644 --- a/src/govoplan_connectors/backend/schemas.py +++ b/src/govoplan_connectors/backend/schemas.py @@ -6,6 +6,68 @@ from typing import Any, Literal from pydantic import BaseModel, Field, model_validator +class FeedAcquireRequest(BaseModel): + url: str = Field(min_length=1, max_length=2000) + max_entries: int = Field(default=2_000, ge=1, le=10_000) + + +class FeedImportRequest(FeedAcquireRequest): + name: str = Field(min_length=1, max_length=300) + source_name: str = Field( + min_length=1, + max_length=120, + pattern=r"^[A-Za-z_][A-Za-z0-9_]*$", + ) + description: str | None = Field(default=None, max_length=4000) + + +class FeedEntryPayload(BaseModel): + id: str = Field(min_length=1, max_length=2000) + title: str = Field(min_length=1, max_length=1000) + url: str | None = Field(default=None, max_length=2000) + summary: str | None = None + content: str | None = None + author: str | None = Field(default=None, max_length=500) + published_at: datetime | None = None + updated_at: datetime | None = None + categories: list[str] = Field(default_factory=list, max_length=100) + enclosures: list[dict[str, Any]] = Field(default_factory=list, max_length=100) + visibility: Literal["public", "tenant", "private"] = "public" + metadata: dict[str, Any] = Field(default_factory=dict) + + +class FeedDocumentResponse(BaseModel): + format: Literal["rss", "atom"] + title: str + source_url: str + description: str | None = None + home_url: str | None = None + language: str | None = None + updated_at: datetime | None = None + acquired_at: datetime | None = None + fresh_until: datetime | None = None + etag: str | None = None + last_modified: str | None = None + content_type: str | None = None + sha256: str + entries: list[FeedEntryPayload] + metadata: dict[str, Any] = Field(default_factory=dict) + + +class FeedRenderPayload(BaseModel): + format: Literal["rss", "atom"] + title: str = Field(min_length=1, max_length=1000) + feed_url: str = Field(min_length=1, max_length=2000) + home_url: str = Field(min_length=1, max_length=2000) + description: str | None = None + language: str | None = Field(default=None, max_length=100) + entries: list[FeedEntryPayload] = Field(default_factory=list, max_length=10_000) + allowed_visibilities: list[Literal["public", "tenant", "private"]] = Field( + default_factory=lambda: ["public"], + max_length=3, + ) + + class SnapshotCreateRequest(BaseModel): name: str = Field(min_length=1, max_length=300) source_name: str = Field( @@ -140,6 +202,11 @@ class SanctionsRefreshResponse(BaseModel): __all__ = [ + "FeedAcquireRequest", + "FeedDocumentResponse", + "FeedEntryPayload", + "FeedImportRequest", + "FeedRenderPayload", "SnapshotCreateRequest", "SanctionsAcquisitionRunListResponse", "SanctionsAcquisitionRunResponse", diff --git a/tests/test_feeds.py b/tests/test_feeds.py new file mode 100644 index 0000000..d04ee3e --- /dev/null +++ b/tests/test_feeds.py @@ -0,0 +1,103 @@ +from __future__ import annotations + +import unittest +from unittest.mock import patch + +from defusedxml import ElementTree as SafeET + +from govoplan_connectors.backend.feeds import ConnectorFeedProvider, feed_rows +from govoplan_core.core.feeds import ( + FeedCapabilityError, + FeedEntry, + FeedRenderRequest, +) +from govoplan_core.security.http_fetch import HttpFetchResponse + + +RSS = b""" + + Decisionshttps://example.test/ + Published decisions + decision-1Decision one + https://example.test/1 + Fri, 31 Jul 2026 10:00:00 GMT + planning + +""" + +ATOM = b""" + + https://example.test/feedUpdates + 2026-07-31T10:00:00Z + + update-1Update one + 2026-07-31T10:00:00Z + + +""" + + +class ConnectorFeedProviderTests(unittest.TestCase): + def setUp(self) -> None: + self.provider = ConnectorFeedProvider() + + def test_rss_and_atom_are_normalized_to_tabular_entries(self) -> None: + rss = self.provider.parse(RSS, source_url="https://example.test/rss") + atom = self.provider.parse(ATOM, source_url="https://example.test/atom") + + self.assertEqual("rss", rss.format) + self.assertEqual("decision-1", rss.entries[0].id) + self.assertEqual("atom", atom.format) + self.assertEqual("https://example.test/update-1", atom.entries[0].url) + self.assertEqual("decision-1", feed_rows(rss)[0]["id"]) + self.assertEqual(64, len(rss.sha256)) + + def test_fetch_records_transport_freshness_and_provenance(self) -> None: + with patch( + "govoplan_connectors.backend.feeds.fetch_http", + return_value=HttpFetchResponse( + status=200, + headers={ + "Content-Type": "application/rss+xml", + "Cache-Control": "public, max-age=600", + "ETag": '"feed-1"', + }, + body=RSS, + ), + ): + document = self.provider.fetch("https://example.test/rss") + + self.assertEqual('"feed-1"', document.etag) + self.assertIsNotNone(document.acquired_at) + self.assertEqual(600, int((document.fresh_until - document.acquired_at).total_seconds())) + self.assertEqual(len(RSS), document.metadata["byte_count"]) + + def test_render_filters_entries_by_explicit_visibility(self) -> None: + request = FeedRenderRequest( + format="atom", + title="Public updates", + feed_url="https://example.test/feed.atom", + home_url="https://example.test/", + entries=( + FeedEntry(id="public-1", title="Public", visibility="public"), + FeedEntry(id="tenant-1", title="Tenant", visibility="tenant"), + ), + allowed_visibilities=frozenset({"public"}), + ) + result = self.provider.render(request) + root = SafeET.fromstring(result.body) + + self.assertEqual(1, result.included_entries) + self.assertEqual(1, result.excluded_entries) + self.assertIn(b"Public", result.body) + self.assertNotIn(b"Tenant", result.body) + self.assertTrue(root.tag.endswith("feed")) + + def test_unsafe_xml_is_rejected(self) -> None: + payload = b']>&y;' + with self.assertRaisesRegex(FeedCapabilityError, "not safe or valid"): + self.provider.parse(payload, source_url="https://example.test/rss") + + +if __name__ == "__main__": + unittest.main()