8136a1a10b
Datenlage auf dev: 7281 Plenum-Votes, 96 Bewertungen, nur 19 Matches. Stimmverhalten-Tab zeigt fast nichts, weil die meisten Vote-Drucksachen keine Bewertung haben. Issue #172 schliesst die Luecke. **Banner im Stimmverhalten-Tab:** - Zeigt Anzahl + Verteilung pro BL der "Vote-only"-Drucksachen - Nur sichtbar wenn count > 0 - Aktion: "Auto-Bewerten Top-N" mit Limit-Selector (5/10/20) **Endpoint `GET /api/auswertungen/vote-orphans`:** LEFT JOIN plenum_vote_results vs assessments, count + by_bundesland + Top-N items sortiert nach parsed_at desc. **Endpoint `POST /api/auswertungen/vote-orphans/auto-rate`:** Admin-only, rate-limited 3/min. Nimmt Top-N Orphans, lädt Antragstext per Adapter, enqueued einen Bewertungs-Job pro Drucksache. Defaults limit=10, max 50. Per-skipped-reason-Liste in der Response (Adapter fehlt, Empty-Text, Queue-full, etc.). **Tests:** 4 neue (`TestGetVoteOrphans`), Suite 1071 gruen. Helper `_enqueue_for_bl` aus dem Batch-Endpoint wird hier indirekt wiederverwendet (gleiche Job-Queue-Pipeline). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
396 lines
18 KiB
Python
396 lines
18 KiB
Python
"""Tests für app.wahlperioden und app.auswertungen.
|
||
|
||
Issue #58 + Roadmap #59 Phase C. Verifiziert die Aggregations-Logik
|
||
gegen eine in-memory SQLite-DB mit kontrollierten Sample-Assessments.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import sqlite3
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
from app.auswertungen import (
|
||
aggregate_matrix,
|
||
aggregate_zeitreihe,
|
||
export_long_format,
|
||
)
|
||
from app.wahlperioden import all_wahlperioden, wahlperiode_for
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# wahlperioden helper
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
class TestWahlperiodeFor:
|
||
def test_current_wp_for_recent_date(self):
|
||
assert wahlperiode_for("2026-03-18", "MV") == "MV-WP8"
|
||
|
||
def test_previous_wp_for_old_date(self):
|
||
# MV WP8 startete am 26.10.2021 — alles davor ist WP7
|
||
assert wahlperiode_for("2020-01-01", "MV") == "MV-WP7"
|
||
|
||
def test_unknown_bl_returns_none(self):
|
||
assert wahlperiode_for("2026-01-01", "XX") is None
|
||
|
||
def test_empty_datum_returns_current_wp(self):
|
||
# Wenn kein Datum bekannt → wir nehmen die aktuelle WP an,
|
||
# weil das die einzig sinnvolle Default-Annahme ist
|
||
assert wahlperiode_for("", "NRW") == "NRW-WP18"
|
||
|
||
def test_all_wahlperioden_lists_each_bl_twice(self):
|
||
out = all_wahlperioden()
|
||
# 16 Bundesländer + BUND × 2 WPs = 34 Einträge (#56 fügt BUND hinzu)
|
||
assert len(out) == 34
|
||
# Aktuelle und vorherige WP für NRW
|
||
assert "NRW-WP18" in out
|
||
assert "NRW-WP17" in out
|
||
# BUND ist auch dabei
|
||
assert "BUND-WP21" in out
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# Test-DB-Fixture
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
@pytest.fixture
|
||
def sample_db(tmp_path: Path) -> Path:
|
||
"""Lege eine Mini-Assessments-DB an, die typische Fälle abdeckt."""
|
||
db = tmp_path / "test_assessments.db"
|
||
conn = sqlite3.connect(str(db))
|
||
conn.execute("""
|
||
CREATE TABLE assessments (
|
||
drucksache TEXT PRIMARY KEY,
|
||
title TEXT,
|
||
fraktionen TEXT,
|
||
datum TEXT,
|
||
bundesland TEXT,
|
||
gwoe_score REAL,
|
||
link TEXT,
|
||
gwoe_begruendung TEXT,
|
||
gwoe_matrix TEXT,
|
||
gwoe_schwerpunkt TEXT,
|
||
wahlprogramm_scores TEXT,
|
||
verbesserungen TEXT,
|
||
staerken TEXT,
|
||
schwaechen TEXT,
|
||
empfehlung TEXT,
|
||
empfehlung_symbol TEXT,
|
||
verbesserungspotenzial TEXT,
|
||
themen TEXT,
|
||
antrag_zusammenfassung TEXT,
|
||
antrag_kernpunkte TEXT,
|
||
source TEXT,
|
||
model TEXT,
|
||
created_at TEXT,
|
||
updated_at TEXT
|
||
)
|
||
""")
|
||
samples = [
|
||
# NRW WP18 — drei Anträge, zwei Parteien
|
||
("18/100", "NRW", "2024-01-15", '["CDU"]', 7.0),
|
||
("18/101", "NRW", "2024-02-15", '["SPD"]', 8.0),
|
||
("18/102", "NRW", "2024-03-15", '["CDU"]', 5.0),
|
||
# MV WP8 — Koalitionsantrag (zwei Parteien zählen beide)
|
||
("8/200", "MV", "2024-04-01", '["SPD","LINKE"]', 6.0),
|
||
("8/201", "MV", "2025-01-10", '["AfD"]', 2.0),
|
||
# MV WP7 — historischer Antrag vor wahlperiode_start (2021-10-26)
|
||
("7/100", "MV", "2020-05-01", '["CDU"]', 4.0),
|
||
# BB — FREIE WÄHLER soll als BVB-FW kanonisiert werden
|
||
("8/2", "BB", "2024-10-17", '["FREIE WÄHLER"]', 6.5),
|
||
]
|
||
now = datetime.utcnow().isoformat()
|
||
for ds, bl, dat, fr, sc in samples:
|
||
conn.execute(
|
||
"INSERT INTO assessments (drucksache, title, fraktionen, datum, bundesland, "
|
||
"gwoe_score, source, model, created_at, updated_at) VALUES "
|
||
"(?, ?, ?, ?, ?, ?, 'test', 'test', ?, ?)",
|
||
(ds, f"Test {ds}", fr, dat, bl, sc, now, now),
|
||
)
|
||
conn.commit()
|
||
conn.close()
|
||
return db
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# aggregate_matrix
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
class TestAggregateMatrix:
|
||
def test_total_count(self, sample_db):
|
||
m = aggregate_matrix(db_path=sample_db)
|
||
assert m["total"] == 7
|
||
|
||
def test_bundeslaender_listed(self, sample_db):
|
||
m = aggregate_matrix(db_path=sample_db)
|
||
assert set(m["bundeslaender"]) == {"NRW", "MV", "BB"}
|
||
|
||
def test_nrw_cdu_average(self, sample_db):
|
||
# NRW-CDU: 7.0 + 5.0 → Avg 6.0, n=2
|
||
m = aggregate_matrix(db_path=sample_db)
|
||
cell = m["cells"]["NRW"]["CDU"]
|
||
assert cell["n"] == 2
|
||
assert cell["avg"] == 6.0
|
||
|
||
def test_koalition_counts_both_parties(self, sample_db):
|
||
# MV-SPD und MV-LINKE bekommen beide den Score 6.0 (n=1)
|
||
m = aggregate_matrix(db_path=sample_db)
|
||
assert m["cells"]["MV"]["SPD"]["n"] == 1
|
||
assert m["cells"]["MV"]["LINKE"]["n"] == 1
|
||
assert m["cells"]["MV"]["SPD"]["avg"] == 6.0
|
||
|
||
def test_filter_by_wahlperiode(self, sample_db):
|
||
# NRW-WP18-Filter → nur die 3 NRW-Anträge
|
||
m = aggregate_matrix(filter_wp="NRW-WP18", db_path=sample_db)
|
||
assert m["total"] == 3
|
||
assert set(m["bundeslaender"]) == {"NRW"}
|
||
|
||
def test_filter_excludes_old_wp(self, sample_db):
|
||
# MV-WP8 darf den 7/100-Antrag (datum=2020) NICHT enthalten
|
||
m = aggregate_matrix(filter_wp="MV-WP8", db_path=sample_db)
|
||
assert m["total"] == 2 # nur 8/200 und 8/201
|
||
# CDU darf NICHT vorkommen, weil der CDU-Antrag in WP7 war
|
||
assert "CDU" not in m["cells"].get("MV", {})
|
||
|
||
def test_bb_freie_waehler_normalized_to_bvb(self, sample_db):
|
||
# Die BB-FW-Drucksache muss als BVB-FW gezählt werden, NICHT als
|
||
# generisches FREIE WÄHLER — das ist der eigentliche Mehrwert
|
||
# des Parteinamen-Mappers (#55)
|
||
m = aggregate_matrix(db_path=sample_db)
|
||
bb_cells = m["cells"]["BB"]
|
||
assert "BVB-FW" in bb_cells
|
||
assert bb_cells["BVB-FW"]["n"] == 1
|
||
assert "FREIE WÄHLER" not in bb_cells
|
||
|
||
def test_empty_db_returns_empty_matrix(self, tmp_path):
|
||
m = aggregate_matrix(db_path=tmp_path / "missing.db")
|
||
assert m["total"] == 0
|
||
assert m["bundeslaender"] == []
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# aggregate_zeitreihe
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
class TestAggregateZeitreihe:
|
||
def test_mv_cdu_two_wps(self, sample_db):
|
||
# MV-CDU hat einen Eintrag in WP7 (4.0) und keinen in WP8
|
||
z = aggregate_zeitreihe("MV", "CDU", db_path=sample_db)
|
||
wps = {entry["wp"]: entry for entry in z["wahlperioden"]}
|
||
assert "MV-WP7" in wps
|
||
assert wps["MV-WP7"]["avg"] == 4.0
|
||
assert wps["MV-WP7"]["n"] == 1
|
||
|
||
def test_nrw_cdu_one_wp(self, sample_db):
|
||
z = aggregate_zeitreihe("NRW", "CDU", db_path=sample_db)
|
||
assert len(z["wahlperioden"]) == 1
|
||
assert z["wahlperioden"][0]["avg"] == 6.0
|
||
|
||
def test_unknown_combination_empty(self, sample_db):
|
||
z = aggregate_zeitreihe("NRW", "AfD", db_path=sample_db)
|
||
assert z["wahlperioden"] == []
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# export_long_format
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
class TestExportLongFormat:
|
||
def test_csv_has_header(self, sample_db):
|
||
csv_text = export_long_format(db_path=sample_db)
|
||
first_line = csv_text.splitlines()[0]
|
||
assert "drucksache" in first_line
|
||
assert "bundesland" in first_line
|
||
assert "wahlperiode" in first_line
|
||
assert "partei" in first_line
|
||
assert "gwoe_score" in first_line
|
||
|
||
def test_koalition_yields_two_rows(self, sample_db):
|
||
csv_text = export_long_format(db_path=sample_db)
|
||
lines = csv_text.splitlines()[1:] # ohne Header
|
||
# 8/200 ist Koalitionsantrag (SPD+LINKE) → 2 Zeilen
|
||
mv_8_200_lines = [l for l in lines if l.startswith("8/200,")]
|
||
assert len(mv_8_200_lines) == 2
|
||
|
||
def test_bb_fw_normalized_in_csv(self, sample_db):
|
||
csv_text = export_long_format(db_path=sample_db)
|
||
assert "BVB-FW" in csv_text
|
||
# Generic FREIE WÄHLER darf in der Zeile NICHT auftauchen
|
||
bb_lines = [l for l in csv_text.splitlines() if "BB" in l and "8/2," in l]
|
||
assert any("BVB-FW" in l for l in bb_lines)
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# Edge-Cases (#134 Coverage-Backfill)
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
class TestLoadAssessmentsRobustness:
|
||
"""_load_assessments toleriert kaputte JSON-Eintraege im fraktionen-Feld."""
|
||
|
||
def test_invalid_json_in_fraktionen_falls_back_to_empty(self, tmp_path):
|
||
from app.auswertungen import _load_assessments
|
||
db = tmp_path / "broken.db"
|
||
conn = sqlite3.connect(str(db))
|
||
conn.execute("""
|
||
CREATE TABLE assessments (
|
||
drucksache TEXT PRIMARY KEY, title TEXT,
|
||
fraktionen TEXT, datum TEXT, bundesland TEXT,
|
||
gwoe_score REAL, link TEXT, gwoe_begruendung TEXT,
|
||
gwoe_matrix TEXT, gwoe_schwerpunkt TEXT,
|
||
wahlprogramm_scores TEXT, verbesserungen TEXT,
|
||
staerken TEXT, schwaechen TEXT, empfehlung TEXT,
|
||
empfehlung_symbol TEXT, verbesserungspotenzial TEXT,
|
||
themen TEXT, antrag_zusammenfassung TEXT,
|
||
antrag_kernpunkte TEXT, source TEXT, model TEXT,
|
||
created_at TEXT, updated_at TEXT
|
||
)
|
||
""")
|
||
# fraktionen-Feld enthaelt kein gueltiges JSON
|
||
conn.execute(
|
||
"INSERT INTO assessments (drucksache, bundesland, datum, fraktionen, gwoe_score) "
|
||
"VALUES (?, ?, ?, ?, ?)",
|
||
("18/777", "NRW", "2024-01-01", "{not json", 5.0),
|
||
)
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
rows = _load_assessments(db)
|
||
assert len(rows) == 1
|
||
assert rows[0]["fraktionen"] == [] # Fallback
|
||
|
||
|
||
class TestAggregateMatrixSkipsBlanks:
|
||
def test_skips_assessments_without_bundesland(self, tmp_path):
|
||
"""Anträge ohne bundesland werden ignoriert (continue-Branch line 115)."""
|
||
from app.auswertungen import aggregate_matrix
|
||
db = tmp_path / "blanks.db"
|
||
conn = sqlite3.connect(str(db))
|
||
conn.execute("""
|
||
CREATE TABLE assessments (
|
||
drucksache TEXT PRIMARY KEY, title TEXT,
|
||
fraktionen TEXT, datum TEXT, bundesland TEXT,
|
||
gwoe_score REAL, link TEXT, gwoe_begruendung TEXT,
|
||
gwoe_matrix TEXT, gwoe_schwerpunkt TEXT,
|
||
wahlprogramm_scores TEXT, verbesserungen TEXT,
|
||
staerken TEXT, schwaechen TEXT, empfehlung TEXT,
|
||
empfehlung_symbol TEXT, verbesserungspotenzial TEXT,
|
||
themen TEXT, antrag_zusammenfassung TEXT,
|
||
antrag_kernpunkte TEXT, source TEXT, model TEXT,
|
||
created_at TEXT, updated_at TEXT
|
||
)
|
||
""")
|
||
conn.execute(
|
||
"INSERT INTO assessments (drucksache, bundesland, datum, fraktionen, gwoe_score) "
|
||
"VALUES (?, ?, ?, ?, ?)",
|
||
("X/1", None, "2024-01-01", '["CDU"]', 7.0), # bundesland NULL
|
||
)
|
||
conn.execute(
|
||
"INSERT INTO assessments (drucksache, bundesland, datum, fraktionen, gwoe_score) "
|
||
"VALUES (?, ?, ?, ?, ?)",
|
||
("18/1", "NRW", "2024-01-01", '["CDU"]', 7.0),
|
||
)
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
m = aggregate_matrix(db_path=db)
|
||
assert m["total"] == 1 # nur der NRW-Eintrag
|
||
assert m["bundeslaender"] == ["NRW"]
|
||
|
||
|
||
class TestGetWahlperioden:
|
||
def test_returns_sorted_list(self, sample_db):
|
||
from app.auswertungen import get_wahlperioden
|
||
wps = get_wahlperioden(db_path=sample_db)
|
||
assert wps == sorted(wps)
|
||
# Sample-DB enthaelt NRW-WP18, MV-WP8, MV-WP7 sowie BB-WP8
|
||
assert any("NRW" in w for w in wps)
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# get_vote_orphans (#172)
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
class TestGetVoteOrphans:
|
||
@pytest.fixture
|
||
def orphan_db(self, tmp_path):
|
||
db = tmp_path / "orphans.db"
|
||
conn = sqlite3.connect(str(db))
|
||
conn.execute("""
|
||
CREATE TABLE assessments (
|
||
drucksache TEXT PRIMARY KEY, title TEXT, fraktionen TEXT,
|
||
datum TEXT, bundesland TEXT, gwoe_score REAL,
|
||
source TEXT, model TEXT, created_at TEXT, updated_at TEXT
|
||
)
|
||
""")
|
||
conn.execute("""
|
||
CREATE TABLE plenum_vote_results (
|
||
bundesland TEXT NOT NULL, drucksache TEXT NOT NULL,
|
||
ergebnis TEXT, einstimmig INTEGER DEFAULT 0,
|
||
fraktionen_ja TEXT DEFAULT '[]',
|
||
fraktionen_nein TEXT DEFAULT '[]',
|
||
fraktionen_enthaltung TEXT DEFAULT '[]',
|
||
quelle_protokoll TEXT NOT NULL,
|
||
quelle_url TEXT,
|
||
parsed_at TEXT NOT NULL DEFAULT (datetime('now')),
|
||
PRIMARY KEY (bundesland, drucksache, quelle_protokoll)
|
||
)
|
||
""")
|
||
# 3 Votes: 2 davon mit Bewertung, 1 orphan
|
||
conn.execute("INSERT INTO assessments VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
||
("18/1", "T", '["CDU"]', "2024-01-01", "NRW", 7.0,
|
||
"test", "qwen", "now", "now"))
|
||
conn.execute(
|
||
"INSERT INTO plenum_vote_results "
|
||
"(bundesland, drucksache, ergebnis, quelle_protokoll, parsed_at) "
|
||
"VALUES (?, ?, ?, ?, ?)",
|
||
("NRW", "18/1", "angenommen", "MMP18-1", "2024-01-02"),
|
||
)
|
||
conn.execute(
|
||
"INSERT INTO plenum_vote_results "
|
||
"(bundesland, drucksache, ergebnis, quelle_protokoll, parsed_at) "
|
||
"VALUES (?, ?, ?, ?, ?)",
|
||
("NRW", "18/2", "abgelehnt", "MMP18-2", "2024-01-05"),
|
||
)
|
||
conn.execute(
|
||
"INSERT INTO plenum_vote_results "
|
||
"(bundesland, drucksache, ergebnis, quelle_protokoll, parsed_at) "
|
||
"VALUES (?, ?, ?, ?, ?)",
|
||
("BB", "8/3", "angenommen", "BB8-1", "2024-01-04"),
|
||
)
|
||
conn.commit()
|
||
conn.close()
|
||
return db
|
||
|
||
def test_count_excludes_rated(self, orphan_db):
|
||
from app.auswertungen import get_vote_orphans
|
||
result = get_vote_orphans(db_path=orphan_db)
|
||
# 18/1 hat assessment → nicht orphan. 18/2 (NRW) + 8/3 (BB) sind orphans.
|
||
assert result["count"] == 2
|
||
assert result["by_bundesland"] == {"NRW": 1, "BB": 1}
|
||
|
||
def test_filter_bl(self, orphan_db):
|
||
from app.auswertungen import get_vote_orphans
|
||
nrw = get_vote_orphans(filter_bl="NRW", db_path=orphan_db)
|
||
assert nrw["count"] == 1
|
||
assert nrw["items"][0]["drucksache"] == "18/2"
|
||
|
||
def test_sort_by_latest_parsed_desc(self, orphan_db):
|
||
from app.auswertungen import get_vote_orphans
|
||
result = get_vote_orphans(db_path=orphan_db)
|
||
# 18/2 ist 2024-01-05, 8/3 ist 2024-01-04 → 18/2 zuerst
|
||
assert result["items"][0]["drucksache"] == "18/2"
|
||
|
||
def test_empty_db(self, tmp_path):
|
||
from app.auswertungen import get_vote_orphans
|
||
result = get_vote_orphans(db_path=tmp_path / "missing.db")
|
||
assert result == {"count": 0, "items": [], "by_bundesland": {}}
|