From db2fdda66b486b261a94efc0ea733d256532b126 Mon Sep 17 00:00:00 2001 From: Dotty Dotter Date: Wed, 6 May 2026 17:15:21 +0200 Subject: [PATCH] test(#191 Phase 10.4): 10 Tests fuer presse_generator Style-Switch MockBewerter zeichnet system_prompt + user_prompt + model auf, damit der Style-Switch isoliert testbar ist. Coverage: - TestStyleSwitch: 'pm'/'thread'/'invalid' nutzen den richtigen Prompt - TestPersist: style-Wert wird korrekt in presse_drafts gespeichert - TestIdempotenz: gleiche (ds, url, style) liefert Cache-Treffer; pm und thread fuer gleiches Paar liefern getrennte Drafts; force=True umgeht den Cache - TestThreadAutoSplit: Auto-Splitter aktiviert sich bei zu langen Threads ohne \\n\\n-Trenner; bereits gesplittete Threads bleiben 10 passed. Co-Authored-By: Claude Opus 4.7 (1M context) --- tests/test_presse_generator_style.py | 259 +++++++++++++++++++++++++++ 1 file changed, 259 insertions(+) create mode 100644 tests/test_presse_generator_style.py diff --git a/tests/test_presse_generator_style.py b/tests/test_presse_generator_style.py new file mode 100644 index 0000000..afe491a --- /dev/null +++ b/tests/test_presse_generator_style.py @@ -0,0 +1,259 @@ +"""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): + return asyncio.get_event_loop().run_until_complete(coro) + + +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="Para 1.\n\nPara 2.\n\nPara 3."): + self.titel = titel + self.body = 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 + b1 = MockBewerter(titel="PM", body="Para1.\n\nPara2.") + 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