gwoe-antragspruefer/tests/test_rss.py

"""Tests fuer den Atom-Feed-Endpoint /api/feed.xml (#125).

Backfill aus #134: vorher nur indirekt im Smoke-Test abgedeckt. Hier:
- Atom-1.0-Validitaet (XML well-formed, Pflicht-Elemente)
- Filter-Parameter wirken (bundesland, partei)
- ETag-Header + 304-Verhalten
- Limit-Clamping
- HTML-Escaping fuer Sonderzeichen in Titeln/Drucksachen
"""
from __future__ import annotations

import pytest
from unittest.mock import patch
from xml.etree import ElementTree as ET

try:
    from fastapi.testclient import TestClient
    from app.main import app
    client = TestClient(app)
    _HAS_APP = True
except ImportError:
    _HAS_APP = False
    client = None


pytestmark = pytest.mark.skipif(not _HAS_APP, reason="app.main not importable")

ATOM_NS = "{http://www.w3.org/2005/Atom}"


def _fake_assessments() -> list[dict]:
    """Drei Fixture-Assessments mit allen Feldern, die der Feed nutzt."""
    return [
        {
            "drucksache": "21/1234",
            "title": "Antrag zu Erneuerbaren Energien",
            "bundesland": "NRW",
            "fraktionen": ["GRÜNE", "SPD"],
            "gwoe_score": 7.5,
            "empfehlung": "Unterstützen mit Änderungen",
            "antrag_zusammenfassung": "Solarpflicht für Neubauten",
            "updated_at": "2026-04-25T10:00:00",
        },
        {
            "drucksache": "8/4242",
            "title": "Anti-Terror-Paket & Überwachung",  # Sonderzeichen
            "bundesland": "MV",
            "fraktionen": ["CDU"],
            "gwoe_score": 2.1,
            "empfehlung": "Ablehnen",
            "antrag_zusammenfassung": None,
            "updated_at": "2026-04-24T08:30:00",
        },
        {
            "drucksache": "19/9999",
            "title": "Bürger:innen-Beteiligung stärken",
            "bundesland": "BE",
            "fraktionen": ["LINKE", "GRÜNE"],
            "gwoe_score": 9.0,
            "empfehlung": "Uneingeschränkt unterstützen",
            "antrag_zusammenfassung": "Bürgerräte etablieren",
            "updated_at": "2026-04-26T12:15:00",
        },
    ]


class TestFeedXml:
    def test_returns_atom_xml(self):
        with patch("app.main.get_all_assessments", return_value=_fake_assessments()):
            resp = client.get("/api/feed.xml")
        assert resp.status_code == 200
        assert "atom+xml" in resp.headers["content-type"]
        # XML well-formed
        root = ET.fromstring(resp.content)
        assert root.tag == f"{ATOM_NS}feed"

    def test_required_atom_elements_present(self):
        with patch("app.main.get_all_assessments", return_value=_fake_assessments()):
            resp = client.get("/api/feed.xml")
        root = ET.fromstring(resp.content)
        # Pflicht-Top-Level-Elemente nach RFC 4287
        for tag in ("id", "title", "updated"):
            assert root.find(f"{ATOM_NS}{tag}") is not None, f"missing <{tag}>"
        # mind. ein self-Link
        self_links = [
            l for l in root.findall(f"{ATOM_NS}link")
            if l.get("rel") == "self"
        ]
        assert len(self_links) == 1

    def test_entry_count_matches_input(self):
        with patch("app.main.get_all_assessments", return_value=_fake_assessments()):
            resp = client.get("/api/feed.xml")
        root = ET.fromstring(resp.content)
        entries = root.findall(f"{ATOM_NS}entry")
        assert len(entries) == 3

    def test_entries_sorted_by_updated_desc(self):
        with patch("app.main.get_all_assessments", return_value=_fake_assessments()):
            resp = client.get("/api/feed.xml")
        root = ET.fromstring(resp.content)
        updateds = [
            e.find(f"{ATOM_NS}updated").text
            for e in root.findall(f"{ATOM_NS}entry")
        ]
        # Strip Z-suffix fuer Vergleich
        bare = [u.rstrip("Z") for u in updateds]
        assert bare == sorted(bare, reverse=True), updateds

    def test_html_escaping_in_titles(self):
        """Anti-Terror-Paket & Überwachung — & muss als &amp; im XML stehen."""
        with patch("app.main.get_all_assessments", return_value=_fake_assessments()):
            resp = client.get("/api/feed.xml")
        # Roh-XML pruefen, nicht den geparsten Inhalt
        body = resp.text
        # Das Ampersand muss als &amp; codiert sein
        assert "Anti-Terror-Paket &amp;" in body or "Anti-Terror-Paket &amp;#" in body
        # Der Roh-String darf kein nacktes & vor Whitespace haben
        assert "Paket & Überw" not in body

    def test_partei_filter_narrows_results(self):
        with patch("app.main.get_all_assessments", return_value=_fake_assessments()):
            resp_all = client.get("/api/feed.xml")
            resp_cdu = client.get("/api/feed.xml?partei=CDU")
        all_count = len(ET.fromstring(resp_all.content).findall(f"{ATOM_NS}entry"))
        cdu_count = len(ET.fromstring(resp_cdu.content).findall(f"{ATOM_NS}entry"))
        assert cdu_count == 1
        assert cdu_count < all_count

    def test_bundesland_filter_passed_to_query(self):
        """Der bundesland-Parameter wird an get_all_assessments durchgereicht."""
        with patch("app.main.get_all_assessments", return_value=_fake_assessments()) as m:
            client.get("/api/feed.xml?bundesland=NRW")
        m.assert_called_once_with("NRW")

    def test_etag_header_set(self):
        with patch("app.main.get_all_assessments", return_value=_fake_assessments()):
            resp = client.get("/api/feed.xml")
        assert "etag" in {k.lower() for k in resp.headers}
        etag = resp.headers["etag"]
        assert etag.startswith('"') and etag.endswith('"')

    def test_etag_304_not_modified(self):
        with patch("app.main.get_all_assessments", return_value=_fake_assessments()):
            resp1 = client.get("/api/feed.xml")
            etag = resp1.headers["etag"]
            resp2 = client.get("/api/feed.xml", headers={"If-None-Match": etag})
        assert resp2.status_code == 304

    def test_limit_clamped_to_200(self):
        big_input = _fake_assessments() * 100  # 300 Eintraege
        with patch("app.main.get_all_assessments", return_value=big_input):
            resp = client.get("/api/feed.xml?limit=500")
        root = ET.fromstring(resp.content)
        entries = root.findall(f"{ATOM_NS}entry")
        assert len(entries) == 200

    def test_limit_clamped_to_min_1(self):
        with patch("app.main.get_all_assessments", return_value=_fake_assessments()):
            resp = client.get("/api/feed.xml?limit=0")
        root = ET.fromstring(resp.content)
        entries = root.findall(f"{ATOM_NS}entry")
        assert len(entries) >= 1

    def test_empty_db_returns_valid_feed(self):
        with patch("app.main.get_all_assessments", return_value=[]):
            resp = client.get("/api/feed.xml")
        assert resp.status_code == 200
        root = ET.fromstring(resp.content)
        # Pflicht-Elemente trotzdem da
        assert root.find(f"{ATOM_NS}id") is not None
        assert root.find(f"{ATOM_NS}title") is not None
        # Aber keine Entries
        assert root.findall(f"{ATOM_NS}entry") == []

    def test_cors_header_present(self):
        with patch("app.main.get_all_assessments", return_value=[]):
            resp = client.get("/api/feed.xml")
        assert resp.headers.get("access-control-allow-origin") == "*"

    def test_self_url_includes_filter_params(self):
        with patch("app.main.get_all_assessments", return_value=_fake_assessments()):
            resp = client.get("/api/feed.xml?bundesland=NRW&partei=GRÜNE")
        root = ET.fromstring(resp.content)
        self_link = [l for l in root.findall(f"{ATOM_NS}link") if l.get("rel") == "self"][0]
        href = self_link.get("href")
        assert "bundesland=NRW" in href
        # partei kann URL-codiert sein
        assert "partei=" in href