1e381d23ab
Stimm-Index pro Fraktion über Quartale. Linien-Chart pro Fraktion, Lücken bei Quartalen mit n<3 (Ja UND Nein). Macht sichtbar, ob sich die Gemeinwohl-Affinität einer Fraktion innerhalb der Wahlperiode verschiebt. - `_quarter_for(datum)` Helper: ISO-Datum → "YYYY-Qn". - `aggregate_stimm_index_zeitreihe()` analog zu pro_wert/pro_gruppe, aber nach Quartal-Bucket statt Achse. - `GET /api/auswertungen/stimm-index-zeitreihe?parteien=CDU,SPD,...` - 4. Sub-Section im Stimmverhalten-Tab: Multi-Linien-Chart mit Partei-Farben (CDU schwarz, SPD rot, GRÜNE grün, FDP gelb, AfD blau, LINKE pink, BSW lila, SSW navy, BVB-FW orange). Bei aktueller Sparse-Datenmenge (35 Assessments × 4 Quartale) ist der Chart heute meist leer — Infrastruktur ist ready, fuellt sich automatisch mit Issue #44 Batch-Bewertung. Tests: 10 neue (4 _quarter_for, 6 aggregate). Suite jetzt 1005 grün. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
623 lines
29 KiB
Python
623 lines
29 KiB
Python
"""Tests fuer app.auswertungen.aggregate_stimm_index/heuchelei/pro_wert/cross_bl.
|
||
|
||
Verifiziert die JOIN-Aggregations-Logik gegen eine in-memory SQLite-DB
|
||
mit kontrollierten Sample-Assessments und Sample-Vote-Results.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import sqlite3
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
from app.auswertungen import (
|
||
aggregate_empfehlungs_konsistenz,
|
||
aggregate_heuchelei,
|
||
aggregate_stimm_index,
|
||
aggregate_stimm_index_cross_bl,
|
||
aggregate_stimm_index_pro_gruppe,
|
||
aggregate_stimm_index_pro_wert,
|
||
aggregate_stimm_index_zeitreihe,
|
||
export_stimmverhalten_csv,
|
||
_gruppen_score_for_assessment,
|
||
_quarter_for,
|
||
_wert_score_for_assessment,
|
||
)
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# Fixture: assessments + plenum_vote_results
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
def _matrix(*ratings: tuple[str, int]) -> str:
|
||
"""Build a gwoe_matrix JSON-Array. Args: (field, rating) tuples."""
|
||
return json.dumps([
|
||
{"field": f, "label": f, "aspect": f[-1], "rating": r,
|
||
"symbol": "++" if r >= 4 else ("+" if r >= 1 else "○")}
|
||
for f, r in ratings
|
||
])
|
||
|
||
|
||
def _wp_scores(*items: tuple[str, int, bool]) -> str:
|
||
"""Build a wahlprogramm_scores JSON-Array. Args: (fraktion, score, is_antrag)."""
|
||
return json.dumps([
|
||
{
|
||
"fraktion": fr,
|
||
"ist_antragsteller": ist,
|
||
"wahlprogramm": {"score": sc, "begründung": "stub"},
|
||
}
|
||
for fr, sc, ist in items
|
||
])
|
||
|
||
|
||
@pytest.fixture
|
||
def sample_db(tmp_path: Path) -> Path:
|
||
"""Lege eine Mini-DB mit assessments + plenum_vote_results an, die typische
|
||
Faelle abdeckt:
|
||
|
||
- NRW Antrag mit hohem GWOe-Score, GRÜNE/SPD ja, CDU/AfD nein
|
||
- NRW Antrag mit niedrigem GWOe-Score, AfD ja, GRÜNE nein (Anti-Pattern)
|
||
- NRW Antrag mit hohem Score, CDU Antragsteller (sollte bei Default
|
||
excluded werden)
|
||
- MV Anträge fuer Cross-BL-Test
|
||
- Heuchelei-Sample: GRÜNE-Antrag mit GRÜNE-wahlprogramm_score=9 aber
|
||
GRÜNE stimmt NEIN (synthetisch konstruiert)
|
||
"""
|
||
db = tmp_path / "test_stimmverhalten.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("""
|
||
CREATE TABLE plenum_vote_results (
|
||
bundesland TEXT NOT NULL,
|
||
drucksache TEXT NOT NULL,
|
||
ergebnis TEXT NOT NULL,
|
||
einstimmig INTEGER NOT NULL DEFAULT 0,
|
||
fraktionen_ja TEXT NOT NULL DEFAULT '[]',
|
||
fraktionen_nein TEXT NOT NULL DEFAULT '[]',
|
||
fraktionen_enthaltung TEXT NOT NULL DEFAULT '[]',
|
||
quelle_protokoll TEXT NOT NULL,
|
||
quelle_url TEXT,
|
||
parsed_at TEXT NOT NULL DEFAULT (datetime('now')),
|
||
PRIMARY KEY (bundesland, drucksache, quelle_protokoll)
|
||
)
|
||
""")
|
||
|
||
# ─── Assessments ───
|
||
# Format: (ds, bl, datum, fraktionen, score, matrix, wp_scores, empfehlung)
|
||
now = datetime.utcnow().isoformat()
|
||
UNTER = "Uneingeschränkt unterstützen"
|
||
AENDR = "Unterstützen mit Änderungen"
|
||
UEBR = "Überarbeiten"
|
||
ABLEH = "Ablehnen"
|
||
assessments = [
|
||
# NRW WP18 — High-Score-Anträge mit GWÖ-Empfehlung positiv
|
||
("18/A", "NRW", "2024-01-15", '["GRÜNE"]', 8.5,
|
||
_matrix(("A1", 4), ("B2", 3), ("C3", 5), ("D4", 4), ("E5", 5)),
|
||
_wp_scores(("GRÜNE", 9, True), ("CDU", 3, False),
|
||
("SPD", 7, False), ("AfD", 1, False)), UNTER),
|
||
("18/B", "NRW", "2024-02-15", '["GRÜNE"]', 7.5,
|
||
_matrix(("A2", 3), ("C3", 4), ("D4", 3), ("E5", 4)),
|
||
_wp_scores(("GRÜNE", 8, True), ("CDU", 2, False),
|
||
("SPD", 6, False), ("AfD", 1, False)), AENDR),
|
||
("18/C", "NRW", "2024-03-15", '["GRÜNE"]', 9.0,
|
||
_matrix(("A3", 5), ("D4", 4), ("E5", 5)),
|
||
_wp_scores(("GRÜNE", 9, True), ("CDU", 3, False),
|
||
("SPD", 7, False), ("AfD", 1, False)), UNTER),
|
||
("18/D", "NRW", "2024-04-15", '["GRÜNE"]', 7.0,
|
||
_matrix(("B2", 3), ("C2", 4), ("D2", 3)),
|
||
_wp_scores(("GRÜNE", 8, True), ("CDU", 4, False),
|
||
("SPD", 7, False), ("AfD", 2, False)), AENDR),
|
||
("18/E", "NRW", "2024-05-15", '["GRÜNE"]', 8.0,
|
||
_matrix(("A1", 4), ("B1", 3), ("E5", 4)),
|
||
_wp_scores(("GRÜNE", 9, True), ("CDU", 3, False),
|
||
("SPD", 7, False), ("AfD", 1, False)), UNTER),
|
||
# AfD-Antrag mit niedrigem Score → empfehlung=ablehnen
|
||
("18/F", "NRW", "2024-06-15", '["AfD"]', 2.0,
|
||
_matrix(("A1", -3), ("B2", -2), ("E5", -4)),
|
||
_wp_scores(("AfD", 8, True), ("GRÜNE", 1, False),
|
||
("CDU", 4, False), ("SPD", 2, False)), ABLEH),
|
||
("18/G", "NRW", "2024-07-15", '["AfD"]', 1.5,
|
||
_matrix(("A1", -4), ("E5", -5)),
|
||
_wp_scores(("AfD", 9, True), ("GRÜNE", 1, False),
|
||
("CDU", 3, False), ("SPD", 1, False)), ABLEH),
|
||
("18/H", "NRW", "2024-08-15", '["CDU"]', 5.0,
|
||
_matrix(("D4", 1), ("D3", 0)),
|
||
_wp_scores(("CDU", 7, True), ("GRÜNE", 4, False),
|
||
("SPD", 5, False), ("AfD", 3, False)), UEBR),
|
||
("18/I", "NRW", "2024-09-15", '["SPD"]', 6.5,
|
||
_matrix(("B2", 3), ("D2", 2)),
|
||
_wp_scores(("SPD", 8, True), ("GRÜNE", 6, False),
|
||
("CDU", 4, False), ("AfD", 1, False)), AENDR),
|
||
("18/J", "NRW", "2024-10-15", '["SPD"]', 4.0,
|
||
_matrix(("D4", 1), ("E5", 0)),
|
||
_wp_scores(("SPD", 5, True), ("GRÜNE", 3, False),
|
||
("CDU", 5, False), ("AfD", 2, False)), UEBR),
|
||
# MV WP8 fuer Cross-BL
|
||
("8/A", "MV", "2024-04-01", '["GRÜNE"]', 7.0,
|
||
_matrix(("A1", 3), ("D4", 4)),
|
||
_wp_scores(("GRÜNE", 8, True), ("CDU", 4, False),
|
||
("SPD", 6, False), ("AfD", 2, False)), AENDR),
|
||
("8/B", "MV", "2024-05-01", '["GRÜNE"]', 8.0,
|
||
_matrix(("B2", 4), ("D4", 5)),
|
||
_wp_scores(("GRÜNE", 9, True), ("CDU", 3, False),
|
||
("SPD", 7, False), ("AfD", 1, False)), UNTER),
|
||
("8/C", "MV", "2024-06-01", '["AfD"]', 2.0,
|
||
_matrix(("A1", -3), ("E5", -4)),
|
||
_wp_scores(("AfD", 9, True), ("GRÜNE", 1, False),
|
||
("CDU", 3, False), ("SPD", 2, False)), ABLEH),
|
||
]
|
||
for ds, bl, dat, fr, sc, mat, wps, emp in assessments:
|
||
conn.execute(
|
||
"INSERT INTO assessments (drucksache, title, fraktionen, datum, "
|
||
"bundesland, gwoe_score, gwoe_matrix, wahlprogramm_scores, "
|
||
"empfehlung, source, model, created_at, updated_at) VALUES "
|
||
"(?, ?, ?, ?, ?, ?, ?, ?, ?, 'test', 'test', ?, ?)",
|
||
(ds, f"Test {ds}", fr, dat, bl, sc, mat, wps, emp, now, now),
|
||
)
|
||
|
||
# ─── Vote-Results ───
|
||
# Jeder NRW-Hoch-Antrag (A-E,I): GRÜNE+SPD ja, CDU+AfD nein
|
||
# AfD-Antrag F,G: AfD ja, GRÜNE+SPD+CDU nein
|
||
# CDU-Antrag H: CDU ja, GRÜNE+SPD+AfD nein
|
||
# SPD-Antrag J: SPD ja, GRÜNE+CDU+AfD nein
|
||
# MV-Hoch-Anträge: GRÜNE+SPD ja, CDU+AfD nein
|
||
# MV-AfD-Antrag: AfD ja, andere nein
|
||
votes = [
|
||
("NRW", "18/A", "angenommen", '["GRÜNE","SPD"]', '["CDU","AfD"]', "MMP18-A"),
|
||
("NRW", "18/B", "angenommen", '["GRÜNE","SPD"]', '["CDU","AfD"]', "MMP18-B"),
|
||
("NRW", "18/C", "angenommen", '["GRÜNE","SPD"]', '["CDU","AfD"]', "MMP18-C"),
|
||
("NRW", "18/D", "angenommen", '["GRÜNE","SPD"]', '["CDU","AfD"]', "MMP18-D"),
|
||
("NRW", "18/E", "angenommen", '["GRÜNE","SPD"]', '["CDU","AfD"]', "MMP18-E"),
|
||
("NRW", "18/F", "abgelehnt", '["AfD"]', '["GRÜNE","SPD","CDU"]', "MMP18-F"),
|
||
("NRW", "18/G", "abgelehnt", '["AfD"]', '["GRÜNE","SPD","CDU"]', "MMP18-G"),
|
||
("NRW", "18/H", "angenommen", '["CDU","AfD"]', '["GRÜNE","SPD"]', "MMP18-H"),
|
||
("NRW", "18/I", "angenommen", '["GRÜNE","SPD"]', '["CDU","AfD"]', "MMP18-I"),
|
||
("NRW", "18/J", "abgelehnt", '["SPD"]', '["GRÜNE","CDU","AfD"]', "MMP18-J"),
|
||
("MV", "8/A", "angenommen", '["GRÜNE","SPD"]', '["CDU","AfD"]', "PlPr8-A"),
|
||
("MV", "8/B", "angenommen", '["GRÜNE","SPD"]', '["CDU","AfD"]', "PlPr8-B"),
|
||
("MV", "8/C", "abgelehnt", '["AfD"]', '["GRÜNE","SPD","CDU"]', "PlPr8-C"),
|
||
]
|
||
for bl, ds, erg, ja, nein, prot in votes:
|
||
conn.execute(
|
||
"INSERT INTO plenum_vote_results "
|
||
"(bundesland, drucksache, ergebnis, fraktionen_ja, fraktionen_nein, "
|
||
" quelle_protokoll) VALUES (?, ?, ?, ?, ?, ?)",
|
||
(bl, ds, erg, ja, nein, prot),
|
||
)
|
||
|
||
conn.commit()
|
||
conn.close()
|
||
return db
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# aggregate_stimm_index
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
class TestAggregateStimmIndex:
|
||
def test_grüne_high_index(self, sample_db):
|
||
"""GRÜNE: stimmt JA bei hohen GWÖ-Anträgen, NEIN bei niedrigen → hoher Index."""
|
||
out = aggregate_stimm_index(db_path=sample_db, min_n=2)
|
||
gruene = next(f for f in out["fraktionen"] if f["partei"] == "GRÜNE")
|
||
# JA: 18/A,B,D,E,I + 8/A,B (NRW+MV high score). Antragsteller-exclude
|
||
# entfernt 18/A-E + 8/A,B fuer GRÜNE → bleibt 18/I als JA. NEIN: 18/H + 18/J + 18/F + 18/G + 8/C.
|
||
assert gruene["n_ja"] >= 1
|
||
assert gruene["n_nein"] >= 4
|
||
# Wenn beide N>0: stimm_index sollte positiv sein
|
||
if gruene["stimm_index"] is not None:
|
||
assert gruene["stimm_index"] > 0
|
||
|
||
def test_afd_low_index(self, sample_db):
|
||
"""AfD: stimmt NEIN bei hohen GWÖ-Anträgen, JA bei niedrigen → negativer Index."""
|
||
out = aggregate_stimm_index(db_path=sample_db, min_n=2)
|
||
afd = next(f for f in out["fraktionen"] if f["partei"] == "AfD")
|
||
# AfD ist Antragsteller bei 18/F,G,8/C → diese werden ausgeschlossen.
|
||
# Bleibt: NEIN bei 18/A-E, 18/I, 18/J, 8/A, 8/B. AfD JA bei 18/H (CDU-Antrag).
|
||
assert afd["n_nein"] >= 4
|
||
# Wenn JA und NEIN beide gefuellt: Index negativ
|
||
if afd["stimm_index"] is not None:
|
||
assert afd["stimm_index"] < 0
|
||
|
||
def test_exclude_antragsteller_default(self, sample_db):
|
||
"""Default schliesst eigene Antraege aus."""
|
||
with_excl = aggregate_stimm_index(
|
||
db_path=sample_db, exclude_antragsteller=True, min_n=2,
|
||
)
|
||
without_excl = aggregate_stimm_index(
|
||
db_path=sample_db, exclude_antragsteller=False, min_n=2,
|
||
)
|
||
# GRÜNE hat 7 eigene Anträge → ohne Exclude mehr n_ja
|
||
gr_with = next(f for f in with_excl["fraktionen"] if f["partei"] == "GRÜNE")
|
||
gr_without = next(f for f in without_excl["fraktionen"] if f["partei"] == "GRÜNE")
|
||
assert gr_without["n_ja"] > gr_with["n_ja"]
|
||
|
||
def test_filter_by_bundesland(self, sample_db):
|
||
nrw = aggregate_stimm_index(db_path=sample_db, filter_bl="NRW", min_n=2)
|
||
assert nrw["n_assessments_matched"] == 10
|
||
|
||
mv = aggregate_stimm_index(db_path=sample_db, filter_bl="MV", min_n=2)
|
||
assert mv["n_assessments_matched"] == 3
|
||
|
||
def test_min_n_threshold(self, sample_db):
|
||
"""Fraktion mit n_ja=2, min_n=5 → ausreichend=False."""
|
||
out = aggregate_stimm_index(db_path=sample_db, min_n=5)
|
||
for f in out["fraktionen"]:
|
||
if f["n_ja"] < 5 or f["n_nein"] < 5:
|
||
assert f["ausreichend"] is False
|
||
|
||
def test_empty_db(self, tmp_path):
|
||
out = aggregate_stimm_index(db_path=tmp_path / "missing.db")
|
||
assert out["n_assessments_matched"] == 0
|
||
assert out["fraktionen"] == []
|
||
|
||
def test_sorted_by_index_desc(self, sample_db):
|
||
"""Output sorted by stimm_index descending — None at end."""
|
||
out = aggregate_stimm_index(db_path=sample_db, min_n=1)
|
||
indices = [f["stimm_index"] for f in out["fraktionen"]]
|
||
# All not-None values should be in descending order
|
||
not_none = [v for v in indices if v is not None]
|
||
assert not_none == sorted(not_none, reverse=True)
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# aggregate_heuchelei
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
class TestAggregateHeuchelei:
|
||
def test_returns_structure(self, sample_db):
|
||
out = aggregate_heuchelei(db_path=sample_db, min_n=1)
|
||
assert "fraktionen" in out
|
||
assert "n_assessments_matched" in out
|
||
|
||
def test_heuchelei_quote_calculation(self, sample_db):
|
||
out = aggregate_heuchelei(
|
||
db_path=sample_db, score_threshold=7.0, min_n=1,
|
||
)
|
||
# SPD: wahlprogramm_score>=7 in 18/A-E (5x). SPD stimmt JA in allen
|
||
# → heuchelei_quote = 0/5 = 0.
|
||
spd = next(f for f in out["fraktionen"] if f["partei"] == "SPD")
|
||
assert spd["n_im_programm"] >= 5
|
||
assert spd["heuchelei_quote"] == 0 or spd["heuchelei_quote"] is None
|
||
|
||
def test_threshold_filter(self, sample_db):
|
||
"""Hoher Threshold reduziert n_im_programm."""
|
||
low = aggregate_heuchelei(db_path=sample_db, score_threshold=1.0, min_n=1)
|
||
high = aggregate_heuchelei(db_path=sample_db, score_threshold=9.5, min_n=1)
|
||
# Bei threshold=1 sind quasi alle Fraktionen drin, bei 9.5 sehr wenige
|
||
low_total = sum(f["n_im_programm"] for f in low["fraktionen"])
|
||
high_total = sum(f["n_im_programm"] for f in high["fraktionen"])
|
||
assert low_total > high_total
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# _wert_score_for_assessment helper
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
class TestWertScoreHelper:
|
||
def test_extracts_per_column(self):
|
||
matrix = [
|
||
{"field": "A1", "rating": 3},
|
||
{"field": "B1", "rating": 5},
|
||
{"field": "C2", "rating": 4},
|
||
]
|
||
result = _wert_score_for_assessment(matrix)
|
||
# Spalte 1 (Würde): A1=3 + B1=5 → Ø=4
|
||
assert result["1"] == 4.0
|
||
# Spalte 2 (Solidarität): C2=4 → Ø=4
|
||
assert result["2"] == 4.0
|
||
|
||
def test_empty_matrix(self):
|
||
assert _wert_score_for_assessment([]) == {}
|
||
|
||
def test_invalid_entries_skipped(self):
|
||
matrix = [
|
||
{"field": "A1", "rating": 3},
|
||
{"field": "", "rating": 5}, # empty field skipped
|
||
{"field": "X9", "rating": 2}, # invalid column suffix
|
||
"not a dict", # type error
|
||
]
|
||
result = _wert_score_for_assessment(matrix)
|
||
assert result == {"1": 3.0}
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# aggregate_stimm_index_pro_wert
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
class TestAggregateProWert:
|
||
def test_structure(self, sample_db):
|
||
out = aggregate_stimm_index_pro_wert(db_path=sample_db, min_n=1)
|
||
assert out["werte"] == [
|
||
"Menschenwürde", "Solidarität", "Ökologische Nachhaltigkeit",
|
||
"Soziale Gerechtigkeit", "Transparenz & Demokratie",
|
||
]
|
||
assert "cells" in out
|
||
for partei in out["fraktionen"]:
|
||
assert set(out["cells"][partei].keys()) == set(out["werte"])
|
||
|
||
def test_cell_format(self, sample_db):
|
||
out = aggregate_stimm_index_pro_wert(db_path=sample_db, min_n=1)
|
||
if not out["fraktionen"]:
|
||
pytest.skip("no parteien — DB empty")
|
||
first_partei = out["fraktionen"][0]
|
||
cell = out["cells"][first_partei]["Menschenwürde"]
|
||
assert "stimm_index" in cell
|
||
assert "n_ja" in cell
|
||
assert "n_nein" in cell
|
||
assert "ausreichend" in cell
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# aggregate_stimm_index_pro_gruppe (#166)
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
class TestGruppenScoreHelper:
|
||
def test_extracts_per_row(self):
|
||
matrix = [
|
||
{"field": "A1", "rating": 3},
|
||
{"field": "A3", "rating": 5},
|
||
{"field": "B2", "rating": 4},
|
||
]
|
||
result = _gruppen_score_for_assessment(matrix)
|
||
# Zeile A: A1=3 + A3=5 → Ø=4
|
||
assert result["A"] == 4.0
|
||
# Zeile B: B2=4 → Ø=4
|
||
assert result["B"] == 4.0
|
||
|
||
def test_empty_matrix(self):
|
||
assert _gruppen_score_for_assessment([]) == {}
|
||
|
||
def test_invalid_entries_skipped(self):
|
||
matrix = [
|
||
{"field": "A1", "rating": 3},
|
||
{"field": "X9", "rating": 2}, # invalid row prefix
|
||
{"field": "", "rating": 5},
|
||
]
|
||
result = _gruppen_score_for_assessment(matrix)
|
||
assert result == {"A": 3.0}
|
||
|
||
|
||
class TestAggregateProGruppe:
|
||
def test_structure(self, sample_db):
|
||
out = aggregate_stimm_index_pro_gruppe(db_path=sample_db, min_n=1)
|
||
# 5 Berührungsgruppen-Labels
|
||
assert len(out["gruppen"]) == 5
|
||
assert "cells" in out
|
||
|
||
def test_cell_format(self, sample_db):
|
||
out = aggregate_stimm_index_pro_gruppe(db_path=sample_db, min_n=1)
|
||
if not out["fraktionen"]:
|
||
pytest.skip("no parteien — DB empty")
|
||
first_partei = out["fraktionen"][0]
|
||
# Pick first gruppe
|
||
first_gruppe = out["gruppen"][0]
|
||
cell = out["cells"][first_partei][first_gruppe]
|
||
assert "stimm_index" in cell
|
||
assert "n_ja" in cell
|
||
assert "n_nein" in cell
|
||
assert "ausreichend" in cell
|
||
|
||
def test_independent_from_pro_wert(self, sample_db):
|
||
"""Pro-Gruppe-Aggregation soll andere Achse als Pro-Wert haben."""
|
||
wert_out = aggregate_stimm_index_pro_wert(db_path=sample_db, min_n=1)
|
||
gruppe_out = aggregate_stimm_index_pro_gruppe(db_path=sample_db, min_n=1)
|
||
# Werte und Gruppen-Labels müssen disjunkt sein
|
||
assert set(wert_out["werte"]).isdisjoint(set(gruppe_out["gruppen"]))
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# aggregate_stimm_index_zeitreihe (#168)
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
class TestQuarterFor:
|
||
def test_q1(self):
|
||
assert _quarter_for("2024-01-15") == "2024-Q1"
|
||
assert _quarter_for("2024-03-31") == "2024-Q1"
|
||
|
||
def test_q2(self):
|
||
assert _quarter_for("2024-04-01") == "2024-Q2"
|
||
assert _quarter_for("2024-06-30") == "2024-Q2"
|
||
|
||
def test_q3_q4(self):
|
||
assert _quarter_for("2024-07-15") == "2024-Q3"
|
||
assert _quarter_for("2024-12-31") == "2024-Q4"
|
||
|
||
def test_invalid(self):
|
||
assert _quarter_for("") is None
|
||
assert _quarter_for("garbage") is None
|
||
assert _quarter_for("2024-13-01") is None # invalid month
|
||
|
||
|
||
class TestAggregateZeitreihe:
|
||
def test_structure(self, sample_db):
|
||
out = aggregate_stimm_index_zeitreihe(db_path=sample_db, min_n_per_bucket=1)
|
||
assert "buckets" in out
|
||
assert "fraktionen" in out
|
||
assert "series" in out
|
||
assert "detail" in out
|
||
assert "n_assessments_matched" in out
|
||
|
||
def test_buckets_sorted(self, sample_db):
|
||
out = aggregate_stimm_index_zeitreihe(db_path=sample_db, min_n_per_bucket=1)
|
||
assert out["buckets"] == sorted(out["buckets"])
|
||
|
||
def test_series_alignment(self, sample_db):
|
||
"""Pro Fraktion: series-Liste muss exakt so lang sein wie buckets."""
|
||
out = aggregate_stimm_index_zeitreihe(db_path=sample_db, min_n_per_bucket=1)
|
||
for partei in out["fraktionen"]:
|
||
assert len(out["series"][partei]) == len(out["buckets"])
|
||
|
||
def test_min_n_per_bucket(self, sample_db):
|
||
"""Mit hohem min_n_per_bucket wird stimm_index meist None."""
|
||
out = aggregate_stimm_index_zeitreihe(
|
||
db_path=sample_db, min_n_per_bucket=100,
|
||
)
|
||
for partei in out["fraktionen"]:
|
||
for idx in out["series"][partei]:
|
||
assert idx is None
|
||
|
||
def test_filter_by_parteien(self, sample_db):
|
||
"""Wenn parteien-Filter gesetzt, nur diese in fraktionen."""
|
||
out = aggregate_stimm_index_zeitreihe(
|
||
db_path=sample_db, parteien=["AfD"], min_n_per_bucket=1,
|
||
)
|
||
assert out["fraktionen"] == ["AfD"]
|
||
|
||
def test_empty_db(self, tmp_path):
|
||
out = aggregate_stimm_index_zeitreihe(db_path=tmp_path / "missing.db")
|
||
assert out["buckets"] == []
|
||
assert out["fraktionen"] == []
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# aggregate_stimm_index_cross_bl
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
class TestAggregateCrossBl:
|
||
def test_structure(self, sample_db):
|
||
out = aggregate_stimm_index_cross_bl(db_path=sample_db, min_n=1)
|
||
assert "NRW" in out["bundeslaender"]
|
||
assert "MV" in out["bundeslaender"]
|
||
assert "fraktionen" in out
|
||
assert "fraktionen_alle" in out
|
||
|
||
def test_only_multi_bl_in_fraktionen(self, sample_db):
|
||
"""Fraktionen-Liste enthält nur Parteien mit ≥2 BL ausreichend-erfüllt."""
|
||
out = aggregate_stimm_index_cross_bl(db_path=sample_db, min_n=2)
|
||
# Bei min_n=2 sind die meisten Kombinationen nicht ausreichend.
|
||
# Test prueft nur Struktur: fraktionen ⊆ fraktionen_alle.
|
||
assert set(out["fraktionen"]).issubset(set(out["fraktionen_alle"]))
|
||
|
||
def test_cell_format(self, sample_db):
|
||
out = aggregate_stimm_index_cross_bl(db_path=sample_db, min_n=1)
|
||
if out["fraktionen_alle"]:
|
||
partei = out["fraktionen_alle"][0]
|
||
for bl in out["bundeslaender"]:
|
||
cell = out["cells"][partei][bl]
|
||
assert "stimm_index" in cell
|
||
assert "n_ja" in cell
|
||
assert "n_nein" in cell
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# aggregate_empfehlungs_konsistenz (#167)
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
class TestAggregateEmpfehlungsKonsistenz:
|
||
def test_structure(self, sample_db):
|
||
out = aggregate_empfehlungs_konsistenz(db_path=sample_db, min_n=1)
|
||
assert "fraktionen" in out
|
||
assert "n_assessments_matched" in out
|
||
assert out["n_assessments_matched"] >= 1
|
||
|
||
def test_afd_high_konsistenz_quote(self, sample_db):
|
||
"""AfD stimmt NEIN bei allen GWÖ-positiv-empfohlenen Anträgen → Quote ~ 1.0."""
|
||
out = aggregate_empfehlungs_konsistenz(db_path=sample_db, min_n=1)
|
||
afd = next((f for f in out["fraktionen"] if f["partei"] == "AfD"), None)
|
||
assert afd is not None
|
||
# AfD wurde bei 18/A,B,C,D,E,8/A,8/B als NEIN eingetragen (alle UNTER/AENDR)
|
||
# und 18/I (AENDR) auch NEIN. → quote sollte hoch sein
|
||
assert afd["n_empfohlen"] >= 5
|
||
if afd["konsistenz_quote"] is not None:
|
||
assert afd["konsistenz_quote"] > 0.5
|
||
|
||
def test_grüne_low_konsistenz_quote(self, sample_db):
|
||
"""GRÜNE stimmt JA bei eigenen GWÖ-positiv Anträgen → Quote sehr niedrig."""
|
||
out = aggregate_empfehlungs_konsistenz(db_path=sample_db, min_n=1)
|
||
gr = next((f for f in out["fraktionen"] if f["partei"] == "GRÜNE"), None)
|
||
assert gr is not None
|
||
if gr["konsistenz_quote"] is not None:
|
||
assert gr["konsistenz_quote"] < 0.3
|
||
|
||
def test_only_positive_empfehlungen_count(self, sample_db):
|
||
"""Anträge mit empfehlung=Ablehnen/Überarbeiten dürfen NICHT zählen."""
|
||
out = aggregate_empfehlungs_konsistenz(db_path=sample_db, min_n=1)
|
||
# 7 Anträge mit empfehlung positiv (UNTER+AENDR): 18/A,B,C,D,E,I + 8/A,8/B
|
||
# = 8 positive. NICHT mitgezählt: 18/F,G,H,J + 8/C
|
||
assert out["n_assessments_matched"] == 8
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# export_stimmverhalten_csv
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
class TestExportStimmverhaltenCsv:
|
||
def test_csv_header(self, sample_db):
|
||
csv_text = export_stimmverhalten_csv(db_path=sample_db)
|
||
first_line = csv_text.splitlines()[0]
|
||
assert "drucksache" in first_line
|
||
assert "partei" in first_line
|
||
assert "vote" in first_line
|
||
assert "empfehlung" in first_line
|
||
assert "ist_antragsteller" in first_line
|
||
|
||
def test_csv_has_rows(self, sample_db):
|
||
csv_text = export_stimmverhalten_csv(db_path=sample_db)
|
||
lines = csv_text.splitlines()
|
||
# Header + 13 Anträge × ~3 Voter pro Antrag (excl Antragsteller)
|
||
assert len(lines) > 20
|
||
|
||
def test_csv_excludes_antragsteller_by_default(self, sample_db):
|
||
csv_text = export_stimmverhalten_csv(db_path=sample_db)
|
||
# GRÜNE ist Antragsteller bei 18/A-E + 8/A,B → keine Zeilen
|
||
# mit ist_antragsteller=1 erlaubt im Default
|
||
for line in csv_text.splitlines()[1:]:
|
||
cols = line.split(",")
|
||
ist_antrag = cols[-1].strip()
|
||
assert ist_antrag == "0"
|
||
|
||
def test_csv_includes_antragsteller_when_disabled(self, sample_db):
|
||
csv_text = export_stimmverhalten_csv(
|
||
db_path=sample_db, exclude_antragsteller=False,
|
||
)
|
||
antragsteller_rows = [
|
||
line for line in csv_text.splitlines()[1:]
|
||
if line.strip().endswith(",1")
|
||
]
|
||
assert len(antragsteller_rows) > 0
|
||
|
||
def test_csv_filter_by_bundesland(self, sample_db):
|
||
csv_text = export_stimmverhalten_csv(db_path=sample_db, filter_bl="MV")
|
||
# nur MV-Drucksachen 8/A, 8/B, 8/C
|
||
for line in csv_text.splitlines()[1:]:
|
||
assert ",MV," in line
|