72fe3b7eda
- Eigene Migration 0001_archiv_tabellen: Spiegel-Tabellen archiv_neuheiten und archiv_planung (ohne FK, damit Archiv-Zeilen Benutzerlöschung überleben) - Service mit eingefrierbarer Uhr: Stichtag „jetzt minus 12 Monate“ in UTC, strenger Vergleich (genau 12 Monate bleibt aktiv); Neuheiten nach Erscheinungsjahr (großzügig Jahresende) sonst Eintragsdatum, Planung nach Eintragsdatum; Monatsarithmetik mit Klemmung (29. Februar) - Täglicher APScheduler-Cron-Job (Standard 03:00, konfigurierbar über SPIELE_ARCHIV_JOB_UHRZEIT / SPIELE_ARCHIV_JOB_AKTIV); je archiviertem Titel ein Audit-Log-Eintrag als System - Admin-Ansicht /archiv: vereinigte Liste beider Tabellen, Suche über Titel/Verlag/Autor, Herkunftsfilter, Wiederherstellen in die aktive Liste (BGG-Konflikt wird abgelehnt, verwaiste Rezensentenzuordnung übernimmt der Admin — gemeldet und auditiert); Rollen: nur Admin - Kern minimal erweitert: PluginContext.registry stellt Hintergrund-Jobs die Plugin-Registry bereit (keine Fachlogik im Kern) - 30 neue Tests mit eingefrorener Uhr (Grenzfälle, Zeitzonen, Schaltjahr, Job-Lauf, Wiederherstellen, Rollen, Suche/Filter); 171 Tests grün - README aktualisiert: neues Plugin-Kapitel, Env-Variablen, Fortschritt #7 ✅
676 lines
25 KiB
Python
676 lines
25 KiB
Python
"""Tests: Plugin „archiv“ — 12-Monats-Regel, Job, Wiederherstellen, Rollen.
|
||
|
||
Alle Zeitpunkte werden mit eingefrorener Uhr geprüft: Der Service nimmt eine
|
||
einfrierbare „jetzt“-Fabrik entgegen (Plugin-Attribut `_jetzt` bzw.
|
||
Konstruktor-Argument), die Systemuhr wird nicht manipuliert. HTTP-Antworten
|
||
nutzen den echten TestClient; die Partner-Plugins (audit-log, neuheiten,
|
||
planung) laufen echt mit — nur ihre Tabellen werden direkt befüllt.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import html
|
||
import tempfile
|
||
from datetime import datetime, timedelta, timezone
|
||
|
||
from fastapi.testclient import TestClient
|
||
from sqlalchemy import insert, select
|
||
|
||
from redaktionskern.app import create_app
|
||
from redaktionskern.auth.models import User
|
||
from redaktionskern.config import Settings
|
||
from redaktionskern.contracts import Migration
|
||
from redaktionskern.db import Base
|
||
from tests.conftest import ADMIN_PASSWORD, lege_benutzer_an, melde_an
|
||
|
||
from plugins.archiv.models import ArchivNeuheit, ArchivPlanung
|
||
from plugins.archiv.service import ArchivService, als_utc, naiv_utc, vor_monaten
|
||
|
||
UTC = timezone.utc
|
||
|
||
|
||
# ---------------- Hilfen ----------------
|
||
|
||
def einfriere_uhr(jahr, monat, tag, stunde=12, minute=0, sekunde=0):
|
||
"""Fabrik für eine eingefrorene UTC-Uhr."""
|
||
moment = datetime(jahr, monat, tag, stunde, minute, sekunde, tzinfo=UTC)
|
||
return lambda: moment
|
||
|
||
|
||
def archiv_service(app, jetzt) -> ArchivService:
|
||
return ArchivService(app.state.session_factory, jetzt=jetzt)
|
||
|
||
|
||
def friere_plugin_uhr_ein(app, jetzt) -> None:
|
||
app.state.registry.get("archiv")._jetzt = jetzt
|
||
|
||
|
||
def fuege_neuheit_hinzu(app, **felder) -> int:
|
||
felder.setdefault("titel", "Testspiel")
|
||
felder.setdefault("bgg_id", 4711)
|
||
tabelle = Base.metadata.tables["neuheiten"]
|
||
with app.state.session_factory() as db:
|
||
ergebnis = db.execute(insert(tabelle).values(**felder))
|
||
db.commit()
|
||
return ergebnis.inserted_primary_key[0]
|
||
|
||
|
||
def fuege_planung_hinzu(app, **felder) -> int:
|
||
felder.setdefault("titel", "Testspiel")
|
||
felder.setdefault("rezensent_id", 1)
|
||
tabelle = Base.metadata.tables["planungsliste"]
|
||
with app.state.session_factory() as db:
|
||
ergebnis = db.execute(insert(tabelle).values(**felder))
|
||
db.commit()
|
||
return ergebnis.inserted_primary_key[0]
|
||
|
||
|
||
def _alle(app, tabellenname: str) -> list[dict]:
|
||
tabelle = Base.metadata.tables[tabellenname]
|
||
with app.state.session_factory() as db:
|
||
return [dict(z) for z in db.execute(select(tabelle)).mappings().all()]
|
||
|
||
|
||
def zaehle(app, tabellenname: str) -> int:
|
||
return len(_alle(app, tabellenname))
|
||
|
||
|
||
def audit_zeilen(app) -> list[dict]:
|
||
return _alle(app, "audit_eintraege")
|
||
|
||
|
||
def sichtbar(antwort) -> str:
|
||
return html.unescape(antwort.text)
|
||
|
||
|
||
def cron_feld(trigger, name: str) -> int:
|
||
"""Liest ein CronTrigger-Feld (z. B. hour=3 → 3)."""
|
||
feld = next(f for f in trigger.fields if f.name == name)
|
||
ausdruck = feld.expressions[0]
|
||
return int(getattr(ausdruck, "first", ausdruck))
|
||
|
||
|
||
JETZT = einfriere_uhr(2025, 8, 21, 12, 0, 0) # Stichtag: 21.08.2024 12:00 UTC
|
||
|
||
|
||
# ---------------- Migration ----------------
|
||
|
||
def test_migration_legt_beide_archiv_tabellen_an(app):
|
||
from sqlalchemy import inspect
|
||
|
||
inspektion = inspect(app.state.engine)
|
||
neuheiten_spalten = {c["name"] for c in inspektion.get_columns("archiv_neuheiten")}
|
||
planung_spalten = {c["name"] for c in inspektion.get_columns("archiv_planung")}
|
||
meta = {"quell_id", "archiviert_am", "archiviert_von", "grund", "referenz_am"}
|
||
assert {
|
||
"id", "titel", "verlag", "autor", "erscheinungsjahr",
|
||
"bgg_id", "status", "quelle", "erstellt_am", "aktualisiert_am",
|
||
} | meta <= neuheiten_spalten
|
||
assert {
|
||
"id", "titel", "verlag", "autor", "bgg_id", "ausgabe", "rezensent_id",
|
||
"status", "notizen", "quelle", "pruefung",
|
||
"erstellt_am", "aktualisiert_am",
|
||
} | meta <= planung_spalten
|
||
|
||
|
||
# ---------------- Monatsarithmetik & Zeitzonen (Einheitstests) ----------------
|
||
|
||
def test_vor_monaten_klemmt_monatstage():
|
||
assert vor_monaten(datetime(2025, 3, 31), 1) == datetime(2025, 2, 28)
|
||
assert vor_monaten(datetime(2024, 2, 29), 12) == datetime(2023, 2, 28)
|
||
assert vor_monaten(datetime(2024, 3, 31), 1) == datetime(2024, 2, 29) # Schaltjahr
|
||
assert vor_monaten(datetime(2025, 1, 15), 12) == datetime(2024, 1, 15)
|
||
|
||
|
||
def test_als_utc_normalisiert_naive_und_bewohnte_zeiten():
|
||
naive = datetime(2025, 3, 10, 23, 30)
|
||
assert als_utc(naive).tzinfo == UTC
|
||
assert als_utc(naive).hour == 23 # naive Werte gelten als UTC
|
||
|
||
berlin = timezone(timedelta(hours=2))
|
||
bewohnt = datetime(2025, 3, 11, 1, 0, tzinfo=berlin) # = 23:00 UTC am 10.03.
|
||
assert als_utc(bewohnt).utcoffset() == timedelta(0)
|
||
assert (als_utc(bewohnt).month, als_utc(bewohnt).day, als_utc(bewohnt).hour) == (3, 10, 23)
|
||
assert naiv_utc(bewohnt) == datetime(2025, 3, 10, 23, 0)
|
||
|
||
|
||
def test_stichtag_rechnet_in_utc_unabhaengig_von_der_uhrzeitzone():
|
||
berlin = timezone(timedelta(hours=2))
|
||
# Die Uhr steht auf Berliner Zeit: 01:30 +02:00 == 23:30 UTC am Vortag.
|
||
service = ArchivService(
|
||
session_factory=None,
|
||
jetzt=lambda: datetime(2025, 3, 11, 1, 30, tzinfo=berlin),
|
||
)
|
||
assert service.jetzt().utcoffset() == timedelta(0)
|
||
assert (service.jetzt().hour, service.jetzt().minute) == (23, 30)
|
||
assert service.stichtag() == datetime(2024, 3, 10, 23, 30, tzinfo=UTC)
|
||
|
||
|
||
# ---------------- 12-Monats-Regel: Neuheiten ----------------
|
||
|
||
def test_alte_neuheit_wird_vollstaendig_archiviert_und_auditiert(app):
|
||
quell_id = fuege_neuheit_hinzu(
|
||
app,
|
||
titel="Alter Klassiker",
|
||
verlag="Alt-Verlag",
|
||
autor="A. Autorin",
|
||
erscheinungsjahr=2023,
|
||
bgg_id=1001,
|
||
status="neuheit",
|
||
quelle="boardgamegeek",
|
||
erstellt_am=datetime(2023, 5, 1, 10, 0),
|
||
)
|
||
|
||
lauf, ereignisse = archiv_service(app, JETZT).lauf()
|
||
# Audit-Einträge schreibt der Aufrufkontext (Route/Job) über die Plugin-API:
|
||
app.state.registry.get("archiv")._auditiere_archivierung(ereignisse)
|
||
|
||
assert (lauf.neuheiten_archiviert, lauf.planung_archiviert) == (1, 0)
|
||
assert len(ereignisse) == 1
|
||
assert zaehle(app, "neuheiten") == 0
|
||
|
||
zeile = _alle(app, "archiv_neuheiten")[0]
|
||
# Sicherungsdatei-Prinzip: alle Felder vollständig erhalten.
|
||
assert zeile["quell_id"] == quell_id
|
||
assert zeile["titel"] == "Alter Klassiker"
|
||
assert zeile["verlag"] == "Alt-Verlag"
|
||
assert zeile["autor"] == "A. Autorin"
|
||
assert zeile["erscheinungsjahr"] == 2023
|
||
assert zeile["bgg_id"] == 1001
|
||
assert zeile["status"] == "neuheit"
|
||
assert zeile["quelle"] == "boardgamegeek"
|
||
assert zeile["erstellt_am"] == datetime(2023, 5, 1, 10, 0)
|
||
# Referenz = Ende des Erscheinungsjahres (BGG liefert nur das Jahr).
|
||
assert zeile["referenz_am"] == datetime(2023, 12, 31, 23, 59, 59)
|
||
assert "2023" in zeile["grund"]
|
||
assert zeile["archiviert_von"] == "System"
|
||
|
||
# Genau ein Audit-Eintrag pro Archivierung, Akteur „System“.
|
||
eintraege = audit_zeilen(app)
|
||
assert len(eintraege) == 1
|
||
audit = eintraege[0]
|
||
assert audit["actor_name"] == "System"
|
||
assert audit["action"] == "verschoben"
|
||
assert audit["objekt_typ"] == "neuheit"
|
||
assert audit["objekt_id"] == str(quell_id)
|
||
assert audit["details"]["titel"] == "Alter Klassiker"
|
||
assert audit["details"]["ziel"] == "archiv"
|
||
assert "12 Monate" in audit["details"]["grund"]
|
||
|
||
|
||
def test_junge_neuheiten_bleiben_in_der_aktiven_liste(app):
|
||
fuege_neuheit_hinzu(app, titel="Frisch 2025", bgg_id=1, erscheinungsjahr=2025)
|
||
fuege_neuheit_hinzu(app, titel="Letztes Jahr", bgg_id=2, erscheinungsjahr=2024)
|
||
fuege_neuheit_hinzu(app, titel="Uralt", bgg_id=3, erscheinungsjahr=2023)
|
||
|
||
lauf, _ = archiv_service(app, JETZT).lauf()
|
||
|
||
# Ein Titel des Jahres 2024 zählt erst nach dem 31.12.2025 als älter als
|
||
# 12 Monate (großzügige Jahres-Auslegung) — nur 2023 fällt unter die Regel.
|
||
assert lauf.neuheiten_archiviert == 1
|
||
assert {z["titel"] for z in _alle(app, "neuheiten")} == {
|
||
"Frisch 2025", "Letztes Jahr",
|
||
}
|
||
assert {z["titel"] for z in _alle(app, "archiv_neuheiten")} == {"Uralt"}
|
||
|
||
|
||
def test_neuheit_ohne_jahr_nutzt_eintragsdatum(app):
|
||
alt = datetime(2024, 6, 1, 9, 0) # > 12 Monate vor dem 21.08.2025
|
||
jung = datetime(2024, 10, 1, 9, 0) # < 12 Monate
|
||
fuege_neuheit_hinzu(app, titel="Ohne Jahr alt", bgg_id=10,
|
||
erscheinungsjahr=None, erstellt_am=alt)
|
||
fuege_neuheit_hinzu(app, titel="Ohne Jahr jung", bgg_id=11,
|
||
erscheinungsjahr=None, erstellt_am=jung)
|
||
|
||
lauf, _ = archiv_service(app, JETZT).lauf()
|
||
|
||
assert lauf.neuheiten_archiviert == 1
|
||
zeile = _alle(app, "archiv_neuheiten")[0]
|
||
assert zeile["titel"] == "Ohne Jahr alt"
|
||
assert zeile["referenz_am"] == alt
|
||
assert "eingetragen" in zeile["grund"]
|
||
|
||
|
||
# ---------------- Grenzfälle & Planungseinträge ----------------
|
||
|
||
def test_grenzfall_genau_12_monate_bleibt_eine_sekunde_elter_faellt(app):
|
||
stichtag = datetime(2024, 3, 15, 12, 0) # exakt „jetzt − 12 Monate“
|
||
fuege_planung_hinzu(app, titel="Exakt 12 Monate", erstellt_am=stichtag)
|
||
fuege_planung_hinzu(
|
||
app, titel="Eine Sekunde älter", erstellt_am=stichtag - timedelta(seconds=1)
|
||
)
|
||
fuege_planung_hinzu(
|
||
app, titel="Eine Sekunde jünger", erstellt_am=stichtag + timedelta(seconds=1)
|
||
)
|
||
|
||
lauf, _ = archiv_service(app, einfriere_uhr(2025, 3, 15, 12, 0, 0)).lauf()
|
||
|
||
# „älter als 12 Monate“ ist streng: genau 12 Monate bleibt aktiv.
|
||
assert lauf.planung_archiviert == 1
|
||
assert {z["titel"] for z in _alle(app, "archiv_planung")} == {"Eine Sekunde älter"}
|
||
assert {z["titel"] for z in _alle(app, "planungsliste")} == {
|
||
"Exakt 12 Monate", "Eine Sekunde jünger",
|
||
}
|
||
|
||
|
||
def test_zeitzone_grenzfall_um_utc_mitternacht(app):
|
||
# Eingefrorene UTC-Uhr 23:30:30 → Stichtag 2024-03-10 23:30:30 UTC.
|
||
# In Berlin wäre es bereits der 11.03., 00:30 — die Entscheidung folgt
|
||
# trotzdem konsequent UTC.
|
||
jetzt = einfriere_uhr(2025, 3, 10, 23, 30, 30)
|
||
fuege_planung_hinzu(
|
||
app, titel="Knapp drunter", erstellt_am=datetime(2024, 3, 10, 23, 30, 0)
|
||
)
|
||
fuege_planung_hinzu(
|
||
app, titel="Knapp drueber", erstellt_am=datetime(2024, 3, 10, 23, 31, 0)
|
||
)
|
||
|
||
lauf, _ = archiv_service(app, jetzt).lauf()
|
||
|
||
assert lauf.planung_archiviert == 1
|
||
assert {z["titel"] for z in _alle(app, "archiv_planung")} == {"Knapp drunter"}
|
||
assert {z["titel"] for z in _alle(app, "planungsliste")} == {"Knapp drueber"}
|
||
|
||
|
||
def test_schaltjahr_29_februar_wird_geklemmt_nicht_uebersprungen(app):
|
||
fuege_planung_hinzu(
|
||
app, titel="29. Februar", erstellt_am=datetime(2024, 2, 29, 12, 0)
|
||
)
|
||
|
||
# 28.02.2025: Stichtag 29.02.2024 wird auf den 28.02 geklemmt — der
|
||
# Eintrag liegt einen Tag „danach“ und bleibt aktiv.
|
||
lauf, _ = archiv_service(app, einfriere_uhr(2025, 2, 28, 12, 0)).lauf()
|
||
assert lauf.planung_archiviert == 0
|
||
|
||
# 01.03.2025: Stichtag 01.03.2024 → der 29.02.2024 ist älter → Archiv.
|
||
lauf, _ = archiv_service(app, einfriere_uhr(2025, 3, 1, 12, 0)).lauf()
|
||
assert lauf.planung_archiviert == 1
|
||
assert {z["titel"] for z in _alle(app, "archiv_planung")} == {"29. Februar"}
|
||
|
||
|
||
def test_planungseintrag_wird_vollstaendig_archiviert(app):
|
||
fuege_planung_hinzu(
|
||
app,
|
||
titel="Altes Planungsspiel",
|
||
verlag="Plan-Verlag",
|
||
autor="B. Autor",
|
||
bgg_id=2002,
|
||
ausgabe="3/2024",
|
||
rezensent_id=1,
|
||
status="in_bearbeitung",
|
||
notizen="Rezension angefangen",
|
||
quelle="neuheiten",
|
||
pruefung=[{"art": "titel", "beschreibung": "ähnlicher Titel"}],
|
||
erstellt_am=datetime(2024, 5, 1, 8, 30),
|
||
)
|
||
|
||
lauf, ereignisse = archiv_service(app, JETZT).lauf()
|
||
app.state.registry.get("archiv")._auditiere_archivierung(ereignisse)
|
||
|
||
assert (lauf.neuheiten_archiviert, lauf.planung_archiviert) == (0, 1)
|
||
assert zaehle(app, "planungsliste") == 0
|
||
|
||
zeile = _alle(app, "archiv_planung")[0]
|
||
assert zeile["titel"] == "Altes Planungsspiel"
|
||
assert zeile["verlag"] == "Plan-Verlag"
|
||
assert zeile["autor"] == "B. Autor"
|
||
assert zeile["bgg_id"] == 2002
|
||
assert zeile["ausgabe"] == "3/2024"
|
||
assert zeile["status"] == "in_bearbeitung"
|
||
assert zeile["notizen"] == "Rezension angefangen"
|
||
assert zeile["quelle"] == "neuheiten"
|
||
assert zeile["pruefung"] == [{"art": "titel", "beschreibung": "ähnlicher Titel"}]
|
||
assert zeile["erstellt_am"] == datetime(2024, 5, 1, 8, 30)
|
||
assert zeile["referenz_am"] == datetime(2024, 5, 1, 8, 30)
|
||
|
||
audit = audit_zeilen(app)[0]
|
||
assert audit["objekt_typ"] == "planungseintrag"
|
||
assert audit["actor_name"] == "System"
|
||
assert audit["details"]["herkunft"] == "planung"
|
||
|
||
|
||
def test_lauf_ist_idempotent(app):
|
||
fuege_neuheit_hinzu(app, titel="Alt", bgg_id=1, erscheinungsjahr=2020)
|
||
fuege_planung_hinzu(app, titel="Alt geplant", erstellt_am=datetime(2023, 1, 1))
|
||
|
||
service = archiv_service(app, JETZT)
|
||
erster, ereignisse_eins = service.lauf()
|
||
zweiter, ereignisse_zwei = service.lauf()
|
||
plugin = app.state.registry.get("archiv")
|
||
plugin._auditiere_archivierung(ereignisse_eins)
|
||
plugin._auditiere_archivierung(ereignisse_zwei)
|
||
|
||
assert erster.gesamt == 2
|
||
assert len(ereignisse_eins) == 2
|
||
assert zweiter.gesamt == 0
|
||
assert ereignisse_zwei == []
|
||
assert zaehle(app, "archiv_neuheiten") == 1
|
||
assert zaehle(app, "archiv_planung") == 1
|
||
assert len([a for a in audit_zeilen(app) if a["actor_name"] == "System"]) == 2
|
||
|
||
|
||
def test_fehlende_partner_tabellen_brechen_nicht(tmp_path):
|
||
"""Ohne neuheiten-/planungsliste-Tabelle läuft der Lauf leer durch."""
|
||
from redaktionskern.db import make_engine, make_session_factory
|
||
|
||
engine = make_engine(f"sqlite:///{tmp_path / 'leer.db'}")
|
||
fabrik = make_session_factory(engine)
|
||
with engine.begin() as conn:
|
||
ArchivNeuheit.__table__.create(conn, checkfirst=True)
|
||
ArchivPlanung.__table__.create(conn, checkfirst=True)
|
||
|
||
try:
|
||
lauf, ereignisse = ArchivService(fabrik, jetzt=JETZT).lauf()
|
||
assert lauf.gesamt == 0
|
||
assert ereignisse == []
|
||
finally:
|
||
engine.dispose()
|
||
|
||
|
||
# ---------------- Hintergrund-Job & Scheduler ----------------
|
||
|
||
def test_job_laeuft_und_schreibt_audit_eintraege(app):
|
||
friere_plugin_uhr_ein(app, JETZT)
|
||
fuege_neuheit_hinzu(app, titel="Job-Kandidat", bgg_id=3003, erscheinungsjahr=2022)
|
||
fuege_planung_hinzu(app, titel="Job-Planung", erstellt_am=datetime(2023, 2, 2))
|
||
|
||
plugin = app.state.registry.get("archiv")
|
||
plugin._archiv_job()
|
||
|
||
assert zaehle(app, "neuheiten") == 0
|
||
assert zaehle(app, "planungsliste") == 0
|
||
assert zaehle(app, "archiv_neuheiten") == 1
|
||
assert zaehle(app, "archiv_planung") == 1
|
||
audit = audit_zeilen(app)
|
||
assert len(audit) == 2
|
||
assert {a["objekt_typ"] for a in audit} == {"neuheit", "planungseintrag"}
|
||
|
||
|
||
def test_job_ohne_audit_log_plugin_laeuft_trotzdem(app, monkeypatch):
|
||
friere_plugin_uhr_ein(app, JETZT)
|
||
fuege_neuheit_hinzu(app, titel="Ohne Audit", bgg_id=3004, erscheinungsjahr=2021)
|
||
|
||
registry = app.state.registry
|
||
original_get = registry.get
|
||
monkeypatch.setattr(
|
||
registry,
|
||
"get",
|
||
lambda name: None if name == "audit-log" else original_get(name),
|
||
)
|
||
|
||
plugin = app.state.registry.get("archiv")
|
||
plugin._archiv_job() # darf nicht werfen
|
||
assert zaehle(app, "archiv_neuheiten") == 1
|
||
assert audit_zeilen(app) == []
|
||
|
||
|
||
def test_scheduler_standardmaessig_deaktiviert_in_tests(app):
|
||
assert app.state.registry.get("archiv")._scheduler is None
|
||
|
||
|
||
def test_scheduler_laesst_sich_aktivieren_und_stoppt_beim_entladen(
|
||
monkeypatch, settings
|
||
):
|
||
monkeypatch.setenv("SPIELE_ARCHIV_JOB_AKTIV", "1")
|
||
|
||
app = create_app(settings)
|
||
plugin = app.state.registry.get("archiv")
|
||
with TestClient(app):
|
||
assert plugin._scheduler is not None
|
||
assert plugin._scheduler.running
|
||
job = plugin._scheduler.get_job("archiv-taeglich")
|
||
assert job is not None
|
||
assert cron_feld(job.trigger, "hour") == 3 # Standard 03:00 …
|
||
assert cron_feld(job.trigger, "minute") == 0 # … täglich
|
||
|
||
assert plugin._scheduler is None # on_unload hat ihn gestoppt
|
||
|
||
|
||
def test_scheduler_uhrzeit_ist_konfigurierbar(monkeypatch, settings):
|
||
monkeypatch.setenv("SPIELE_ARCHIV_JOB_AKTIV", "1")
|
||
monkeypatch.setenv("SPIELE_ARCHIV_JOB_UHRZEIT", "05:30")
|
||
app = create_app(settings)
|
||
plugin = app.state.registry.get("archiv")
|
||
with TestClient(app):
|
||
job = plugin._scheduler.get_job("archiv-taeglich")
|
||
assert cron_feld(job.trigger, "hour") == 5
|
||
assert cron_feld(job.trigger, "minute") == 30
|
||
|
||
|
||
def test_scheduler_unsinnige_uhrzeit_faellt_auf_standard(monkeypatch, settings):
|
||
monkeypatch.setenv("SPIELE_ARCHIV_JOB_AKTIV", "1")
|
||
monkeypatch.setenv("SPIELE_ARCHIV_JOB_UHRZEIT", "25:99")
|
||
app = create_app(settings)
|
||
plugin = app.state.registry.get("archiv")
|
||
with TestClient(app):
|
||
job = plugin._scheduler.get_job("archiv-taeglich")
|
||
assert cron_feld(job.trigger, "hour") == 3
|
||
assert cron_feld(job.trigger, "minute") == 0
|
||
|
||
|
||
def test_scheduler_aktiv_ist_der_standard(monkeypatch, settings):
|
||
monkeypatch.setenv("SPIELE_ARCHIV_JOB_AKTIV", "1")
|
||
app = create_app(settings)
|
||
plugin = app.state.registry.get("archiv")
|
||
with TestClient(app):
|
||
assert plugin._scheduler is not None and plugin._scheduler.running
|
||
|
||
|
||
# ---------------- Wiederherstellen ----------------
|
||
|
||
def test_wiederherstellen_neuheit_zurueck_in_aktive_liste(app, client):
|
||
melde_an(client)
|
||
friere_plugin_uhr_ein(app, JETZT)
|
||
fuege_neuheit_hinzu(
|
||
app, titel="Rückkehrer", verlag="V", autor="A",
|
||
erscheinungsjahr=2023, bgg_id=4004, status="neuheit",
|
||
erstellt_am=datetime(2023, 2, 1),
|
||
)
|
||
app.state.registry.get("archiv")._archiv_job()
|
||
archiv_id = _alle(app, "archiv_neuheiten")[0]["id"]
|
||
|
||
folgeseite = client.post(f"/archiv/neuheiten/{archiv_id}/wiederherstellen")
|
||
assert folgeseite.status_code == 200
|
||
assert zaehle(app, "archiv_neuheiten") == 0
|
||
|
||
aktiv = _alle(app, "neuheiten")
|
||
assert len(aktiv) == 1
|
||
eintrag = aktiv[0]
|
||
assert eintrag["titel"] == "Rückkehrer"
|
||
assert eintrag["bgg_id"] == 4004
|
||
assert eintrag["erscheinungsjahr"] == 2023
|
||
assert eintrag["status"] == "neuheit"
|
||
|
||
seite = sichtbar(folgeseite)
|
||
assert "zurück in die Neuheitenliste" in seite
|
||
|
||
audit = [a for a in audit_zeilen(app) if a["actor_name"] == "admin"]
|
||
assert len(audit) == 1
|
||
assert audit[0]["action"] == "verschoben"
|
||
assert audit[0]["objekt_typ"] == "neuheit"
|
||
assert audit[0]["objekt_id"] == str(eintrag["id"])
|
||
assert audit[0]["details"]["von"] == "archiv"
|
||
assert audit[0]["details"]["ziel"] == "neuheiten"
|
||
assert audit[0]["details"]["titel"] == "Rückkehrer"
|
||
|
||
|
||
def test_wiederherstellen_planung_mit_urspruenglichem_rezensenten(app, client):
|
||
lege_benutzer_an(app, "reza", "rezensent")
|
||
with app.state.session_factory() as db:
|
||
reza_id = db.scalar(select(User.id).where(User.username == "reza"))
|
||
|
||
melde_an(client)
|
||
friere_plugin_uhr_ein(app, JETZT)
|
||
fuege_planung_hinzu(
|
||
app, titel="Plan-Rückkehrer", rezensent_id=reza_id, status="offen",
|
||
ausgabe="1/2024", erstellt_am=datetime(2023, 3, 3),
|
||
)
|
||
app.state.registry.get("archiv")._archiv_job()
|
||
archiv_id = _alle(app, "archiv_planung")[0]["id"]
|
||
|
||
antwort = client.post(
|
||
f"/archiv/planung/{archiv_id}/wiederherstellen", follow_redirects=False
|
||
)
|
||
assert antwort.status_code == 303
|
||
|
||
aktiv = _alle(app, "planungsliste")
|
||
assert len(aktiv) == 1
|
||
assert aktiv[0]["rezensent_id"] == reza_id
|
||
assert aktiv[0]["status"] == "offen"
|
||
assert aktiv[0]["ausgabe"] == "1/2024"
|
||
assert zaehle(app, "archiv_planung") == 0
|
||
|
||
|
||
def test_wiederherstellen_neuheit_konflikt_bgg_id_wird_abgelehnt(app, client):
|
||
melde_an(client)
|
||
friere_plugin_uhr_ein(app, JETZT)
|
||
fuege_neuheit_hinzu(
|
||
app, titel="Konflikt-Kandidat", bgg_id=5005, erscheinungsjahr=2022
|
||
)
|
||
app.state.registry.get("archiv")._archiv_job()
|
||
|
||
# Der Sync hat den Titel inzwischen erneut geliefert — BGG-ID wieder aktiv.
|
||
fuege_neuheit_hinzu(
|
||
app, titel="Konflikt-Kandidat (neu)", bgg_id=5005, erscheinungsjahr=2025
|
||
)
|
||
archiv_id = _alle(app, "archiv_neuheiten")[0]["id"]
|
||
|
||
antwort = client.post(f"/archiv/neuheiten/{archiv_id}/wiederherstellen")
|
||
assert antwort.status_code == 200
|
||
assert "bereits einen aktiven Neuheiten-Eintrag" in sichtbar(antwort)
|
||
|
||
# Archiv unverändert, keine Duplikate entstanden.
|
||
assert zaehle(app, "archiv_neuheiten") == 1
|
||
assert zaehle(app, "neuheiten") == 1
|
||
|
||
|
||
def test_wiederherstellen_planung_ersatz_rezensent_bei_geloeschtem_benutzer(
|
||
app, client
|
||
):
|
||
lege_benutzer_an(app, "weg", "rezensent")
|
||
with app.state.session_factory() as db:
|
||
weg_id = db.scalar(select(User.id).where(User.username == "weg"))
|
||
db.delete(db.get(User, weg_id))
|
||
db.commit()
|
||
|
||
melde_an(client)
|
||
friere_plugin_uhr_ein(app, JETZT)
|
||
fuege_planung_hinzu(
|
||
app, titel="Verwaist", rezensent_id=weg_id, erstellt_am=datetime(2023, 4, 4)
|
||
)
|
||
app.state.registry.get("archiv")._archiv_job()
|
||
archiv_id = _alle(app, "archiv_planung")[0]["id"]
|
||
|
||
with app.state.session_factory() as db:
|
||
admin_id = db.scalar(select(User.id).where(User.username == "admin"))
|
||
|
||
antwort = client.post(f"/archiv/planung/{archiv_id}/wiederherstellen")
|
||
assert antwort.status_code == 200
|
||
assert "zugeordnet" in sichtbar(antwort)
|
||
|
||
aktiv = _alle(app, "planungsliste")
|
||
assert aktiv[0]["rezensent_id"] == admin_id
|
||
audit = [a for a in audit_zeilen(app) if a["actor_name"] == "admin"][0]
|
||
assert audit["details"]["rezensent_ersetzt"] is True
|
||
|
||
|
||
def test_wiederherstellen_unbekannter_eintrag_zeigt_fehler(app, client):
|
||
melde_an(client)
|
||
antwort = client.post("/archiv/neuheiten/9999/wiederherstellen")
|
||
assert antwort.status_code == 200
|
||
assert "nicht gefunden" in sichtbar(antwort)
|
||
|
||
|
||
def test_wiederherstellen_unbekannte_herkunft_wird_abgelehnt(app, client):
|
||
melde_an(client)
|
||
folgeseite = client.post("/archiv/unsinn/1/wiederherstellen")
|
||
assert folgeseite.status_code == 200
|
||
assert "Unbekannte Herkunft" in sichtbar(folgeseite)
|
||
|
||
|
||
# ---------------- Rollen ----------------
|
||
|
||
def test_archiv_ansicht_nur_fuer_admin(app, client):
|
||
# anonym → Login
|
||
antwort = client.get("/archiv", follow_redirects=False)
|
||
assert antwort.status_code == 303
|
||
assert antwort.headers["location"].startswith("/login")
|
||
|
||
# Rezensentin → 403
|
||
lege_benutzer_an(app, "rezensentin", "rezensent")
|
||
melde_an(client, "rezensentin", "test-12345678")
|
||
assert client.get("/archiv").status_code == 403
|
||
client.post("/logout")
|
||
|
||
# Redakteur → 403 (Archiv ist reine Admin-Sicht)
|
||
lege_benutzer_an(app, "redakteur", "redakteur")
|
||
melde_an(client, "redakteur", "test-12345678")
|
||
assert client.get("/archiv").status_code == 403
|
||
client.post("/logout")
|
||
|
||
# Admin → 200
|
||
melde_an(client)
|
||
assert client.get("/archiv").status_code == 200
|
||
|
||
|
||
def test_wiederherstellen_nur_fuer_admin(app, client):
|
||
friere_plugin_uhr_ein(app, JETZT)
|
||
fuege_neuheit_hinzu(app, titel="Geschützt", bgg_id=6006, erscheinungsjahr=2022)
|
||
app.state.registry.get("archiv")._archiv_job()
|
||
archiv_id = _alle(app, "archiv_neuheiten")[0]["id"]
|
||
|
||
lege_benutzer_an(app, "rezensentin", "rezensent")
|
||
melde_an(client, "rezensentin", "test-12345678")
|
||
antwort = client.post(
|
||
f"/archiv/neuheiten/{archiv_id}/wiederherstellen", follow_redirects=False
|
||
)
|
||
assert antwort.status_code == 403
|
||
assert zaehle(app, "archiv_neuheiten") == 1 # nichts verschoben
|
||
assert zaehle(app, "neuheiten") == 0
|
||
|
||
|
||
# ---------------- Admin-UI: Suche & Filter ----------------
|
||
|
||
def test_suche_filtert_ueber_titel_verlag_autor(app, client):
|
||
melde_an(client)
|
||
friere_plugin_uhr_ein(app, JETZT)
|
||
fuege_neuheit_hinzu(app, titel="Alpha Spiel", verlag="Beta Verlag",
|
||
bgg_id=1, erscheinungsjahr=2020)
|
||
fuege_neuheit_hinzu(app, titel="Gamma Spiel", autor="Delta Autorin",
|
||
bgg_id=2, erscheinungsjahr=2020)
|
||
fuege_planung_hinzu(app, titel="Epsilon Spiel",
|
||
erstellt_am=datetime(2020, 1, 1))
|
||
app.state.registry.get("archiv")._archiv_job()
|
||
|
||
def treffer(q="", quelle=""):
|
||
seite = sichtbar(client.get(f"/archiv?q={q}&quelle={quelle}"))
|
||
return [
|
||
t for t in ("Alpha Spiel", "Gamma Spiel", "Epsilon Spiel")
|
||
if t in seite
|
||
]
|
||
|
||
assert treffer(q="alpha") == ["Alpha Spiel"]
|
||
assert treffer(q="beta") == ["Alpha Spiel"] # Treffer im Verlag
|
||
assert treffer(q="delta") == ["Gamma Spiel"] # Treffer im Autor
|
||
assert treffer(quelle="planung") == ["Epsilon Spiel"]
|
||
assert treffer(quelle="neuheiten") == ["Alpha Spiel", "Gamma Spiel"]
|
||
assert treffer() == ["Alpha Spiel", "Gamma Spiel", "Epsilon Spiel"]
|
||
|
||
leer = sichtbar(client.get("/archiv?q=nirgends"))
|
||
assert "Keine Treffer" in leer
|
||
|
||
|
||
def test_archiv_seite_zeigt_details_und_job_status(app, client):
|
||
melde_an(client)
|
||
friere_plugin_uhr_ein(app, JETZT)
|
||
fuege_neuheit_hinzu(app, titel="Anzeige-Test", bgg_id=7007,
|
||
erscheinungsjahr=2021)
|
||
app.state.registry.get("archiv")._archiv_job()
|
||
|
||
seite = sichtbar(client.get("/archiv"))
|
||
assert "Anzeige-Test" in seite
|
||
assert "Erscheinungsjahr 2021" in seite # Grund der Archivierung
|
||
assert "Neuheiten" in seite # Herkunfts-Badge
|
||
assert "Wiederherstellen" in seite
|
||
assert "täglich automatisch" in seite # Erklärungstext
|
||
assert "Täglicher Archiv-Job: aus" in seite # in Tests standardmäßig aus
|