Plugin benachrichtigung: E-Mail, Telegram, In-App per Adapter-Muster
- Adapter-Pattern mit drei Kanaelen: E-Mail (SMTP per Env, TLS starttls/ssl, Dev-Fallback: Protokoll), Telegram (Bot-API, Token per Env, Chat-ID pro Benutzer), In-App (persistente Nachrichten mit Unread-Counter und 'Alle als gelesen markieren') - Pro Benutzer Kanal-Praeferenzen (Einstellungsseite, mehrere Kanaele gleichzeitig) plus eigene Kontakt-Tabelle (E-Mail-Adresse, Chat-ID); Kern und users-Tabelle unveraendert - Oeffentliche Plugin-API: await send_notification(user, titel, text, kategorie) - andere Plugins holen das Plugin ueber app.state.registry - Eigene Migration (0001_tabellen), eigene Routen/Templates, deutsche UI - Tests: Adapter-Auswahl nach Praeferenz, In-App-Persistenz, Dev-Log- Adapter, SMTP-Versand (gemockt), Telegram-API-Aufruf, Einstellungsseite - README: Plugin-Doku + neue Env-Variablen; docker-compose: Platzhalter
This commit is contained in:
Flo Hartmann
439
tests/test_benachrichtigung.py
Normal file
439
tests/test_benachrichtigung.py
Normal file
@@ -0,0 +1,439 @@
|
||||
"""Tests: Plugin „benachrichtigung“ — Adapter-Auswahl, In-App-Persistenz,
|
||||
Dev-Log-Fallback, Einstellungsseite und öffentliche API."""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import smtplib
|
||||
|
||||
from sqlalchemy import select
|
||||
|
||||
from redaktionskern.auth.models import User
|
||||
from redaktionskern.config import Settings
|
||||
|
||||
from plugins.benachrichtigung import BenachrichtigungPlugin
|
||||
from plugins.benachrichtigung.adapter import DevLogAdapter, EmailSmtpAdapter, TelegramAdapter
|
||||
from plugins.benachrichtigung.models import Benachrichtigung, Kontakt, Praeferenz
|
||||
|
||||
from tests.conftest import melde_an
|
||||
|
||||
|
||||
def hole_plugin(app):
|
||||
plugin = app.state.registry.get("benachrichtigung")
|
||||
assert plugin is not None, "Plugin 'benachrichtigung' ist nicht geladen"
|
||||
return plugin
|
||||
|
||||
|
||||
def adapter_modul(plugin):
|
||||
"""Adapter-Modul der geladenen Plugin-Instanz.
|
||||
|
||||
Der Loader führt Plugins unter dem Namensraum
|
||||
`spiele_redaktion_plugins.…` aus; Klassen-Identität nur gegen dieses
|
||||
Modul prüfen, nicht gegen den Importpfad `plugins.…`.
|
||||
"""
|
||||
import sys
|
||||
|
||||
return sys.modules[f"{type(plugin).__module__}.adapter"]
|
||||
|
||||
|
||||
def hole_benutzer(app, username: str = "admin") -> User:
|
||||
with app.state.session_factory() as db:
|
||||
return db.scalar(select(User).where(User.username == username))
|
||||
|
||||
|
||||
def praeferez_setzen(app, user_id: int, kanaele: str) -> None:
|
||||
with app.state.session_factory() as db:
|
||||
praef = db.get(Praeferenz, user_id)
|
||||
if praef is None:
|
||||
praef = Praeferenz(user_id=user_id)
|
||||
db.add(praef)
|
||||
praef.kanaele = kanaele
|
||||
db.commit()
|
||||
|
||||
|
||||
def kontakt_setzen(app, user_id: int, **felder) -> None:
|
||||
with app.state.session_factory() as db:
|
||||
kontakt = db.get(Kontakt, user_id)
|
||||
if kontakt is None:
|
||||
kontakt = Kontakt(user_id=user_id)
|
||||
db.add(kontakt)
|
||||
for feld, wert in felder.items():
|
||||
setattr(kontakt, feld, wert)
|
||||
db.commit()
|
||||
|
||||
|
||||
class AufzeichnenderAdapter(DevLogAdapter):
|
||||
"""Test-Adapter: zeichnet Aufrufe auf, stellt nichts zu."""
|
||||
|
||||
aufrufe: list[tuple]
|
||||
|
||||
def __init__(self, schluessel: str) -> None:
|
||||
super().__init__(schluessel, schluessel)
|
||||
self.aufrufe = []
|
||||
|
||||
async def senden(self, user_id, empfaenger, titel, text, kategorie):
|
||||
self.aufrufe.append((self.schluessel, user_id, empfaenger, titel))
|
||||
|
||||
|
||||
# ---------------- Laden & Migration ----------------
|
||||
|
||||
def test_plugin_geladen_mit_migration_und_navigation(app):
|
||||
plugin = hole_plugin(app)
|
||||
assert plugin.name == "benachrichtigung"
|
||||
assert any(e.url == "/benachrichtigung" for e in plugin.navigation())
|
||||
|
||||
from sqlalchemy import text
|
||||
|
||||
with app.state.engine.connect() as conn:
|
||||
zeilen = conn.execute(
|
||||
text("SELECT version FROM schema_migrations WHERE plugin='benachrichtigung'")
|
||||
).fetchall()
|
||||
tabellen = {
|
||||
z[0]
|
||||
for z in conn.execute(text("SELECT name FROM sqlite_master WHERE type='table'"))
|
||||
}
|
||||
assert [z[0] for z in zeilen] == ["0001_tabellen"]
|
||||
assert {
|
||||
"benachrichtigung",
|
||||
"benachrichtigung_praeferenz",
|
||||
"benachrichtigung_kontakt",
|
||||
} <= tabellen
|
||||
|
||||
|
||||
def test_seiten_erfordern_anmeldung(client):
|
||||
for pfad in ("/benachrichtigung", "/benachrichtigung/einstellungen"):
|
||||
antwort = client.get(pfad, follow_redirects=False)
|
||||
assert antwort.status_code == 303
|
||||
assert antwort.headers["location"] == "/login"
|
||||
|
||||
|
||||
# ---------------- Öffentliche API & Adapter-Auswahl ----------------
|
||||
|
||||
def test_in_app_persistenz_als_standard(client):
|
||||
"""Ohne Präferenz wird In-App zugestellt (Standard-Kanal) und persistiert."""
|
||||
app = client.app
|
||||
plugin = hole_plugin(app)
|
||||
benutzer = hole_benutzer(app)
|
||||
|
||||
bericht = asyncio.run(
|
||||
plugin.send_notification(benutzer, "Doppelter Titel", "BGG-Treffer", "dedup")
|
||||
)
|
||||
assert bericht.zugestellt == ["inapp"]
|
||||
assert bericht.fehlgeschlagen == []
|
||||
|
||||
with app.state.session_factory() as db:
|
||||
nachrichten = db.scalars(select(Benachrichtigung)).all()
|
||||
assert len(nachrichten) == 1
|
||||
n = nachrichten[0]
|
||||
assert n.user_id == benutzer.id
|
||||
assert n.titel == "Doppelter Titel"
|
||||
assert n.text == "BGG-Treffer"
|
||||
assert n.kategorie == "dedup"
|
||||
assert n.gelesen is False
|
||||
|
||||
|
||||
def test_adapter_auswahl_nach_praeferenz(client):
|
||||
"""Nur die gewählten Kanäle werden angesprochen — auch mehrere gleichzeitig."""
|
||||
app = client.app
|
||||
plugin = hole_plugin(app)
|
||||
benutzer = hole_benutzer(app)
|
||||
|
||||
email_adapter = AufzeichnenderAdapter("email")
|
||||
telegram_adapter = AufzeichnenderAdapter("telegram")
|
||||
inapp_adapter = AufzeichnenderAdapter("inapp")
|
||||
plugin._adapters = {
|
||||
"email": email_adapter,
|
||||
"telegram": telegram_adapter,
|
||||
"inapp": inapp_adapter,
|
||||
}
|
||||
|
||||
# Nur E-Mail gewählt → kein In-App-Eintrag.
|
||||
praeferez_setzen(app, benutzer.id, "email")
|
||||
kontakt_setzen(app, benutzer.id, email="redakteur@example.org")
|
||||
bericht = asyncio.run(plugin.send_notification(benutzer, "T1", "Text", "test"))
|
||||
assert bericht.zugestellt == ["email"]
|
||||
assert email_adapter.aufrufe == [
|
||||
("email", benutzer.id, "redakteur@example.org", "T1")
|
||||
]
|
||||
assert inapp_adapter.aufrufe == []
|
||||
with app.state.session_factory() as db:
|
||||
assert db.scalars(select(Benachrichtigung)).all() == []
|
||||
|
||||
# Telegram + In-App gleichzeitig → beide bedient.
|
||||
praeferez_setzen(app, benutzer.id, "telegram,inapp")
|
||||
kontakt_setzen(app, benutzer.id, telegram_chat_id="12345")
|
||||
bericht = asyncio.run(plugin.send_notification(benutzer, "T2", "Text", "test"))
|
||||
assert sorted(bericht.zugestellt) == ["inapp", "telegram"]
|
||||
assert telegram_adapter.aufrufe == [("telegram", benutzer.id, "12345", "T2")]
|
||||
assert len(inapp_adapter.aufrufe) == 1
|
||||
|
||||
|
||||
def test_fehlende_empfaengeradresse_faellt_kanalweise_aus(client):
|
||||
"""E-Mail gewählt, aber keine Adresse → Kanal fällt aus, In-App bleibt."""
|
||||
app = client.app
|
||||
plugin = hole_plugin(app)
|
||||
benutzer = hole_benutzer(app)
|
||||
|
||||
praeferez_setzen(app, benutzer.id, "email,inapp")
|
||||
bericht = asyncio.run(plugin.send_notification(benutzer, "T", "X", "test"))
|
||||
assert bericht.zugestellt == ["inapp"]
|
||||
assert bericht.fehlgeschlagen == ["email"]
|
||||
|
||||
|
||||
def test_leere_kanal_auswahl_zustellt_nichts(client):
|
||||
app = client.app
|
||||
plugin = hole_plugin(app)
|
||||
benutzer = hole_benutzer(app)
|
||||
|
||||
praeferez_setzen(app, benutzer.id, "")
|
||||
bericht = asyncio.run(plugin.send_notification(benutzer, "T", "X", "test"))
|
||||
assert bericht.zugestellt == []
|
||||
with app.state.session_factory() as db:
|
||||
assert db.scalars(select(Benachrichtigung)).all() == []
|
||||
|
||||
|
||||
# ---------------- Dev-Log-Adapter ----------------
|
||||
|
||||
def test_dev_log_adapter_ohne_smtp_konfiguration(client, caplog):
|
||||
"""Ohne SPIELE_SMTP_HOST protokolliert der E-Mail-Kanal nur (dev-Fallback)."""
|
||||
app = client.app
|
||||
plugin = hole_plugin(app)
|
||||
benutzer = hole_benutzer(app)
|
||||
kontakt_setzen(app, benutzer.id, email="chef@example.org")
|
||||
praeferez_setzen(app, benutzer.id, "email")
|
||||
|
||||
DevLogAdapterGeladen = adapter_modul(plugin).DevLogAdapter
|
||||
assert isinstance(plugin.adapters["email"], DevLogAdapterGeladen)
|
||||
assert plugin.adapters["email"].dev_fallback is True
|
||||
|
||||
with caplog.at_level(logging.INFO, logger="redaktion.benachrichtigung"):
|
||||
bericht = asyncio.run(
|
||||
plugin.send_notification(
|
||||
benutzer, "Erinnerung", "Deadline naht", "erinnerung"
|
||||
)
|
||||
)
|
||||
assert bericht.zugestellt == ["email"]
|
||||
texte = [eintrag.getMessage() for eintrag in caplog.records]
|
||||
assert any("chef@example.org" in t and "Erinnerung" in t for t in texte)
|
||||
|
||||
|
||||
def test_smtp_adapter_wird_bei_konfiguration_gewaehlt(tmp_path, monkeypatch):
|
||||
"""Mit SPIELE_SMTP_HOST wird der echte SMTP-Adapter gebaut und versendet."""
|
||||
monkeypatch.setenv("SPIELE_SMTP_HOST", "mail.example.org")
|
||||
monkeypatch.setenv("SPIELE_SMTP_BENUTZER", "bot@example.org")
|
||||
monkeypatch.setenv("SPIELE_SMTP_PASSWORT", "geheim")
|
||||
|
||||
gesendet: list = []
|
||||
|
||||
class FakeSMTP:
|
||||
letzte_instanz: "FakeSMTP | None" = None
|
||||
|
||||
def __init__(self, host, port, timeout=None):
|
||||
self.host = host
|
||||
self.tls = False
|
||||
FakeSMTP.letzte_instanz = self
|
||||
|
||||
def starttls(self, context=None):
|
||||
self.tls = True
|
||||
|
||||
def login(self, user, passwort):
|
||||
self.login_user = user
|
||||
|
||||
def send_message(self, nachricht):
|
||||
gesendet.append(nachricht)
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, *args):
|
||||
return False
|
||||
|
||||
monkeypatch.setattr(smtplib, "SMTP", FakeSMTP)
|
||||
|
||||
settings = Settings(
|
||||
database_url=f"sqlite:///{tmp_path / 'smtp.db'}",
|
||||
session_secret="s",
|
||||
initial_admin_password="x",
|
||||
plugins_dir=None,
|
||||
)
|
||||
plugin = BenachrichtigungPlugin()
|
||||
plugin.on_load(_mini_kontext(settings))
|
||||
|
||||
assert isinstance(plugin.adapters["email"], EmailSmtpAdapter)
|
||||
assert plugin.adapters["telegram"].dev_fallback is True # ohne Token
|
||||
|
||||
with plugin.context.session_factory() as db:
|
||||
db.add(Praeferenz(user_id=1, kanaele="email"))
|
||||
db.add(Kontakt(user_id=1, email="ziel@example.org"))
|
||||
db.commit()
|
||||
|
||||
benutzer = User(id=1, username="x", display_name="x", password_hash="x")
|
||||
bericht = asyncio.run(plugin.send_notification(benutzer, "Betreff", "Inhaltstext"))
|
||||
|
||||
assert bericht.zugestellt == ["email"]
|
||||
assert len(gesendet) == 1
|
||||
assert gesendet[0]["To"] == "ziel@example.org"
|
||||
assert gesendet[0]["Subject"] == "[Spiele-Redaktion] Betreff"
|
||||
server = FakeSMTP.letzte_instanz
|
||||
assert server is not None
|
||||
assert server.host == "mail.example.org"
|
||||
assert server.tls is True
|
||||
assert server.login_user == "bot@example.org"
|
||||
|
||||
|
||||
def test_telegram_adapter_ruft_bot_api_mit_chat_id(client, monkeypatch):
|
||||
"""Telegram-Adapter übergibt Chat-ID und Nachricht an die Bot-API."""
|
||||
app = client.app
|
||||
plugin = hole_plugin(app)
|
||||
benutzer = hole_benutzer(app)
|
||||
|
||||
api_aufrufe: list[tuple[str, str]] = []
|
||||
adapter = TelegramAdapter("123:testtoken")
|
||||
|
||||
def fake_api(chat_id: str, nachricht: str) -> dict:
|
||||
api_aufrufe.append((chat_id, nachricht))
|
||||
return {"ok": True}
|
||||
|
||||
monkeypatch.setattr(adapter, "_api_senden", fake_api)
|
||||
plugin._adapters["telegram"] = adapter
|
||||
|
||||
praeferez_setzen(app, benutzer.id, "telegram")
|
||||
kontakt_setzen(app, benutzer.id, telegram_chat_id="-100999")
|
||||
bericht = asyncio.run(
|
||||
plugin.send_notification(benutzer, "Titel TG", "Text TG", "test")
|
||||
)
|
||||
|
||||
assert bericht.zugestellt == ["telegram"]
|
||||
assert api_aufrufe == [("-100999", "Titel TG\n\nText TG")]
|
||||
|
||||
|
||||
# ---------------- Posteingang: Unread-Counter & alle gelesen ----------------
|
||||
|
||||
def test_posteingang_unread_counter_und_alle_gelesen(client):
|
||||
app = client.app
|
||||
plugin = hole_plugin(app)
|
||||
benutzer = hole_benutzer(app)
|
||||
melde_an(client)
|
||||
|
||||
for i in range(3):
|
||||
asyncio.run(plugin.send_notification(benutzer, f"M{i}", "Inhalt", "test"))
|
||||
|
||||
seite = client.get("/benachrichtigung")
|
||||
assert seite.status_code == 200
|
||||
assert "3 ungelesen" in seite.text
|
||||
for i in range(3):
|
||||
assert f"M{i}" in seite.text
|
||||
|
||||
antwort = client.post("/benachrichtigung/alle-gelesen", follow_redirects=False)
|
||||
assert antwort.status_code == 303
|
||||
|
||||
with app.state.session_factory() as db:
|
||||
rest = db.scalars(
|
||||
select(Benachrichtigung).where(Benachrichtigung.gelesen.is_(False))
|
||||
).all()
|
||||
assert rest == []
|
||||
assert "0 ungelesen" in client.get("/benachrichtigung").text
|
||||
|
||||
|
||||
def test_einzelne_nachricht_als_gelesen_markieren(client):
|
||||
app = client.app
|
||||
plugin = hole_plugin(app)
|
||||
benutzer = hole_benutzer(app)
|
||||
melde_an(client)
|
||||
asyncio.run(plugin.send_notification(benutzer, "Nur eine", "Inhalt", "test"))
|
||||
|
||||
with app.state.session_factory() as db:
|
||||
id = db.scalar(select(Benachrichtigung.id).limit(1))
|
||||
antwort = client.post(f"/benachrichtigung/{id}/gelesen", follow_redirects=False)
|
||||
assert antwort.status_code == 303
|
||||
|
||||
with app.state.session_factory() as db:
|
||||
assert db.get(Benachrichtigung, id).gelesen is True
|
||||
|
||||
|
||||
# ---------------- Einstellungsseite ----------------
|
||||
|
||||
def test_einstellungen_speichern_und_laden(client):
|
||||
app = client.app
|
||||
benutzer = hole_benutzer(app)
|
||||
melde_an(client)
|
||||
|
||||
formular = client.get("/benachrichtigung/einstellungen")
|
||||
assert formular.status_code == 200
|
||||
assert 'name="kanaele" value="email"' in formular.text
|
||||
assert "Telegram-Chat-ID" in formular.text
|
||||
|
||||
antwort = client.post(
|
||||
"/benachrichtigung/einstellungen",
|
||||
data={
|
||||
"kanaele": ["email", "inapp"],
|
||||
"emailadresse": "redakteur@example.org",
|
||||
"telegram_chat_id": "",
|
||||
},
|
||||
follow_redirects=False,
|
||||
)
|
||||
assert antwort.status_code == 303
|
||||
assert antwort.headers["location"].startswith("/benachrichtigung/einstellungen")
|
||||
|
||||
with app.state.session_factory() as db:
|
||||
praef = db.get(Praeferenz, benutzer.id)
|
||||
kontakt = db.get(Kontakt, benutzer.id)
|
||||
assert praef.kanaele == "email,inapp"
|
||||
assert kontakt.email == "redakteur@example.org"
|
||||
|
||||
seite = client.get("/benachrichtigung/einstellungen?gespeichert=true")
|
||||
assert "Einstellungen gespeichert." in seite.text
|
||||
assert 'value="redakteur@example.org"' in seite.text
|
||||
|
||||
|
||||
def test_einstellungen_validierung(client):
|
||||
app = client.app
|
||||
benutzer = hole_benutzer(app)
|
||||
melde_an(client)
|
||||
|
||||
antwort = client.post(
|
||||
"/benachrichtigung/einstellungen",
|
||||
data={
|
||||
"kanaele": ["email"],
|
||||
"emailadresse": "kaputte adresse",
|
||||
"telegram_chat_id": "",
|
||||
},
|
||||
follow_redirects=True,
|
||||
)
|
||||
assert "gültig" in antwort.text
|
||||
|
||||
antwort2 = client.post(
|
||||
"/benachrichtigung/einstellungen",
|
||||
data={"kanaele": ["telegram"], "emailadresse": "", "telegram_chat_id": "abc"},
|
||||
follow_redirects=True,
|
||||
)
|
||||
assert "nur Ziffern" in antwort2.text
|
||||
|
||||
with app.state.session_factory() as db:
|
||||
assert db.get(Kontakt, benutzer.id) is None # bei Fehlern nichts gespeichert
|
||||
|
||||
|
||||
# ---------------- Helfer ----------------
|
||||
|
||||
def _mini_kontext(settings: Settings):
|
||||
"""Minimaler PluginContext mit eigener DB für Adapter-Tests ohne App."""
|
||||
from fastapi.templating import Jinja2Templates
|
||||
from jinja2 import Environment, select_autoescape
|
||||
|
||||
from redaktionskern.contracts import PluginContext
|
||||
from redaktionskern.db import make_engine, make_session_factory
|
||||
from redaktionskern.migrationen import run_migrations
|
||||
|
||||
engine = make_engine(settings.database_url)
|
||||
session_factory = make_session_factory(engine)
|
||||
with engine.begin() as conn:
|
||||
run_migrations(conn, "benachrichtigung", BenachrichtigungPlugin().migrations())
|
||||
templates = Jinja2Templates(
|
||||
env=Environment(autoescape=select_autoescape(default=True))
|
||||
)
|
||||
return PluginContext(
|
||||
engine=engine,
|
||||
session_factory=session_factory,
|
||||
settings=settings,
|
||||
templates=templates,
|
||||
)
|
||||
Reference in New Issue
Block a user