Plugin erinnerung: Redaktionsschluss pro Ausgabe, 4-Wochen-Erinnerung
- Ausgaben (Name + Redaktionsschluss-Datum) verwalten, mehrere parallel, Anlegen/Bearbeiten/Löschen nur für Admin, mit Audit-Log-Einträgen - Erinnerung 28 Tage vor dem Redaktionsschluss an alle aktiven Rezensenten mit offener (oder fehlender) Planung, Versand über das benachrichtigung- Plugin je Kanal-Präferenz, Audit als System/benachrichtigt - Täglicher APScheduler-Job (SPIELE_ERINNERUNG_JOB_UHRZEIT, abschaltbar) plus manuelle Sofort-Prüfung in der Admin-Ansicht - Doppelschutz: Protokolltabelle mit Unique (Ausgabe, Benutzer) - Admin-UI: Ausgabenliste mit Restzeit-Badges, Erinnerungsstatus (wer wurde wann über welche Kanäle erinnert), deutsche Templates - Tests mit eingefrorener Uhr (freezegun): Fenster-Grenzen, kein Versand vor Frist/nach Schluss, Empfänger-/Rollenregeln, keine Duplikate, Scheduler-Konfiguration; Integrationstest mit den echten Plugins (167 Tests grün)
This commit is contained in:
@@ -22,6 +22,7 @@ def _hintergrundjobs_deaktiviert(monkeypatch):
|
||||
monkeypatch.setenv("SPIELE_BGG_SYNC_AKTIV", "1").
|
||||
"""
|
||||
monkeypatch.setenv("SPIELE_BGG_SYNC_AKTIV", "0")
|
||||
monkeypatch.setenv("SPIELE_ERINNERUNG_JOB_AKTIV", "0")
|
||||
# Auch der BGG-Hilfsclient des dedup-Plugins bleibt in Tests offline.
|
||||
monkeypatch.setenv("SPIELE_DEDUP_BGG_AKTIV", "0")
|
||||
|
||||
|
||||
550
tests/test_erinnerung.py
Normal file
550
tests/test_erinnerung.py
Normal file
@@ -0,0 +1,550 @@
|
||||
"""Tests: Plugin „erinnerung“ — Fristen, Empfänger, Doppelschutz, Rollen.
|
||||
|
||||
Die Uhr wird mit freezegun eingefroren; alle datumsabhängigen Prüfungen
|
||||
laufen damit deterministisch. benachrichtigung und audit-log werden durch
|
||||
aufzeichnende Stubs ersetzt (wie in test_planung.py); ein Test beweist
|
||||
zusätzlich die Integration mit den echten Plugins.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import date
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
from freezegun import freeze_time
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy import select, text
|
||||
|
||||
from redaktionskern.app import create_app
|
||||
from redaktionskern.auth.models import User
|
||||
from tests.conftest import lege_benutzer_an, melde_an
|
||||
|
||||
|
||||
# ---------------- Hilfen ----------------
|
||||
|
||||
def plugin_modul(app, name: str):
|
||||
plugin_instanz = app.state.registry.get(name)
|
||||
return sys.modules[type(plugin_instanz).__module__]
|
||||
|
||||
|
||||
def ausgabe_anlegen(app, name: str, redaktionsschluss: date) -> int:
|
||||
Ausgabe = plugin_modul(app, "erinnerung").Ausgabe
|
||||
with app.state.session_factory() as db:
|
||||
ausgabe = Ausgabe(name=name, redaktionsschluss=redaktionsschluss)
|
||||
db.add(ausgabe)
|
||||
db.commit()
|
||||
return ausgabe.id
|
||||
|
||||
|
||||
def protokoll_zeilen(app) -> set[tuple[int, int]]:
|
||||
ErinnerungsProtokoll = plugin_modul(app, "erinnerung").ErinnerungsProtokoll
|
||||
with app.state.session_factory() as db:
|
||||
return {
|
||||
(z.ausgabe_id, z.user_id)
|
||||
for z in db.scalars(select(ErinnerungsProtokoll)).all()
|
||||
}
|
||||
|
||||
|
||||
def planung_anlegen(app, titel: str, rezensent_id: int, status: str = "offen") -> int:
|
||||
Planungseintrag = plugin_modul(app, "planung").Planungseintrag
|
||||
with app.state.session_factory() as db:
|
||||
eintrag = Planungseintrag(
|
||||
titel=titel, rezensent_id=rezensent_id, status=status, ausgabe=""
|
||||
)
|
||||
db.add(eintrag)
|
||||
db.commit()
|
||||
return eintrag.id
|
||||
|
||||
|
||||
def benutzer_objekt(app, username: str):
|
||||
with app.state.session_factory() as db:
|
||||
return db.scalar(select(User).where(User.username == username))
|
||||
|
||||
|
||||
def erinnerungs_plugin(app):
|
||||
plugin = app.state.registry.get("erinnerung")
|
||||
plugin._merke_registry(app.state.registry) # Partner sauber auflösen
|
||||
return plugin
|
||||
|
||||
|
||||
def sichtbar(antwort) -> str:
|
||||
import html
|
||||
|
||||
return html.unescape(antwort.text)
|
||||
|
||||
|
||||
# ---------------- Stubs (gemockte Abhängigkeiten) ----------------
|
||||
|
||||
@dataclass
|
||||
class Aufgezeichnet:
|
||||
daten: dict = field(default_factory=dict)
|
||||
|
||||
|
||||
class StubBenachrichtigung:
|
||||
name = "benachrichtigung"
|
||||
|
||||
def __init__(self):
|
||||
self.gesendet: list[Aufgezeichnet] = []
|
||||
|
||||
async def send_notification(self, user, titel, text, kategorie="allgemein"):
|
||||
self.gesendet.append(Aufgezeichnet({
|
||||
"user_id": user.id, "titel": titel, "text": text, "kategorie": kategorie,
|
||||
}))
|
||||
return SimpleNamespace(zugestellt=["inapp"], fehlgeschlagen=[])
|
||||
|
||||
|
||||
class StubAudit:
|
||||
name = "audit-log"
|
||||
|
||||
def __init__(self):
|
||||
self.eintraege: list[Aufgezeichnet] = []
|
||||
|
||||
async def log(self, actor, action, objekt_typ="", objekt_id=None,
|
||||
details=None, **kwargs):
|
||||
self.eintraege.append(Aufgezeichnet({
|
||||
"actor": actor, "action": action, "objekt_typ": objekt_typ,
|
||||
"objekt_id": objekt_id, "details": details,
|
||||
}))
|
||||
|
||||
def log_sync(self, *args, **kwargs):
|
||||
import asyncio
|
||||
|
||||
asyncio.run(self.log(*args, **kwargs))
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def stubs(app):
|
||||
"""Ersetzt benachrichtigung + audit-log im Registry-Zustand der App."""
|
||||
benachrichtigung = StubBenachrichtigung()
|
||||
audit = StubAudit()
|
||||
registry_inhalt = app.state.registry._plugins
|
||||
registry_inhalt["benachrichtigung"] = benachrichtigung
|
||||
registry_inhalt["audit-log"] = audit
|
||||
erinnerungs_plugin(app) # Registry-Referenz auffrischen
|
||||
return SimpleNamespace(benachrichtigung=benachrichtigung, audit=audit)
|
||||
|
||||
|
||||
# ---------------- Migration & Zugriff ----------------
|
||||
|
||||
def test_migration_legt_beide_tabellen_an(app):
|
||||
from sqlalchemy import inspect
|
||||
|
||||
inspektion = inspect(app.state.engine)
|
||||
ausgaben_spalten = {c["name"] for c in inspektion.get_columns("erinnerung_ausgabe")}
|
||||
protokoll_spalten = {c["name"] for c in inspektion.get_columns("erinnerung_protokoll")}
|
||||
assert {"id", "name", "redaktionsschluss", "erstellt_am", "aktualisiert_am"} <= ausgaben_spalten
|
||||
assert {"id", "ausgabe_id", "user_id", "kanaele", "erstellt_am"} <= protokoll_spalten
|
||||
with app.state.engine.connect() as conn:
|
||||
versionen = [
|
||||
z[0] for z in conn.execute(
|
||||
text("SELECT version FROM schema_migrations WHERE plugin='erinnerung'")
|
||||
).fetchall()
|
||||
]
|
||||
assert versionen == ["0001_ausgaben_und_protokoll"]
|
||||
|
||||
|
||||
def test_anonym_wird_zum_login_umgeleitet(client):
|
||||
antwort = client.get("/erinnerung", follow_redirects=False)
|
||||
assert antwort.status_code == 303
|
||||
assert antwort.headers["location"].startswith("/login")
|
||||
|
||||
|
||||
# ---------------- Rollen & UI ----------------
|
||||
|
||||
def test_rezensent_sieht_seite_ohne_verwaltung(app, client):
|
||||
lege_benutzer_an(app, "rez1", "rezensent")
|
||||
melde_an(client, "rez1", "test-12345678")
|
||||
antwort = client.get("/erinnerung")
|
||||
assert antwort.status_code == 200
|
||||
inhalt = sichtbar(antwort)
|
||||
assert "Redaktionsschluss" in inhalt
|
||||
assert "Neue Ausgabe anlegen" not in inhalt # keine Admin-Formulare
|
||||
assert "Jetzt prüfen und erinnern" not in inhalt
|
||||
|
||||
|
||||
def test_admin_sieht_verwaltung_und_status(client):
|
||||
melde_an(client)
|
||||
inhalt = sichtbar(client.get("/erinnerung"))
|
||||
assert "Neue Ausgabe anlegen" in inhalt
|
||||
assert "Erinnerungsstatus" in inhalt
|
||||
|
||||
|
||||
def test_nur_admin_darf_ausgaben_verwalten(app, client):
|
||||
"""Anlegen/Bearbeiten/Löschen/Prüfen: Admin ja, Redakteur und Rezensent 403."""
|
||||
for rolle in ("redakteur", "rezensent"):
|
||||
lege_benutzer_an(app, f"user-{rolle}", rolle)
|
||||
melde_an(client, f"user-{rolle}", "test-12345678")
|
||||
for methode, url, daten in (
|
||||
("post", "/erinnerung/ausgaben", {"name": "X", "redaktionsschluss": "2030-01-01"}),
|
||||
("get", "/erinnerung/ausgaben/1/bearbeiten", None),
|
||||
("post", "/erinnerung/ausgaben/1/bearbeiten",
|
||||
{"name": "Y", "redaktionsschluss": "2030-01-01"}),
|
||||
("post", "/erinnerung/ausgaben/1/loeschen", {}),
|
||||
("post", "/erinnerung/pruefen", {}),
|
||||
):
|
||||
if methode == "get":
|
||||
antwort = client.get(url)
|
||||
else:
|
||||
antwort = client.post(url, data=daten)
|
||||
assert antwort.status_code == 403, f"{rolle} auf {url} muss 403 geben"
|
||||
client.post("/logout")
|
||||
|
||||
|
||||
# ---------------- Ausgaben-CRUD (Admin) ----------------
|
||||
|
||||
def test_admin_legt_ausgabe_an_mit_audit(client, app, stubs):
|
||||
melde_an(client)
|
||||
antwort = client.post(
|
||||
"/erinnerung/ausgaben",
|
||||
data={"name": "3/2025", "redaktionsschluss": "2030-06-15"},
|
||||
follow_redirects=True,
|
||||
)
|
||||
assert antwort.status_code == 200
|
||||
assert "3/2025" in sichtbar(antwort)
|
||||
|
||||
Ausgabe = plugin_modul(app, "erinnerung").Ausgabe
|
||||
with app.state.session_factory() as db:
|
||||
ausgabe = db.scalar(select(Ausgabe))
|
||||
assert ausgabe.name == "3/2025"
|
||||
assert ausgabe.redaktionsschluss == date(2030, 6, 15)
|
||||
|
||||
assert [e.daten["action"] for e in stubs.audit.eintraege] == ["erstellt"]
|
||||
eintrag = stubs.audit.eintraege[0].daten
|
||||
assert eintrag["objekt_typ"] == "ausgabe"
|
||||
assert eintrag["details"]["name"] == "3/2025"
|
||||
|
||||
|
||||
def test_mehrere_ausgaben_parallel(app, client):
|
||||
melde_an(client)
|
||||
for name, schluss in (("3/2025", "2030-06-15"), ("4/2025", "2030-09-01")):
|
||||
client.post("/erinnerung/ausgaben", data={"name": name, "redaktionsschluss": schluss})
|
||||
inhalt = sichtbar(client.get("/erinnerung"))
|
||||
assert "3/2025" in inhalt and "4/2025" in inhalt
|
||||
|
||||
|
||||
def test_ausgabe_ohne_name_wird_abgewiesen(client, app):
|
||||
melde_an(client)
|
||||
antwort = client.post(
|
||||
"/erinnerung/ausgaben",
|
||||
data={"name": "", "redaktionsschluss": "2030-06-15"},
|
||||
follow_redirects=True,
|
||||
)
|
||||
assert "Bitte einen Ausgaben-Namen angeben" in antwort.text
|
||||
|
||||
|
||||
def test_ausgabe_mit_kaputtem_datum_wird_abgewiesen(client, app):
|
||||
melde_an(client)
|
||||
antwort = client.post(
|
||||
"/erinnerung/ausgaben",
|
||||
data={"name": "X", "redaktionsschluss": "15.06.2030"},
|
||||
follow_redirects=True,
|
||||
)
|
||||
assert "Ungültiges Datum" in antwort.text
|
||||
|
||||
|
||||
def test_admin_bearbeitet_ausgabe_mit_audit(client, app, stubs):
|
||||
melde_an(client)
|
||||
ausgabe_id = ausgabe_anlegen(app, "Alt", date(2030, 6, 15))
|
||||
antwort = client.post(
|
||||
f"/erinnerung/ausgaben/{ausgabe_id}/bearbeiten",
|
||||
data={"name": "Neu", "redaktionsschluss": "2030-07-01"},
|
||||
follow_redirects=True,
|
||||
)
|
||||
assert "gespeichert" in antwort.text
|
||||
Ausgabe = plugin_modul(app, "erinnerung").Ausgabe
|
||||
with app.state.session_factory() as db:
|
||||
ausgabe = db.get(Ausgabe, ausgabe_id)
|
||||
assert ausgabe.name == "Neu"
|
||||
assert ausgabe.redaktionsschluss == date(2030, 7, 1)
|
||||
aktionen = [e.daten["action"] for e in stubs.audit.eintraege]
|
||||
assert aktionen == ["geaendert"] # Anlegen geschah direkt per Helper
|
||||
geaendert = stubs.audit.eintraege[-1].daten
|
||||
assert geaendert["details"]["alt"]["name"] == "Alt"
|
||||
assert geaendert["details"]["neu"]["name"] == "Neu"
|
||||
|
||||
|
||||
def test_admin_loescht_ausgabe_mit_audit(client, app, stubs):
|
||||
melde_an(client)
|
||||
ausgabe_id = ausgabe_anlegen(app, "Weg damit", date(2030, 6, 15))
|
||||
antwort = client.post(f"/erinnerung/ausgaben/{ausgabe_id}/loeschen", follow_redirects=True)
|
||||
assert "gelöscht" in antwort.text
|
||||
Ausgabe = plugin_modul(app, "erinnerung").Ausgabe
|
||||
with app.state.session_factory() as db:
|
||||
assert db.get(Ausgabe, ausgabe_id) is None
|
||||
assert [e.daten["action"] for e in stubs.audit.eintraege] == ["geloescht"]
|
||||
|
||||
|
||||
# ---------------- Erinnerungslogik mit eingefrorener Uhr ----------------
|
||||
|
||||
def _drei_beteiligte(app):
|
||||
"""Rezensent mit offenem Eintrag, Rezensent ohne Einträge; Admin existiert."""
|
||||
lege_benutzer_an(app, "rez-offen", "rezensent")
|
||||
lege_benutzer_an(app, "rez-leer", "rezensent")
|
||||
rez_offen = benutzer_objekt(app, "rez-offen")
|
||||
planung_anlegen(app, "Offenes Spiel", rez_offen.id, status="offen")
|
||||
return rez_offen, benutzer_objekt(app, "rez-leer")
|
||||
|
||||
|
||||
@freeze_time("2030-05-18") # genau 28 Tage vor dem 15.06.2030
|
||||
def test_erinnerung_genau_beim_fensterstart(client, app, stubs):
|
||||
rez_offen, rez_leer = _drei_beteiligte(app)
|
||||
ausgabe_id = ausgabe_anlegen(app, "3/2025", date(2030, 6, 15))
|
||||
melde_an(client)
|
||||
|
||||
bericht = erinnerungs_plugin(app).pruefe_und_erinnere_sync()
|
||||
|
||||
assert bericht.gepruefte_ausgaben == 1
|
||||
assert {v["benutzer"] for v in bericht.versendet} == {"rez-offen", "rez-leer"}
|
||||
assert all(v["tage_rest"] == 28 for v in bericht.versendet)
|
||||
assert protokoll_zeilen(app) == {(ausgabe_id, rez_offen.id), (ausgabe_id, rez_leer.id)}
|
||||
# Benachrichtigungen über das Partner-Plugin, Kategorie „erinnerung“:
|
||||
assert {g.daten["user_id"] for g in stubs.benachrichtigung.gesendet} == {
|
||||
rez_offen.id, rez_leer.id,
|
||||
}
|
||||
assert all(g.daten["kategorie"] == "erinnerung" for g in stubs.benachrichtigung.gesendet)
|
||||
assert "Redaktionsschluss: 3/2025" == stubs.benachrichtigung.gesendet[0].daten["titel"]
|
||||
# Audit: jede Erinnerung als System-Ereignis protokolliert:
|
||||
aktionen = [e.daten["action"] for e in stubs.audit.eintraege]
|
||||
assert sorted(aktionen) == ["benachrichtigt", "benachrichtigt"]
|
||||
assert all(e.daten["actor"] is None for e in stubs.audit.eintraege)
|
||||
|
||||
|
||||
@freeze_time("2030-05-17") # 29 Tage vorher — Fenster noch zu
|
||||
def test_kein_versand_vor_der_vier_wochen_frist(client, app, stubs):
|
||||
_drei_beteiligte(app)
|
||||
ausgabe_anlegen(app, "3/2025", date(2030, 6, 15))
|
||||
melde_an(client)
|
||||
|
||||
bericht = erinnerungs_plugin(app).pruefe_und_erinnere_sync()
|
||||
|
||||
assert bericht.gepruefte_ausgaben == 0
|
||||
assert bericht.versendet == []
|
||||
assert stubs.benachrichtigung.gesendet == []
|
||||
assert protokoll_zeilen(app) == set()
|
||||
assert stubs.audit.eintraege == []
|
||||
|
||||
|
||||
@freeze_time("2030-06-16") # einen Tag nach dem Redaktionsschluss
|
||||
def test_kein_versand_nach_dem_redaktionsschluss(client, app, stubs):
|
||||
_drei_beteiligte(app)
|
||||
ausgabe_anlegen(app, "3/2025", date(2030, 6, 15))
|
||||
melde_an(client)
|
||||
|
||||
bericht = erinnerungs_plugin(app).pruefe_und_erinnere_sync()
|
||||
|
||||
assert bericht.gepruefte_ausgaben == 0
|
||||
assert stubs.benachrichtigung.gesendet == []
|
||||
|
||||
|
||||
@freeze_time("2030-05-25") # 21 Tage Rest → im Fenster
|
||||
def test_rezensent_mit_nur_abgeschlossenen_eintraegen_wird_nicht_erinnert(client, app, stubs):
|
||||
for name, status in (
|
||||
("rez-fertig", "abgeschlossen"),
|
||||
("x-in_arbeit", "in_bearbeitung"),
|
||||
("x-fertig", "abgeschlossen"),
|
||||
):
|
||||
lege_benutzer_an(app, name, "rezensent")
|
||||
planung_anlegen(app, f"Spiel von {name}",
|
||||
benutzer_objekt(app, name).id, status=status)
|
||||
ausgabe_anlegen(app, "3/2025", date(2030, 6, 15)) # 21 Tage Rest → im Fenster
|
||||
melde_an(client)
|
||||
|
||||
bericht = erinnerungs_plugin(app).pruefe_und_erinnere_sync()
|
||||
|
||||
# „offen“ heißt: nicht abgeschlossen (offen oder in Bearbeitung).
|
||||
erinnerte = {v["benutzer"] for v in bericht.versendet}
|
||||
assert erinnerte == {"x-in_arbeit"}
|
||||
|
||||
|
||||
@freeze_time("2030-05-20")
|
||||
def test_keine_doppelerinnerung_pro_ausgabe_und_benutzer(client, app, stubs):
|
||||
_drei_beteiligte(app)
|
||||
ausgabe_anlegen(app, "3/2025", date(2030, 6, 15))
|
||||
melde_an(client)
|
||||
plugin = erinnerungs_plugin(app)
|
||||
|
||||
erster = plugin.pruefe_und_erinnere_sync()
|
||||
zweiter = plugin.pruefe_und_erinnere_sync()
|
||||
dritter = plugin.pruefe_und_erinnere_sync(jetzt=date(2030, 6, 14)) # später, noch im Fenster
|
||||
|
||||
assert erster.anzahl_versendet > 0
|
||||
assert zweiter.anzahl_versendet == 0
|
||||
assert zweiter.uebersprungen == erster.anzahl_versendet
|
||||
assert dritter.anzahl_versendet == 0
|
||||
assert len(stubs.benachrichtigung.gesendet) == erster.anzahl_versendet
|
||||
assert len(protokoll_zeilen(app)) == erster.anzahl_versendet
|
||||
|
||||
|
||||
@freeze_time("2030-05-20")
|
||||
def test_zweite_ausgabe_wird_unabhaengig_erinnert(client, app, stubs):
|
||||
"""Doppelschutz gilt pro Ausgabe — eine zweite Ausgabe erinnert erneut."""
|
||||
_drei_beteiligte(app)
|
||||
erste = ausgabe_anlegen(app, "3/2025", date(2030, 6, 10)) # im Fenster
|
||||
zweite = ausgabe_anlegen(app, "4/2025", date(2030, 6, 12)) # auch im Fenster
|
||||
melde_an(client)
|
||||
|
||||
bericht = erinnerungs_plugin(app).pruefe_und_erinnere_sync()
|
||||
|
||||
assert bericht.gepruefte_ausgaben == 2
|
||||
assert {v["ausgabe"] for v in bericht.versendet} == {"3/2025", "4/2025"}
|
||||
ids = protokoll_zeilen(app)
|
||||
rez_offen = benutzer_objekt(app, "rez-offen")
|
||||
assert (erste, rez_offen.id) in ids
|
||||
assert (zweite, rez_offen.id) in ids
|
||||
|
||||
|
||||
@freeze_time("2030-05-20")
|
||||
def test_rollenfilter_admin_und_redakteur_werden_nie_erinnert(client, app, stubs):
|
||||
lege_benutzer_an(app, "chef-redakteur", "redakteur")
|
||||
redakteur = benutzer_objekt(app, "chef-redakteur")
|
||||
# Redakteur mit offenen Planungseinträgen bleibt trotzdem außen vor:
|
||||
planung_anlegen(app, "Vom Redakteur übernommen", redakteur.id, status="offen")
|
||||
|
||||
ausgabe_anlegen(app, "3/2025", date(2030, 6, 15))
|
||||
melde_an(client)
|
||||
bericht = erinnerungs_plugin(app).pruefe_und_erinnere_sync()
|
||||
|
||||
erinnerte = {v["benutzer"] for v in bericht.versendet}
|
||||
assert "chef-redakteur" not in erinnerte
|
||||
assert "admin" not in erinnerte
|
||||
|
||||
|
||||
@freeze_time("2030-05-20")
|
||||
def test_inaktiver_rezensent_wird_nicht_erinnert(app, client, stubs):
|
||||
lege_benutzer_an(app, "ex-rez", "rezensent")
|
||||
with app.state.session_factory() as db:
|
||||
ex_rez = db.scalar(select(User).where(User.username == "ex-rez"))
|
||||
ex_rez.active = False
|
||||
db.commit()
|
||||
ausgabe_anlegen(app, "3/2025", date(2030, 6, 15))
|
||||
|
||||
bericht = erinnerungs_plugin(app).pruefe_und_erinnere_sync()
|
||||
assert bericht.versendet == []
|
||||
|
||||
|
||||
@freeze_time("2030-05-20")
|
||||
def test_monkeypatched_uhr_steuert_job_pfad(client, app, stubs, monkeypatch):
|
||||
"""Auch ohne `jetzt`-Argument nutzt der Durchlauf die eingefrierbare Uhr."""
|
||||
dienst = plugin_modul(app, "erinnerung").dienst
|
||||
monkeypatch.setattr(dienst, "heute", lambda: date(2030, 5, 20))
|
||||
_drei_beteiligte(app)
|
||||
ausgabe_anlegen(app, "3/2025", date(2030, 6, 15))
|
||||
|
||||
bericht = erinnerungs_plugin(app).pruefe_und_erinnere_sync()
|
||||
assert bericht.anzahl_versendet == 2
|
||||
|
||||
|
||||
@freeze_time("2030-05-20")
|
||||
def test_manuelle_pruefung_ueber_route(client, app, stubs):
|
||||
_drei_beteiligte(app)
|
||||
ausgabe_anlegen(app, "3/2025", date(2030, 6, 15))
|
||||
melde_an(client)
|
||||
antwort = client.post("/erinnerung/pruefen", follow_redirects=True)
|
||||
assert "2 Erinnerung(en) versendet" in sichtbar(antwort)
|
||||
|
||||
|
||||
@freeze_time("2030-05-20")
|
||||
def test_integration_mit_echten_plugins(settings):
|
||||
"""Echte Plugins statt Stubs: In-App-Nachricht + Audit-Eintrag in der DB."""
|
||||
from tests.conftest import ADMIN_PASSWORD
|
||||
from redaktionskern.auth.security import hash_password
|
||||
|
||||
with freeze_time("2030-05-20"):
|
||||
app2 = create_app(settings)
|
||||
with TestClient(app2):
|
||||
# Rezensenten anlegen (Admin existiert bereits als Initial-Admin):
|
||||
with app2.state.session_factory() as db:
|
||||
db.add(User(username="rez-integration",
|
||||
display_name="Rez Integration",
|
||||
password_hash=hash_password(ADMIN_PASSWORD),
|
||||
role="rezensent", active=True))
|
||||
db.commit()
|
||||
rez = None
|
||||
with app2.state.session_factory() as db:
|
||||
rez = db.scalar(select(User).where(User.username == "rez-integration"))
|
||||
|
||||
ausgabe_anlegen(app2, "3/2025", date(2030, 6, 15))
|
||||
bericht = erinnerungs_plugin(app2).pruefe_und_erinnere_sync()
|
||||
assert bericht.anzahl_versendet == 1
|
||||
assert bericht.versendet[0]["kanaele"] == ["inapp"]
|
||||
|
||||
Benachrichtigung = sys.modules[
|
||||
type(app2.state.registry.get("benachrichtigung")).__module__
|
||||
].Benachrichtigung
|
||||
AuditEintrag = sys.modules[
|
||||
type(app2.state.registry.get("audit-log")).__module__
|
||||
].AuditEintrag
|
||||
ErinnerungsProtokoll = sys.modules[
|
||||
type(app2.state.registry.get("erinnerung")).__module__
|
||||
].ErinnerungsProtokoll
|
||||
with app2.state.session_factory() as db:
|
||||
nachricht = db.scalar(select(Benachrichtigung))
|
||||
assert nachricht is not None
|
||||
assert nachricht.user_id == rez.id
|
||||
assert nachricht.titel == "Redaktionsschluss: 3/2025"
|
||||
assert "Redaktionsschluss" in nachricht.text
|
||||
|
||||
eintraege = list(db.scalars(
|
||||
select(AuditEintrag).where(AuditEintrag.action == "benachrichtigt")
|
||||
).all())
|
||||
assert len(eintraege) == 1
|
||||
assert eintraege[0].actor_name == "System"
|
||||
assert eintraege[0].objekt_typ == "erinnerung"
|
||||
assert eintraege[0].details["ausgabe"] == "3/2025"
|
||||
|
||||
zeile = db.scalar(select(ErinnerungsProtokoll))
|
||||
assert zeile.user_id == rez.id
|
||||
assert zeile.kanaele == "inapp"
|
||||
|
||||
|
||||
# ---------------- Scheduler-Lifecycle & Konfiguration ----------------
|
||||
|
||||
def test_scheduler_standardmaessig_deaktiviert_in_tests(settings):
|
||||
app = create_app(settings)
|
||||
assert app.state.registry.get("erinnerung")._scheduler is None
|
||||
|
||||
|
||||
def test_scheduler_laesst_sich_aktivieren_und_konfigurieren(monkeypatch, settings):
|
||||
monkeypatch.setenv("SPIELE_ERINNERUNG_JOB_AKTIV", "1")
|
||||
monkeypatch.setenv("SPIELE_ERINNERUNG_JOB_UHRZEIT", "7:30")
|
||||
app = create_app(settings)
|
||||
plugin = app.state.registry.get("erinnerung")
|
||||
with TestClient(app):
|
||||
assert plugin._scheduler is not None
|
||||
job = plugin._scheduler.get_job("erinnerungs-check")
|
||||
assert job is not None
|
||||
felder = {feld.name: str(feld) for feld in job.trigger.fields}
|
||||
assert felder["hour"] == "7"
|
||||
assert felder["minute"] == "30"
|
||||
assert plugin._scheduler is None # on_unload hat ihn gestoppt
|
||||
|
||||
|
||||
def test_scheduler_kaputte_uhrzeit_faellt_auf_standard(monkeypatch, settings):
|
||||
monkeypatch.setenv("SPIELE_ERINNERUNG_JOB_AKTIV", "1")
|
||||
monkeypatch.setenv("SPIELE_ERINNERUNG_JOB_UHRZEIT", "garstin:keinzahl")
|
||||
app = create_app(settings)
|
||||
plugin = app.state.registry.get("erinnerung")
|
||||
with TestClient(app):
|
||||
job = plugin._scheduler.get_job("erinnerungs-check")
|
||||
felder = {feld.name: str(feld) for feld in job.trigger.fields}
|
||||
assert felder["hour"] == "8"
|
||||
assert felder["minute"] == "0"
|
||||
|
||||
|
||||
# ---------------- Fenster-Grenzlogik (Unit) ----------------
|
||||
|
||||
def test_fenster_grenzen():
|
||||
from plugins.erinnerung.dienst import (
|
||||
ERINNERUNGSFENSTER_TAGE,
|
||||
im_erinnerungsfenster,
|
||||
tage_bis,
|
||||
)
|
||||
|
||||
schluss = date(2030, 6, 15)
|
||||
assert tage_bis(schluss, date(2030, 5, 18)) == ERINNERUNGSFENSTER_TAGE
|
||||
assert im_erinnerungsfenster(schluss, date(2030, 5, 18)) # genau 28 Tage
|
||||
assert not im_erinnerungsfenster(schluss, date(2030, 5, 17)) # 29 Tage → zu früh
|
||||
assert im_erinnerungsfenster(schluss, date(2030, 6, 15)) # am Schluss selbst
|
||||
assert not im_erinnerungsfenster(schluss, date(2030, 6, 16)) # vorbei
|
||||
@@ -41,7 +41,9 @@ def test_plugins_haben_navigation(app):
|
||||
|
||||
# Voll implementierte Plugins tragen keinen Platzhalter-Text mehr auf ihrer
|
||||
# Seite; die Prüfung läuft weiter für alle übrigen Platzhalter.
|
||||
VOLL_IMPLEMENTIERT = {"audit-log", "neuheiten", "benachrichtigung", "dedup", "planung"}
|
||||
VOLL_IMPLEMENTIERT = {
|
||||
"audit-log", "neuheiten", "benachrichtigung", "dedup", "planung", "erinnerung",
|
||||
}
|
||||
PLATZHALTER_STUBS = ERWARTETE_STUBS - VOLL_IMPLEMENTIERT
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user