a8f85bf3ee
Smoketests fuer alle 7 Stimmverhalten-Aggregat-Endpoints (stimm-index, heuchelei, empfehlungs-konsistenz, pro-wert, cross-bl, zeitreihe) plus zwei CSV-Tests (Header-Spalten + konsistente Datenzeilen-Spaltenzahl). Refs: ADR 0010 Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
328 lines
11 KiB
Python
328 lines
11 KiB
Python
"""Smoke-Tests für die neuen API-Endpoints.
|
||
|
||
Diese Tests prüfen nur ob die Endpoints antworten und das richtige
|
||
Format zurückgeben — keine echten LLM-Calls oder DB-Writes.
|
||
"""
|
||
import pytest
|
||
|
||
# Skip wenn die App nicht importierbar ist (missing dependencies lokal)
|
||
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")
|
||
|
||
|
||
class TestQueueStatus:
|
||
def test_returns_json(self):
|
||
resp = client.get("/api/queue/status")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert "pending" in data
|
||
assert "worker_running" in data
|
||
|
||
def test_pending_starts_at_zero(self):
|
||
data = client.get("/api/queue/status").json()
|
||
assert data["pending"] == 0
|
||
|
||
|
||
class TestAuthMe:
|
||
def test_unauthenticated_returns_false(self):
|
||
resp = client.get("/api/auth/me")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["authenticated"] is False
|
||
|
||
|
||
class TestAuthLoginUrl:
|
||
def test_returns_enabled_flag(self):
|
||
resp = client.get("/api/auth/login-url")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert "enabled" in data
|
||
|
||
|
||
class TestBundeslaender:
|
||
def test_returns_list(self):
|
||
resp = client.get("/api/bundeslaender")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert isinstance(data, list)
|
||
assert len(data) > 10 # mindestens 10 BLs
|
||
|
||
|
||
class TestProgramme:
|
||
def test_returns_list(self):
|
||
resp = client.get("/api/programme")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert isinstance(data, list)
|
||
assert len(data) > 50 # mindestens 50 Programme
|
||
|
||
def test_status_returns_indexed_count(self):
|
||
resp = client.get("/api/programme/status")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert "indexed" in data
|
||
assert "total" in data
|
||
|
||
|
||
class TestHealth:
|
||
def test_health(self):
|
||
resp = client.get("/health")
|
||
assert resp.status_code == 200
|
||
|
||
|
||
class TestVoteOrphansEndpoint:
|
||
"""GET /api/auswertungen/vote-orphans (öffentlich)."""
|
||
|
||
def test_returns_json_structure(self):
|
||
resp = client.get("/api/auswertungen/vote-orphans?limit=5")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert "count" in data
|
||
assert "items" in data
|
||
assert "by_bundesland" in data
|
||
assert isinstance(data["items"], list)
|
||
|
||
def test_filter_bundesland_param(self):
|
||
resp = client.get("/api/auswertungen/vote-orphans?bundesland=NRW&limit=3")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
# Wenn items vorhanden, alle aus NRW
|
||
for it in data["items"]:
|
||
assert it["bundesland"] == "NRW"
|
||
|
||
|
||
class TestVoteOrphansAutoRateAuth:
|
||
"""POST /api/auswertungen/vote-orphans/auto-rate erfordert Admin."""
|
||
|
||
def test_unauthenticated_rejected(self):
|
||
resp = client.post(
|
||
"/api/auswertungen/vote-orphans/auto-rate",
|
||
data={"limit": 5},
|
||
)
|
||
# Auth-Wall greift entweder direkt 401, 403 oder Redirect (307/302)
|
||
assert resp.status_code in (401, 403, 307, 302)
|
||
|
||
|
||
class TestBatchAnalyzeAuth:
|
||
"""POST /api/batch-analyze erfordert Admin."""
|
||
|
||
def test_unauthenticated_rejected(self):
|
||
resp = client.post(
|
||
"/api/batch-analyze",
|
||
data={"bundesland": "NRW", "limit": 5},
|
||
)
|
||
assert resp.status_code in (401, 403, 307, 302)
|
||
|
||
def test_all_bl_unauthenticated_also_rejected(self):
|
||
resp = client.post(
|
||
"/api/batch-analyze",
|
||
data={"bundesland": "ALL", "limit": 10},
|
||
)
|
||
assert resp.status_code in (401, 403, 307, 302)
|
||
|
||
|
||
class TestAktuelleThemenEndpoints:
|
||
"""GET /api/aktuelle-themen/* sind oeffentlich."""
|
||
|
||
def test_top_returns_buckets(self):
|
||
resp = client.get("/api/aktuelle-themen/top?days=7&top_k=3")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert "buckets" in data
|
||
assert "n_total_news" in data
|
||
assert "filter" in data
|
||
|
||
def test_top_with_single_date(self):
|
||
resp = client.get("/api/aktuelle-themen/top?date=2026-05-01")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["filter"]["single_date"] == "2026-05-01"
|
||
|
||
def test_top_with_only_relevant(self):
|
||
resp = client.get("/api/aktuelle-themen/top?only_relevant=true&top_k=5")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["filter"]["only_relevant"] is True
|
||
|
||
def test_zeitreihe(self):
|
||
resp = client.get("/api/aktuelle-themen/zeitreihe?days=14")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert "buckets" in data
|
||
assert "sources" in data
|
||
assert "series" in data
|
||
|
||
def test_top_antraege(self):
|
||
resp = client.get("/api/aktuelle-themen/top-antraege?min_gwoe_score=8.0")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert "antraege" in data
|
||
|
||
def test_cluster(self):
|
||
resp = client.get("/api/aktuelle-themen/cluster?days=7")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert "clusters" in data
|
||
|
||
def test_drafts_list(self):
|
||
resp = client.get("/api/aktuelle-themen/drafts?limit=5")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert "drafts" in data
|
||
|
||
def test_drafts_versions(self):
|
||
resp = client.get(
|
||
"/api/aktuelle-themen/drafts-versions"
|
||
"?drucksache=missing&news_url=https://example.com/x"
|
||
)
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert "versions" in data
|
||
assert isinstance(data["versions"], list)
|
||
|
||
|
||
class TestScoreHistogramEndpoint:
|
||
"""GET /api/auswertungen/score-histogram (öffentlich)."""
|
||
|
||
def test_returns_11_buckets(self):
|
||
resp = client.get("/api/auswertungen/score-histogram")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
# 11 Buckets: 0–1, 1–2, ..., 10–11
|
||
assert len(data["buckets"]) == 11
|
||
for b in data["buckets"]:
|
||
assert "score_min" in b
|
||
assert "score_max" in b
|
||
assert "count" in b
|
||
assert b["count"] >= 0
|
||
|
||
def test_total_matches_sum_of_buckets(self):
|
||
resp = client.get("/api/auswertungen/score-histogram")
|
||
data = resp.json()
|
||
total_sum = sum(b["count"] for b in data["buckets"])
|
||
assert data["total"] == total_sum
|
||
|
||
def test_filter_bundesland(self):
|
||
resp = client.get("/api/auswertungen/score-histogram?bundesland=NRW")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["filter"]["bundesland"] == "NRW"
|
||
|
||
def test_filter_wahlperiode(self):
|
||
resp = client.get("/api/auswertungen/score-histogram?wahlperiode=NRW-WP18")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["filter"]["wahlperiode"] == "NRW-WP18"
|
||
|
||
|
||
class TestAdminStandAuth:
|
||
"""/v2/admin/stand + /api/admin/stand erfordern Admin."""
|
||
|
||
def test_page_unauthenticated_rejected(self):
|
||
resp = client.get("/v2/admin/stand")
|
||
assert resp.status_code in (401, 403, 307, 302)
|
||
|
||
def test_api_unauthenticated_rejected(self):
|
||
resp = client.get("/api/admin/stand")
|
||
assert resp.status_code in (401, 403, 307, 302)
|
||
|
||
|
||
class TestStimmverhaltenAggregateEndpoints:
|
||
"""Die 7 Stimmverhalten-Aggregat-Endpoints liefern strukturierte JSONs (siehe ADR 0010)."""
|
||
|
||
def test_stimm_index_returns_fraktionen(self):
|
||
resp = client.get("/api/auswertungen/stimm-index")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert "fraktionen" in data
|
||
assert isinstance(data["fraktionen"], list)
|
||
assert "n_assessments_matched" in data
|
||
|
||
def test_stimm_index_min_n_filter(self):
|
||
resp = client.get("/api/auswertungen/stimm-index?min_n=1")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["filter"]["min_n"] == 1
|
||
|
||
def test_stimm_index_exclude_antragsteller_default(self):
|
||
"""Default `exclude_antragsteller=True`."""
|
||
resp = client.get("/api/auswertungen/stimm-index")
|
||
data = resp.json()
|
||
assert data["filter"]["exclude_antragsteller"] in (True, "true", 1)
|
||
|
||
def test_heuchelei_returns_fraktionen(self):
|
||
resp = client.get("/api/auswertungen/heuchelei")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert "fraktionen" in data
|
||
|
||
def test_empfehlungs_konsistenz(self):
|
||
resp = client.get("/api/auswertungen/empfehlungs-konsistenz")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert "fraktionen" in data or "items" in data
|
||
|
||
def test_stimm_index_pro_wert_5_werte(self):
|
||
resp = client.get("/api/auswertungen/stimm-index-pro-wert")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert "werte" in data
|
||
# 5 GWÖ-Werte: Würde, Solidarität, Nachhaltigkeit, Gerechtigkeit, Demokratie
|
||
assert len(data["werte"]) == 5
|
||
|
||
def test_stimm_index_cross_bl(self):
|
||
resp = client.get("/api/auswertungen/stimm-index-cross-bl")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert "fraktionen" in data
|
||
assert "bundeslaender" in data
|
||
assert "cells" in data
|
||
|
||
def test_stimm_index_zeitreihe(self):
|
||
resp = client.get("/api/auswertungen/stimm-index-zeitreihe")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
# Erwartete Felder: buckets (Zeit-Achse), fraktionen, series
|
||
assert "buckets" in data
|
||
assert "fraktionen" in data
|
||
assert "series" in data
|
||
|
||
|
||
class TestStimmverhaltenCsvExport:
|
||
"""CSV-Export Long-Format aus #168."""
|
||
|
||
def test_csv_returns_text(self):
|
||
resp = client.get("/api/auswertungen/stimmverhalten.csv")
|
||
assert resp.status_code == 200
|
||
ctype = resp.headers.get("content-type", "")
|
||
assert "text/csv" in ctype or "text/plain" in ctype
|
||
|
||
def test_csv_has_header_row(self):
|
||
resp = client.get("/api/auswertungen/stimmverhalten.csv")
|
||
body = resp.text
|
||
# Header-Zeile mit den erwarteten Long-Format-Spalten
|
||
first_line = body.split("\n", 1)[0]
|
||
# Mindestens drei Pflicht-Spalten erwartet — Partei statt Fraktion in CSV
|
||
for required in ("drucksache", "bundesland", "partei", "vote"):
|
||
assert required in first_line, f"Spalte {required!r} fehlt im Header: {first_line!r}"
|
||
|
||
def test_csv_data_rows_have_consistent_columns(self):
|
||
"""Datenzeilen haben gleiche Spalten-Anzahl wie der Header."""
|
||
resp = client.get("/api/auswertungen/stimmverhalten.csv")
|
||
body = resp.text.strip()
|
||
if not body:
|
||
pytest.skip("kein CSV-Body — DB enthält evtl. keine Vote-Matches")
|
||
lines = body.split("\n")
|
||
n_cols = lines[0].count(",") + 1
|
||
for line in lines[1:6]: # Samples
|
||
assert line.count(",") + 1 == n_cols, f"Spaltenzahl-Mismatch: {line!r}"
|