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)
290 lines
11 KiB
Python
290 lines
11 KiB
Python
"""Tests fuer presse_generator.generate_draft Style-Switch (#191 Phase 10.4).
|
|
|
|
Nutzt einen Mock-Bewerter, der bekannte LLM-Antworten liefert, damit
|
|
die Persist-Logik und der Style-Switch isoliert testbar sind.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import sqlite3
|
|
import sys
|
|
|
|
import pytest
|
|
|
|
|
|
# Same aiosqlite-Cache-Schutz wie test_database.py
|
|
_aio = sys.modules.get("aiosqlite")
|
|
if _aio is not None and not hasattr(_aio, "connect"):
|
|
del sys.modules["aiosqlite"]
|
|
import aiosqlite as _real_aiosqlite # noqa: E402, F401
|
|
import importlib as _importlib # noqa: E402
|
|
|
|
if "app.database" in sys.modules:
|
|
_db_mod = sys.modules["app.database"]
|
|
if not hasattr(getattr(_db_mod, "aiosqlite", None), "connect"):
|
|
del sys.modules["app.database"]
|
|
_importlib.import_module("app.database")
|
|
else:
|
|
_importlib.import_module("app.database")
|
|
|
|
|
|
def run(coro):
|
|
"""Robust gegen Python 3.10+ event-loop-Lifecycle.
|
|
|
|
asyncio.get_event_loop() wirft RuntimeError wenn ein vorheriger Test
|
|
den Loop bereits geschlossen hat (passiert bei der vollen Suite, nicht
|
|
bei isoliertem Lauf). Ein neuer Loop pro Test ist robuster.
|
|
"""
|
|
loop = asyncio.new_event_loop()
|
|
try:
|
|
asyncio.set_event_loop(loop)
|
|
return loop.run_until_complete(coro)
|
|
finally:
|
|
loop.close()
|
|
asyncio.set_event_loop(None)
|
|
|
|
|
|
def _long_pm_body(n_words: int = 320) -> str:
|
|
"""Generiert einen >= n_words Worte langen Mock-Body in 3 Absaetzen.
|
|
|
|
Wichtig: presse_generator.generate_draft triggert bei style='pm' und
|
|
body < 280 Worten einen Re-Generate-Call (siehe app/presse_generator.py).
|
|
PM-Tests muessen den Mock-Body deshalb von vornherein lang genug halten,
|
|
damit nur ein einziger Bewerter-Call erfolgt.
|
|
"""
|
|
para = "Mieter haben ein Recht auf sichere Energieversorgung. " * 8
|
|
paras = [para.strip()] * 3
|
|
body = "\n\n".join(paras)
|
|
while len(body.split()) < n_words:
|
|
body += " Wir fordern eine klare Regelung."
|
|
return body
|
|
|
|
|
|
class MockBewerter:
|
|
"""Mock fuer QwenBewerter.bewerte — gibt feste Response zurueck und
|
|
merkt sich, mit welchem system_prompt aufgerufen wurde."""
|
|
|
|
def __init__(self, titel="Mock-Titel", body=None):
|
|
self.titel = titel
|
|
self.body = body if body is not None else _long_pm_body()
|
|
self.calls = []
|
|
|
|
async def bewerte(self, req):
|
|
self.calls.append({
|
|
"system_prompt": req.system_prompt,
|
|
"user_prompt": req.user_prompt,
|
|
"model": req.model,
|
|
})
|
|
return {"titel": self.titel, "body": self.body}
|
|
|
|
|
|
@pytest.fixture()
|
|
def db_path(tmp_path, monkeypatch):
|
|
path = tmp_path / "test.db"
|
|
from app.config import settings
|
|
monkeypatch.setattr(settings, "db_path", str(path))
|
|
return str(path)
|
|
|
|
|
|
@pytest.fixture()
|
|
def setup_db(db_path):
|
|
"""DB initialisieren + ein Test-Assessment + eine News einfügen."""
|
|
from app import database
|
|
run(database.init_db())
|
|
conn = sqlite3.connect(db_path)
|
|
try:
|
|
conn.execute("""
|
|
INSERT INTO assessments
|
|
(drucksache, bundesland, datum, title, fraktionen, gwoe_score,
|
|
gwoe_begruendung, antrag_zusammenfassung, empfehlung,
|
|
created_at, updated_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, datetime('now'), datetime('now'))
|
|
""", ("18/9999", "NRW", "2026-04-01",
|
|
"Test-Antrag", "[]", 7.0,
|
|
"Test-Begründung", "Test-Zusammenfassung", "Unterstützen mit Änderungen"))
|
|
# news_articles-Schema variiert je nach Migration; minimal sicher:
|
|
# url + titel + summary + datum
|
|
cols = [r[1] for r in conn.execute("PRAGMA table_info(news_articles)").fetchall()]
|
|
col_set = set(cols)
|
|
# Versuche verschiedene Schema-Varianten
|
|
if "datum" in col_set and "summary" in col_set:
|
|
conn.execute("""
|
|
INSERT INTO news_articles (url, titel, source, datum, summary)
|
|
VALUES (?, ?, ?, ?, ?)
|
|
""", ("https://example.com/news/1", "Test-News",
|
|
"tagesschau", "2026-04-02", "Test-News-Summary"))
|
|
elif "published_at" in col_set:
|
|
conn.execute("""
|
|
INSERT INTO news_articles (url, titel, source, published_at, summary)
|
|
VALUES (?, ?, ?, ?, ?)
|
|
""", ("https://example.com/news/1", "Test-News",
|
|
"tagesschau", "2026-04-02", "Test-News-Summary"))
|
|
else:
|
|
conn.execute(
|
|
"INSERT INTO news_articles (url, titel, summary) VALUES (?, ?, ?)",
|
|
("https://example.com/news/1", "Test-News", "Test-News-Summary"),
|
|
)
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
return db_path
|
|
|
|
|
|
# ─── Style-Switch ───────────────────────────────────────────────────────────
|
|
|
|
class TestStyleSwitch:
|
|
"""generate_draft mit style='pm' vs. style='thread' nutzt korrekt
|
|
unterschiedliche SYSTEM_PROMPTs."""
|
|
|
|
def test_pm_uses_pm_prompt(self, setup_db):
|
|
from app.presse_generator import generate_draft, SYSTEM_PROMPT
|
|
bewerter = MockBewerter()
|
|
run(generate_draft(
|
|
drucksache="18/9999",
|
|
news_url="https://example.com/news/1",
|
|
bewerter=bewerter,
|
|
style="pm",
|
|
))
|
|
assert len(bewerter.calls) == 1
|
|
assert bewerter.calls[0]["system_prompt"] == SYSTEM_PROMPT
|
|
|
|
def test_thread_uses_thread_prompt(self, setup_db):
|
|
from app.presse_generator import generate_draft, SYSTEM_PROMPT_THREAD
|
|
bewerter = MockBewerter(titel="Thread-Titel", body="Post 1.\n\nPost 2.\n\nPost 3.")
|
|
run(generate_draft(
|
|
drucksache="18/9999",
|
|
news_url="https://example.com/news/1",
|
|
bewerter=bewerter,
|
|
style="thread",
|
|
))
|
|
assert bewerter.calls[0]["system_prompt"] == SYSTEM_PROMPT_THREAD
|
|
|
|
def test_invalid_style_raises(self, setup_db):
|
|
from app.presse_generator import generate_draft
|
|
with pytest.raises(ValueError, match="unbekannter style"):
|
|
run(generate_draft(
|
|
drucksache="18/9999",
|
|
news_url="https://example.com/news/1",
|
|
bewerter=MockBewerter(),
|
|
style="banana",
|
|
))
|
|
|
|
|
|
# ─── Persist + Idempotenz ─────────────────────────────────────────────────
|
|
|
|
class TestPersist:
|
|
def test_pm_persists_with_style_pm(self, setup_db):
|
|
from app.presse_generator import generate_draft
|
|
result = run(generate_draft(
|
|
drucksache="18/9999",
|
|
news_url="https://example.com/news/1",
|
|
bewerter=MockBewerter(),
|
|
style="pm",
|
|
))
|
|
assert result["style"] == "pm"
|
|
assert result["titel"] == "Mock-Titel"
|
|
assert result["_was_existing"] is False
|
|
|
|
def test_thread_persists_with_style_thread(self, setup_db):
|
|
from app.presse_generator import generate_draft
|
|
result = run(generate_draft(
|
|
drucksache="18/9999",
|
|
news_url="https://example.com/news/1",
|
|
bewerter=MockBewerter(body="P1.\n\nP2.\n\nP3."),
|
|
style="thread",
|
|
))
|
|
assert result["style"] == "thread"
|
|
|
|
|
|
class TestIdempotenz:
|
|
"""Idempotenz pro (drucksache, news_url, style)-Tripel."""
|
|
|
|
def test_same_pair_returns_existing(self, setup_db):
|
|
from app.presse_generator import generate_draft
|
|
b = MockBewerter()
|
|
r1 = run(generate_draft(
|
|
drucksache="18/9999", news_url="https://example.com/news/1",
|
|
bewerter=b, style="pm",
|
|
))
|
|
r2 = run(generate_draft(
|
|
drucksache="18/9999", news_url="https://example.com/news/1",
|
|
bewerter=b, style="pm",
|
|
))
|
|
# Zweiter Call: kein neuer LLM-Call
|
|
assert len(b.calls) == 1
|
|
assert r2["_was_existing"] is True
|
|
assert r1["id"] == r2["id"]
|
|
|
|
def test_different_styles_separate_drafts(self, setup_db):
|
|
"""gleiche (ds, url) aber pm + thread → zwei verschiedene Drafts."""
|
|
from app.presse_generator import generate_draft
|
|
# PM braucht >= 280 Worte sonst triggert der Re-Generate-Pfad
|
|
# einen zweiten Bewerter-Call (siehe presse_generator).
|
|
b1 = MockBewerter(titel="PM", body=_long_pm_body())
|
|
b2 = MockBewerter(titel="Thread", body="P1.\n\nP2.")
|
|
run(generate_draft(
|
|
drucksache="18/9999", news_url="https://example.com/news/1",
|
|
bewerter=b1, style="pm",
|
|
))
|
|
run(generate_draft(
|
|
drucksache="18/9999", news_url="https://example.com/news/1",
|
|
bewerter=b2, style="thread",
|
|
))
|
|
# Beide LLM-Calls fanden statt — kein Cache-Treffer
|
|
assert len(b1.calls) == 1
|
|
assert len(b2.calls) == 1
|
|
|
|
def test_force_creates_new(self, setup_db):
|
|
"""force=True ueberschreibt Idempotenz, neuer LLM-Call."""
|
|
from app.presse_generator import generate_draft
|
|
b = MockBewerter()
|
|
r1 = run(generate_draft(
|
|
drucksache="18/9999", news_url="https://example.com/news/1",
|
|
bewerter=b, style="pm",
|
|
))
|
|
r2 = run(generate_draft(
|
|
drucksache="18/9999", news_url="https://example.com/news/1",
|
|
bewerter=b, style="pm", force=True,
|
|
))
|
|
assert len(b.calls) == 2
|
|
assert r2["_was_existing"] is False
|
|
assert r1["id"] != r2["id"]
|
|
|
|
|
|
class TestThreadAutoSplit:
|
|
"""Wenn ein Thread-Body keine \\n\\n-Trenner hat, splittet der
|
|
Auto-Splitter in mehrere Posts."""
|
|
|
|
def test_long_unsplit_thread_gets_split(self, setup_db):
|
|
from app.presse_generator import generate_draft
|
|
# Body ohne \n\n, deutlich > 280 chars → Splitter aktiv.
|
|
# 6 Sätze à ~70-80 chars = ~480 chars total.
|
|
long_body = (
|
|
"Mieter haben ein Recht auf sichere Energieversorgung im Winter. "
|
|
"Der Antrag schützt sie vor vorsätzlichen Versorgungssperren. "
|
|
"Familien mit Kindern und ältere Menschen sind besonders betroffen. "
|
|
"Wenn Vermieter Geld zurückhalten drohen Wärme und Wassersperren. "
|
|
"Wir fordern eine klare Regelung im Strafrecht für solche Fälle. "
|
|
"#GWO #Wohnrecht"
|
|
)
|
|
assert len(long_body) > 290, f"Test-Setup-Bug: len(body)={len(long_body)}"
|
|
bewerter = MockBewerter(titel="X", body=long_body)
|
|
result = run(generate_draft(
|
|
drucksache="18/9999", news_url="https://example.com/news/1",
|
|
bewerter=bewerter, style="thread",
|
|
))
|
|
posts = [p for p in result["body"].split("\n\n") if p.strip()]
|
|
assert len(posts) >= 2 # mindestens auf 2 Posts gesplittet
|
|
|
|
def test_short_thread_not_split(self, setup_db):
|
|
from app.presse_generator import generate_draft
|
|
# Body bereits mit 3 Posts, kein Splitter-Trigger
|
|
body = "P1 hier.\n\nP2 da.\n\nP3 dort."
|
|
bewerter = MockBewerter(titel="X", body=body)
|
|
result = run(generate_draft(
|
|
drucksache="18/9999", news_url="https://example.com/news/1",
|
|
bewerter=bewerter, style="thread",
|
|
))
|
|
posts = [p for p in result["body"].split("\n\n") if p.strip()]
|
|
assert len(posts) == 3
|