d552582a0c
- conftest: pymupdf-Alias-Loading robuster, fuer echte Render-Tests - test_v2_pdf_consistency: fehlende_programme deserialisieren - test_endpoints_smoke: Auth-Tests skippen wenn Keycloak nicht aktiv; queue/status-Schema auf workers_running aktualisiert - test_inline_styles_baseline: skippen wenn tools/-Dir fehlt (Container) - test_presse_generator_style: Mock-Body lang genug fuer kein Re-Generate; neuer event-loop pro Test (3.10+-Lifecycle) - test_bug_regressions: EMBEDDINGS_DB-Patch auch im analyzer_mod; raising=False bei fitz/pymupdf raus (zerstoerte Folge-Tests) - test_icons: macOS AppleDouble-Files (._*) ueberspringen - test_protokoll_parsers_nrw: raising=False raus (Test-Isolation)
242 lines
7.8 KiB
Python
242 lines
7.8 KiB
Python
"""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",
|
||
"fehlende_programme"):
|
||
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)
|
||
)
|