"""Konsistenz-Tests v2-Detail ↔ PDF (#176 generalisiert). Stichprobe aus assessments. Pro Drucksache vergleicht der Test die kritischen Felder: - gwoe_score - empfehlung - Matrix-Cells (rating + abgeleitetes Symbol + abgeleitete Klasse) - wahlprogramm_scores (pro Fraktion: WP-Score, PP-Score, Begründungen) Quelle für v2: app.main._row_to_detail(row). Quelle für PDF: app.models.Assessment.model_validate(row) + die get_rating_symbol/get_rating_class-Funktionen aus app.report. Wenn beide Renderer dieselben kritischen Werte aus row ziehen, gibt es keine v2/PDF-Drift mehr. """ from __future__ import annotations import sqlite3 import sys from pathlib import Path import pytest def _has_app() -> bool: try: from app import main, report # noqa return True except Exception: return False pytestmark = pytest.mark.skipif(not _has_app(), reason="app nicht importierbar") def _live_rows(limit: int = 20) -> list[dict]: """Hole bis zu N Assessment-Rows direkt aus der prod-DB. Der Test läuft nur wenn die Container-DB lokal mountbar ist — sonst skippen wir mit Hinweis (nicht failen). """ candidates = [ Path("/app/data/gwoe-antraege.db"), # innerhalb des Containers Path(__file__).resolve().parent.parent / "data" / "gwoe-antraege.db", ] db = next((p for p in candidates if p.exists()), None) if not db: pytest.skip(f"keine gwoe-antraege.db gefunden (versucht: {candidates})") conn = sqlite3.connect(str(db)) conn.row_factory = sqlite3.Row rows = conn.execute( "SELECT * FROM assessments WHERE gwoe_score IS NOT NULL " "ORDER BY datum DESC LIMIT ?", (limit,), ).fetchall() conn.close() out = [] for r in rows: d = dict(r) # JSON-Felder deserialisieren wie _row_to_detail es erwartet import json as _json for k in ("fraktionen", "gwoe_matrix", "gwoe_schwerpunkt", "wahlprogramm_scores", "verbesserungen", "staerken", "schwaechen", "themen", "antrag_kernpunkte"): if k in d and isinstance(d[k], str): try: d[k] = _json.loads(d[k]) except Exception: d[k] = None out.append(d) return out def _v2_score(row: dict): """Wert wie er im v2-Detail erscheint.""" return row.get("gwoe_score") def _pdf_score(row: dict): """Wert wie er im PDF erscheint.""" from app.models import Assessment a = Assessment.model_validate(row) return a.gwoe_score def _v2_empfehlung(row: dict): return row.get("empfehlung") def _pdf_empfehlung(row: dict): from app.models import Assessment a = Assessment.model_validate(row) return a.empfehlung.value def _v2_matrix_keys(row: dict) -> dict[str, tuple[int, str, str]]: """Dict {field: (rating, symbol, class)} wie es im v2-Template gerendert wird.""" out = {} for cell in (row.get("gwoe_matrix") or []): if not isinstance(cell, dict): continue field = cell.get("field") if not field: continue rating_raw = cell.get("rating", 0) try: rating = int(rating_raw) except (TypeError, ValueError): rating = 0 if rating < -5: rating = -5 if rating > 5: rating = 5 if rating >= 4: sym, cls = "++", "m-pp" elif rating >= 1: sym, cls = "+", "m-p" elif rating == 0: sym, cls = "○", "m-0" elif rating <= -4: sym, cls = "−−", "m-nn" else: sym, cls = "−", "m-n" out[field] = (rating, sym, cls) return out def _pdf_matrix_keys(row: dict) -> dict[str, tuple[int, str, str]]: from app.models import Assessment from app.report import get_rating_symbol, get_rating_class a = Assessment.model_validate(row) out = {} # Mapping: rating-pp/p/0/n/nn ↔ m-pp/p/0/n/nn cls_map = { "rating-pp": "m-pp", "rating-p": "m-p", "rating-0": "m-0", "rating-n": "m-n", "rating-nn": "m-nn", } for entry in a.gwoe_matrix: sym = get_rating_symbol(entry.rating) cls = cls_map.get(get_rating_class(entry.rating), "?") out[entry.field] = (entry.rating, sym, cls) return out # ─── Tests ────────────────────────────────────────────────────────────────── @pytest.fixture(scope="module") def rows(): return _live_rows(limit=30) def test_score_consistency(rows): if not rows: pytest.skip("DB leer") for r in rows: assert _v2_score(r) == _pdf_score(r), ( f"score-Drift bei {r.get('drucksache')}: " f"v2={_v2_score(r)} pdf={_pdf_score(r)}" ) def test_empfehlung_consistency(rows): if not rows: pytest.skip("DB leer") for r in rows: assert _v2_empfehlung(r) == _pdf_empfehlung(r), ( f"empfehlung-Drift bei {r.get('drucksache')}: " f"v2={_v2_empfehlung(r)!r} pdf={_pdf_empfehlung(r)!r}" ) def test_matrix_consistency(rows): if not rows: pytest.skip("DB leer") drift_count = 0 drift_examples: list[str] = [] for r in rows: v2 = _v2_matrix_keys(r) pdf = _pdf_matrix_keys(r) common_keys = set(v2.keys()) & set(pdf.keys()) for k in common_keys: if v2[k] != pdf[k]: drift_count += 1 drift_examples.append( f"{r.get('drucksache')} {k}: v2={v2[k]} pdf={pdf[k]}" ) if len(drift_examples) >= 10: break if len(drift_examples) >= 10: break assert drift_count == 0, ( f"{drift_count} Matrix-Drifts. Erste 10:\n " + "\n ".join(drift_examples) ) def test_matrix_field_set_consistency(rows): """v2 und PDF rendern für jede Drucksache dieselben Matrix-Felder (kein 'verschluckter' Cell durch unterschiedliche Schema-Lookups).""" if not rows: pytest.skip("DB leer") for r in rows: v2_keys = set(_v2_matrix_keys(r).keys()) pdf_keys = set(_pdf_matrix_keys(r).keys()) assert v2_keys == pdf_keys, ( f"Field-Set-Drift bei {r.get('drucksache')}: " f"v2={sorted(v2_keys - pdf_keys)} pdf={sorted(pdf_keys - v2_keys)}" ) def test_wahlprogramm_scores_consistency(rows): """Für jede Fraktion: WP-Score und PP-Score und Begründung gleich.""" if not rows: pytest.skip("DB leer") from app.models import Assessment drift_count = 0 drift_examples: list[str] = [] for r in rows: a = Assessment.model_validate(r) # v2-Sicht: aus row.wahlprogramm_scores roh v2_by_fr = { (wp.get("fraktion") or ""): wp for wp in (r.get("wahlprogramm_scores") or []) if isinstance(wp, dict) } for s in a.wahlprogramm_scores: v2 = v2_by_fr.get(s.fraktion or "") if not v2: continue v2_wp = float((v2.get("wahlprogramm") or {}).get("score") or 0) v2_pp = float((v2.get("parteiprogramm") or {}).get("score") or 0) pdf_wp = float(s.wahlprogramm.score or 0) pdf_pp = float(s.parteiprogramm.score or 0) if v2_wp != pdf_wp or v2_pp != pdf_pp: drift_count += 1 drift_examples.append( f"{r.get('drucksache')} {s.fraktion}: " f"v2_wp={v2_wp} pdf_wp={pdf_wp} v2_pp={v2_pp} pdf_pp={pdf_pp}" ) if len(drift_examples) >= 10: break assert drift_count == 0, ( f"{drift_count} WP/PP-Score-Drifts. Erste 10:\n " + "\n ".join(drift_examples) )