Add governed RSS and Atom connectors

2026-07-31 22:48:07 +02:00
parent c5a43b3dae
commit 52fe33568c
5 changed files with 757 additions and 1 deletions
@@ -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"""<?xml version="1.0"?>
+<rss version="2.0"><channel>
+  <title>Decisions</title><link>https://example.test/</link>
+  <description>Published decisions</description>
+  <item><guid>decision-1</guid><title>Decision one</title>
+    <link>https://example.test/1</link>
+    <pubDate>Fri, 31 Jul 2026 10:00:00 GMT</pubDate>
+    <category>planning</category>
+  </item>
+</channel></rss>"""
+
+ATOM = b"""<?xml version="1.0"?>
+<feed xmlns="http://www.w3.org/2005/Atom">
+  <id>https://example.test/feed</id><title>Updates</title>
+  <updated>2026-07-31T10:00:00Z</updated>
+  <link href="https://example.test/" />
+  <entry><id>update-1</id><title>Update one</title>
+    <updated>2026-07-31T10:00:00Z</updated>
+    <link href="https://example.test/update-1" />
+  </entry>
+</feed>"""
+
+
+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'<!DOCTYPE x [<!ENTITY y SYSTEM "file:///etc/passwd">]><rss>&y;</rss>'
+        with self.assertRaisesRegex(FeedCapabilityError, "not safe or valid"):
+            self.provider.parse(payload, source_url="https://example.test/rss")
+
+
+if __name__ == "__main__":
+    unittest.main()