From 72fe3b7eda1aff65ce37e4d048aa88372538778a Mon Sep 17 00:00:00 2001 From: Flo Hartmann Date: Fri, 21 Aug 2026 20:26:00 +0000 Subject: [PATCH] Plugin archiv: 12-Monats-Autopilot mit Sicherungsdatei-Prinzip MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 ✅ --- README.md | 80 ++- plugins/archiv/__init__.py | 265 +++++++- plugins/archiv/models.py | 108 ++++ plugins/archiv/service.py | 494 +++++++++++++++ plugins/archiv/templates/archiv/index.html | 119 +++- src/redaktionskern/app.py | 1 + src/redaktionskern/contracts.py | 6 + tests/conftest.py | 2 + tests/test_archiv.py | 675 +++++++++++++++++++++ tests/test_plugin_loader.py | 2 +- 10 files changed, 1727 insertions(+), 25 deletions(-) create mode 100644 plugins/archiv/models.py create mode 100644 plugins/archiv/service.py create mode 100644 tests/test_archiv.py diff --git a/README.md b/README.md index 1f83c96..226acec 100644 --- a/README.md +++ b/README.md @@ -4,7 +4,7 @@ modularer Plugin-Architektur. Lauffähiger Kern mit Plugin-System, Authentifizierung/Rollen und Migrationen; vollständig implementiert sind bisher **Audit-Log**, **Benachrichtigung**, **Neuheiten (BGG-Sync)**, -**Dedup-Prüfung** und **Planungsliste**. +**Dedup-Prüfung**, **Planungsliste** und **Archiv (12-Monats-Autopilot)**. ## Stack @@ -54,6 +54,8 @@ Beim ersten Start wird automatisch ein Admin-Konto angelegt: | `SPIELE_BGG_SUCHBEGRIFFE` | `brettspiel` | Komma-getrennte Suchbegriffe für den regelmäßigen Sync | | `SPIELE_BGG_MAX_TREFFER_PRO_SUCHE` | `25` | Obergrenze Treffer je Suchbegriff (schont das BGG-Rate-Limit) | | `SPIELE_DEDUP_BGG_AKTIV` | `1` | BGG-Zusatzdaten für die Dedup-Prüfung an (`1`) oder aus (`0`): Alternate-Names und Erweiterungs-Relationen | +| `SPIELE_ARCHIV_JOB_AKTIV` | `1` | Täglicher Archivierungs-Job an (`1`) oder aus (`0`) | +| `SPIELE_ARCHIV_JOB_UHRZEIT` | `03:00` | Tageszeit des täglichen Archiv-Laufs im Format `HH:MM` | ## Tests @@ -75,6 +77,11 @@ Fake-BGG-Client, Prüfprotokoll, Prüfseite) und **planung** (beide Eintragswege mit Stub-gemockter dedup-/Benachrichtigungs-/Audit-API, Verlags-Auswahl- Dialog, Statuswechsel, Bearbeiten/Löschen, Rechte pro Rolle) plus ein Integrationstest mit den echten Plugins (In-App-Nachricht + Audit-Eintrag). +Dazu das **archiv**-Plugin: 12-Monats-Regel mit eingefrorener Uhr (Grenzfälle +„genau 12 Monate“, Schaltjahr/29. Februar, UTC-Konsistenz bei +Zeitzonen-Unterschieden), Job-Lauf mit Audit-Einträgen, Wiederherstellen +inkl. BGG-Konflikt und Ersatz-Rezensent, Rollen sowie Suche/Filter in der +Admin-Ansicht. ## Architektur @@ -94,7 +101,8 @@ plugins/ ein Ordner pro Funktion, ladbar über den Plugin-Load ├── benachrichtigung/ VOLL IMPLEMENTIERT: E-Mail/Telegram/In-App ├── dedup/ VOLL IMPLEMENTIERT: check_titel-API, Heuristik, Protokoll ├── planung/ VOLL IMPLEMENTIERT: Planungsliste, Verschiebung, Dialoge -├── archiv/ erinnerung/ export/ (Stubs, ladbar) +├── archiv/ VOLL IMPLEMENTIERT: 12-Monats-Autopilot, Job, UI +├── erinnerung/ export/ (Stubs, ladbar) └── … je __init__.py + templates// ``` @@ -122,7 +130,9 @@ class MeinPlugin(BasePlugin): return [NavEntry(label=self.title, url="/meinplugin")] def on_load(self, context): # Lifecycle-Hook beim Start - super().on_load(context) # context.engine/.session_factory/.templates/.settings + super().on_load(context) # context.engine/.session_factory/.templates/ + # .settings/.registry (Plugin-Registry für + # Hintergrund-Jobs ohne Request) def on_unload(self): # Lifecycle-Hook beim Herunterfahren super().on_unload() @@ -363,6 +373,64 @@ Rezensenten können **nur ihre eigenen Einträge** bearbeiten (Statuswechsel, Bearbeiten, Löschen); Admins und Redakteure alle. Jede Änderung wird im Audit-Log protokolliert (`erstellt`/`verschoben`/`geaendert`/`geloescht`). +## Plugin „archiv“ (implementiert) + +Automatische Archivierung nach dem **Sicherungsdatei-Prinzip**: Titel, deren +Erscheinungs- bzw. Eintragsdatum länger als 12 Monate zurückliegt, werden +vollständig in eigene Archiv-Tabellen verschoben und aus den aktiven Listen +entfernt. Nichts wird gelöscht oder verändert — jeder Eintrag lässt sich +unverändert zurück in die aktive Liste verschieben. + +### Datentabellen + +Eigene Migration `0001_archiv_tabellen` mit zwei Spiegel-Tabellen +(SQLite ↔ Postgres portabel, ohne Fremdschlüssel — archivierte Zeilen +überleben auch das Löschen eines Benutzers): + +| Tabelle | Spiegel von | Besondere Spalten | +|---------|-------------|-------------------| +| `archiv_neuheiten` | `neuheiten` | alle Originalspalten plus `quell_id`, `archiviert_am`, `archiviert_von`, `grund`, `referenz_am` | +| `archiv_planung` | `planungsliste` | alle Originalspalten (inkl. `pruefung`-JSON) plus dieselben Archiv-Metadaten | + +### Die 12-Monats-Regel + +- **Stichtag:** „jetzt − 12 Monate“ in UTC; archiviert wird, wer *streng + älter* ist — genau 12 Monate gilt noch nicht als „älter als“. +- **Referenzdatum:** + - Neuheiten: das Erscheinungsdatum. BGG liefert nur das Jahr, daher + großzügig das Jahresende (31.12., 23:59:59) — ein Titel von 2023 wird + erst ab dem 01.01.2025 archiviert. Ohne Jahr zählt das Eintragsdatum. + - Planungseinträge: das Eintragsdatum, unabhängig vom Status. +- **Zeitzonen:** verglichen wird durchgängig in UTC; naive Datenbank- + Zeitstempel gelten als UTC, zeitzonenbewusste Uhrzeiten werden nach UTC + verschoben (per Tests mit eingefrorener Uhr abgedeckt, inkl. 29. Februar). + +### Täglicher Hintergrund-Job + +Beim Start registriert das Plugin einen APScheduler-Cron-Job (Standard +täglich **03:00**, konfigurierbar über `SPIELE_ARCHIV_JOB_UHRZEIT`, +abschaltbar über `SPIELE_ARCHIV_JOB_AKTIV=0`). Pro archiviertem Titel +schreibt der Job einen **Audit-Log-Eintrag** mit dem Akteur „System“ +(Aktion `verschoben`, Grund und Referenzdatum in den Details). Der +Registry-Zugriff aus dem Hintergrund-Job läuft über den neuen +`context.registry`-Hook des Plugin-Kontexts. + +### Admin-Ansicht mit Wiederherstellen + +Unter **Archiv** (`/archiv`, nur Rolle **Admin**) werden beide Tabellen als +vereinigte Liste angezeigt (neueste Archivierung zuerst): + +- Suche über Titel/Verlag/Autor und Filter nach Herkunft (Neuheiten/Planung). +- Pro Eintrag: Herkunfts-Badge, Status, Rezensent, Archivierungszeitpunkt + und der Grund (z. B. „älter als 12 Monate (Erscheinungsjahr 2023)“). +- **Wiederherstellen** verschiebt den Eintrag vollständig zurück in die + aktive Liste und protokolliert dies im Audit-Log (Akteur: der Admin). + Konfliktfälle werden sicher abgelehnt: existiert in der Neuheitenliste + bereits ein aktiver Eintrag mit derselben BGG-ID (erneuter Sync), wird + nicht wiederhergestellt; existiert der zugeordnete Rezensent nicht mehr, + übernimmt der wiederherstellende Admin die Zuordnung (gemeldet und + auditiert). Rezensenten und Redakteure erhalten 403. + ## Deployment (später) Docker/Podman Compose ist vorgesehen (Henry-Lab, danach Kundenhardware). @@ -379,9 +447,9 @@ geschrieben. | 2 | Plugin `benachrichtigung` (E-Mail/Telegram/In-App, User-Präferenzen) | ✅ fertig | | 3 | Plugin `audit-log` (Wer/Was/Wann, Admin-Ansicht, öffentliche API) | ✅ fertig | | 4 | Plugin `neuheiten` (BGG-Sync, APScheduler, Filter) | ✅ fertig | -| 5 | Plugin `dedup` (Verlags-Konflikt, deutsche Version, Vorgänger-/Planungs-Check) | ✅ fertig (141 Tests grün) | -| 6 | Plugin `planung` (Verschiebung, händischer Eintrag + Prüfungen + Benachrichtigung) | ✅ fertig (141 Tests grün) | -| 7 | Plugin `archiv` (12-Monats-Autopilot) | ⏳ offen | +| 5 | Plugin `dedup` (Verlags-Konflikt, deutsche Version, Vorgänger-/Planungs-Check) | ✅ fertig | +| 6 | Plugin `planung` (Verschiebung, händischer Eintrag + Prüfungen + Benachrichtigung) | ✅ fertig | +| 7 | Plugin `archiv` (12-Monats-Autopilot, Wiederherstellen, täglicher Job) | ✅ fertig (171 Tests grün) | | 8 | Plugin `erinnerung` (Redaktionsschluss pro Ausgabe, 4-Wochen-Erinnerung) | ⏳ offen | | 9 | Plugin `export` (CSV + PDF via WeasyPrint) | ⏳ offen | | 10 | Integrationstests über alle Plugins | ⏳ offen | diff --git a/plugins/archiv/__init__.py b/plugins/archiv/__init__.py index 5b6bac5..1b55768 100644 --- a/plugins/archiv/__init__.py +++ b/plugins/archiv/__init__.py @@ -1,41 +1,278 @@ -"""Plugin „archiv“ — Platzhalter gemäß Plugin-Vertrag. +"""Plugin „archiv“ — automatische Archivierung nach dem Sicherungsdatei- +Prinzip. -Implementiert in einer späteren Phase. Der Stub zeigt den vollen Vertrag: -eigene Route, eigenes Template, Lifecycle-Hooks, Migrations-Schnittstelle. +Titel der Neuheitenliste und der Planungsliste, deren Erscheinungsdatum +bzw. Eintragsdatum länger als 12 Monate zurückliegt, werden vollständig in +die Archiv-Tabellen (`archiv_neuheiten`, `archiv_planung`) verschoben und +aus den aktiven Listen entfernt. Nichts wird gelöscht oder verändert — +Admins können jeden Eintrag über die Archiv-Ansicht jederzeit wieder in die +aktive Liste zurückverschieben. + +Umfang: +- Eigene Migration mit beiden Spiegel-Tabellen. +- Täglicher Hintergrund-Job (APScheduler, Cron um „03:00“ lokal). +- Admin-Ansicht `/archiv` mit Suche, Herkunftsfilter und Wiederherstellen. +- Je ein Audit-Log-Eintrag pro Archivierung (Akteur „System“) und pro + Wiederherstellung (Akteur der Aktion). + +Konfiguration über Umgebungsvariablen (gelesen beim Plugin-Start): +- SPIELE_ARCHIV_JOB_AKTIV „1“ (Standard) = täglicher Job an, „0“ = aus +- SPIELE_ARCHIV_JOB_UHRZEIT Tageszeit des Laufs im Format HH:MM (Standard 03:00) + +Die Fachlogik liegt im Service (`service.py`) mit eingefrierbarer Uhr; +Partner-Tabellen werden wie in dedup/planung nur über die gemeinsame +SQLAlchemy-Metadata angesprochen. Der Kern bleibt unberührt. """ from __future__ import annotations -from fastapi import Depends, Request +import logging +import os +from urllib.parse import quote -from redaktionskern.auth.deps import require_user -from redaktionskern.auth.models import User -from redaktionskern.contracts import BasePlugin, NavEntry +from fastapi import Depends, Request +from fastapi.responses import RedirectResponse +from sqlalchemy import select +from sqlalchemy.orm import Session + +from redaktionskern.auth.deps import get_db, require_roles +from redaktionskern.auth.models import Role, User +from redaktionskern.contracts import BasePlugin, Migration, NavEntry + +from .models import STATUS_ANZEIGE, ArchivNeuheit, ArchivPlanung +from .service import ( + GUELTIGE_QUELLEN, + QUELLE_NEUHEITEN, + OBJEKT_TYPEN, + ArchivService, +) + +_logger = logging.getLogger("plugins.archiv") + +JOB_ID = "archiv-taeglich" +STANDARD_UHRZEIT = "03:00" + + +def _uhrzeit_parsen(roh: str | None) -> tuple[int, int]: + """Parst „HH:MM“; bei Unsinn gilt der Standard 03:00.""" + teile = (roh or "").strip().split(":") + try: + stunde = int(teile[0]) + minute = int(teile[1]) if len(teile) > 1 else 0 + except (ValueError, IndexError): + return 3, 0 + if not (0 <= stunde <= 23 and 0 <= minute <= 59): + return 3, 0 + return stunde, minute + + +def _tabelle_anlegen(conn) -> None: + """Migration 0001: legt beide Spiegel-Tabellen an (portabel, idempotent).""" + ArchivNeuheit.__table__.create(conn, checkfirst=True) + ArchivPlanung.__table__.create(conn, checkfirst=True) class ArchivPlugin(BasePlugin): name = "archiv" title = "Archiv" - description = "Automatische Archivierung von Titeln älter als 12 Monate (Platzhalter)." + description = ( + "Verschiebt Titel älter als 12 Monate automatisch aus den aktiven " + "Listen in das Archiv — vollständig wiederherstellbar." + ) def __init__(self) -> None: super().__init__() + self._scheduler = None # BackgroundScheduler, falls aktiviert + #: Einfrierbare Uhr für Tests (Callable → aware datetime); None = echte Zeit. + self._jetzt = None + self._routen_registrieren() + + # ---------- Plugin-Vertrag ---------- + + def migrations(self) -> list[Migration]: + return [Migration(version="0001_archiv_tabellen", up=_tabelle_anlegen)] + + def navigation(self) -> list[NavEntry]: + return [NavEntry(label=self.title, url="/archiv")] + + def on_load(self, context) -> None: + super().on_load(context) + if os.environ.get("SPIELE_ARCHIV_JOB_AKTIV", "1").strip() == "1": + self._scheduler_starten( + os.environ.get("SPIELE_ARCHIV_JOB_UHRZEIT", STANDARD_UHRZEIT) + ) + + def on_unload(self) -> None: + if self._scheduler is not None: + self._scheduler.shutdown(wait=False) + self._scheduler = None + super().on_unload() + + # ---------- Internas ---------- + + def _neuer_service(self) -> ArchivService: + """Fabrik für den Service; die Uhr ist von Tests überschreibbar.""" + return ArchivService(self.context.session_factory, jetzt=self._jetzt) + + def _audit_plugin(self): + registry = getattr(self.context, "registry", None) + return registry.get("audit-log") if registry is not None else None + + def _auditiere_archivierung(self, ereignisse) -> int: + """Schreibt je archiviertem Titel einen Audit-Eintrag (Akteur: System).""" + audit = self._audit_plugin() + if audit is None: + return 0 + geschrieben = 0 + for ereignis in ereignisse: + try: + audit.log_sync( + None, + "verschoben", + ereignis.objekt_typ, + ereignis.objekt_id, + ereignis.details, + ) + geschrieben += 1 + except Exception: # noqa: BLE001 - Audit darf den Lauf nicht brechen + _logger.exception("Audit-Log-Eintrag für Archivierung fehlgeschlagen.") + return geschrieben + + def _archiv_job(self) -> None: + """Hintergrund-Job: läuft täglich und archiviert fällige Titel.""" + service = self._neuer_service() + lauf, ereignisse = service.lauf() + self._auditiere_archivierung(ereignisse) + if lauf.gesamt: + _logger.info("Täglicher Archiv-Lauf abgeschlossen: %s", lauf.als_text()) + + def _scheduler_starten(self, uhrzeit_roh: str | None) -> None: + from apscheduler.schedulers.background import BackgroundScheduler + + stunde, minute = _uhrzeit_parsen(uhrzeit_roh) + scheduler = BackgroundScheduler() + scheduler.add_job( + self._archiv_job, + trigger="cron", + hour=stunde, + minute=minute, + id=JOB_ID, + replace_existing=True, + ) + scheduler.start() + self._scheduler = scheduler + _logger.info( + "Automatische Archivierung aktiv: täglich um %02d:%02d Uhr.", + stunde, + minute, + ) + + # ---------- Routen ---------- + + def _routen_registrieren(self) -> None: @self.router.get("/archiv") - def seite(request: Request, user: User = Depends(require_user)): - """Platzhalterseite des Plugins.""" + def ansicht( + request: Request, + db: Session = Depends(get_db), + user: User = Depends(require_roles(Role.ADMIN.value)), + q: str = "", + quelle: str = "", + meldung: str = "", + fehler: str = "", + ): + """Archiv-Ansicht (nur Admin): Suche, Filter, Wiederherstellen.""" + service = self._neuer_service() + eintraege = service.ansicht(db) + + alle_neuheiten = sum(1 for e in eintraege if e["quelle"] == QUELLE_NEUHEITEN) + alle_planung = len(eintraege) - alle_neuheiten + + such = q.strip().lower() + if such: + eintraege = [ + e + for e in eintraege + if such in (e["titel"] or "").lower() + or such in (e["verlag"] or "").lower() + or such in (e["autor"] or "").lower() + ] + if quelle in GUELTIGE_QUELLEN: + eintraege = [e for e in eintraege if e["quelle"] == quelle] + + namen = { + benutzer.id: (benutzer.display_name or benutzer.username) + for benutzer in db.scalars(select(User)).all() + } + return self.context.templates.TemplateResponse( request=request, name="archiv/index.html", context={ "user": user, "titel": self.title, - "name": self.name, - "version": self.version, + "eintraege": eintraege, + "namen": namen, + "status_anzeige": STATUS_ANZEIGE, + "q": q, + "quelle_filter": quelle if quelle in GUELTIGE_QUELLEN else "", + "anzahl": len(eintraege), + "anzahl_neuheiten": alle_neuheiten, + "anzahl_planung": alle_planung, + "meldung": meldung[:300], + "fehler": fehler[:300], + "job_aktiv": self._scheduler is not None and self._scheduler.running, }, ) - def navigation(self) -> list[NavEntry]: - return [NavEntry(label=self.title, url="/archiv")] + @self.router.post("/archiv/{quelle}/{archiv_id}/wiederherstellen") + async def wiederherstellen( + request: Request, + quelle: str, + archiv_id: int, + db: Session = Depends(get_db), + user: User = Depends(require_roles(Role.ADMIN.value)), + ): + """Stellt einen archivierten Eintrag in die aktive Liste zurück.""" + ziel_url = "/archiv" + if quelle not in GUELTIGE_QUELLEN: + return RedirectResponse( + f"{ziel_url}?fehler={quote('Unbekannte Herkunft.')}", + status_code=303, + ) + service = self._neuer_service() + ergebnis = service.wiederherstellen( + db, quelle, archiv_id, ersatz_rezensent_id=user.id + ) + if not ergebnis.ok: + return RedirectResponse( + f"{ziel_url}?fehler={quote(ergebnis.meldung)}", status_code=303 + ) + + details = { + "titel": ergebnis.titel, + "von": "archiv", + "ziel": ergebnis.ziel, + "archiv_id": archiv_id, + } + if ergebnis.rezensent_ersetzt: + details["rezensent_ersetzt"] = True + audit = request.app.state.registry.get("audit-log") + if audit is not None: + try: + await audit.log( + user, + "verschoben", + OBJEKT_TYPEN[quelle], + ergebnis.neuer_id, + details, + ) + except Exception: # noqa: BLE001 - Audit blockiert nicht + _logger.exception( + "Audit-Log-Eintrag für Wiederherstellung fehlgeschlagen." + ) + return RedirectResponse( + f"{ziel_url}?meldung={quote(ergebnis.meldung)}", status_code=303 + ) plugin = ArchivPlugin() diff --git a/plugins/archiv/models.py b/plugins/archiv/models.py new file mode 100644 index 0000000..3186b29 --- /dev/null +++ b/plugins/archiv/models.py @@ -0,0 +1,108 @@ +"""Datenmodelle des Plugins „archiv“. + +Das Archiv folgt dem Sicherungsdatei-Prinzip: Beim Archivieren wird jede +Zeile vollständig in eine eigene Archiv-Tabelle kopiert und aus der aktiven +Liste entfernt — nichts wird verändert oder gelöscht, ein Eintrag lässt sich +jederzeit unverändert zurück in die aktive Liste verschieben. + +Es gibt zwei Spiegel-Tabellen (eine pro aktiver Liste): + +- `archiv_neuheiten` ← Spiegel von `neuheiten` +- `archiv_planung` ← Spiegel von `planungsliste` + +Bewusst ohne Fremdschlüssel: Archivierte Zeilen müssen auch dann erhalten +bleiben, wenn z. B. der zugeordnete Rezensent gelöscht wird. Bei der +Wiederherstellung wird die Gültigkeit geprüft und notfalls ein Ersatz- +Rezensent zugeordnet. + +Bewusst keine zusätzlichen Indizes: Bei doppelt geladenem Modul (direkter +Import im Test + Plugin-Loader-Namensraum) würden sie doppelt angelegt; die +Nutzerzahl ist klein. + +Die Archiv-Metadaten (wer/wann/warum) stehen in jeder Zeile: `quell_id` +(ursprüngliche ID in der aktiven Liste), `archiviert_am`, `archiviert_von`, +`grund` und `referenz_am` (Zeitpunkt, an dem die 12-Monats-Regel angesetzt +hat — Erscheinungstermin bzw. Eintragsdatum). +""" +from __future__ import annotations + +from datetime import datetime + +from sqlalchemy import DateTime, Integer, JSON, String, Text, func +from sqlalchemy.orm import Mapped, mapped_column + +from redaktionskern.db import Base + + +class _ArchivMeta: + """Gemeinsame Archiv-Metadaten beider Spiegel-Tabellen (Mixin).""" + + #: ID der Zeile in der ursprünglichen aktiven Liste. + quell_id: Mapped[int] = mapped_column(Integer) + #: Zeitpunkt der Archivierung (naive UTC, wie überall in der App). + archiviert_am: Mapped[datetime] = mapped_column(DateTime, server_default=func.now()) + #: „System“ beim automatischen Lauf, sonst Benutzername der Wiederherstellung… # noqa + archiviert_von: Mapped[str] = mapped_column(String(100), default="System") + #: Menschlich lesbarer Grund, z. B. „älter als 12 Monate (Erscheinungsjahr 2023)“. + grund: Mapped[str] = mapped_column(String(200), default="") + #: Datum, an dem die 12-Monats-Frist angesetzt hat (Erscheinung bzw. Eintrag). + referenz_am: Mapped[datetime | None] = mapped_column(DateTime) + + +class ArchivNeuheit(_ArchivMeta, Base): + """Archivierter Neuheiten-Eintrag (Spiegel der Tabelle `neuheiten`).""" + + __tablename__ = "archiv_neuheiten" + # extend_existing: Das Plugin kann doppelt geladen werden (direkter Import + # im Test + Plugin-Loader-Namensraum) und teilt sich dieselbe Metadata. + __table_args__ = {"extend_existing": True} + + id: Mapped[int] = mapped_column(primary_key=True) + titel: Mapped[str] = mapped_column(String(300)) + verlag: Mapped[str | None] = mapped_column(String(300)) + autor: Mapped[str | None] = mapped_column(String(300)) + erscheinungsjahr: Mapped[int | None] = mapped_column(Integer) + bgg_id: Mapped[int | None] = mapped_column(Integer) + status: Mapped[str] = mapped_column(String(50), default="neuheit") + quelle: Mapped[str] = mapped_column(String(100), default="boardgamegeek") + erstellt_am: Mapped[datetime | None] = mapped_column(DateTime) + aktualisiert_am: Mapped[datetime | None] = mapped_column(DateTime) + + def __repr__(self) -> str: # pragma: no cover - Debug-Hilfe + return f"" + + +class ArchivPlanung(_ArchivMeta, Base): + """Archivierter Planungseintrag (Spiegel der Tabelle `planungsliste`).""" + + __tablename__ = "archiv_planung" + __table_args__ = {"extend_existing": True} + + id: Mapped[int] = mapped_column(primary_key=True) + titel: Mapped[str] = mapped_column(String(300)) + verlag: Mapped[str | None] = mapped_column(String(300)) + autor: Mapped[str | None] = mapped_column(String(300)) + bgg_id: Mapped[int | None] = mapped_column(Integer) + ausgabe: Mapped[str] = mapped_column(String(100), default="") + #: Kein FK — Archiv-Zeilen überleben das Löschen des Benutzers. + rezensent_id: Mapped[int | None] = mapped_column(Integer) + status: Mapped[str] = mapped_column(String(50), default="offen") + notizen: Mapped[str | None] = mapped_column(Text) + quelle: Mapped[str] = mapped_column(String(50), default="manuell") + pruefung: Mapped[list | None] = mapped_column(JSON) + erstellt_am: Mapped[datetime | None] = mapped_column(DateTime) + aktualisiert_am: Mapped[datetime | None] = mapped_column(DateTime) + + def __repr__(self) -> str: # pragma: no cover - Debug-Hilfe + return f"" + + +#: Deutsche Anzeigenamen der Statuswerte (bewusst als Kopie ohne +#: Import-Abhängigkeit zu den Partner-Plugins, siehe planung/KONFLIKT_ANZEIGE). +STATUS_ANZEIGE: dict[str, str] = { + "neuheit": "Neuheit", + "planung": "In Planung", + "offen": "Offen", + "in_bearbeitung": "In Bearbeitung", + "abgeschlossen": "Abgeschlossen", +} diff --git a/plugins/archiv/service.py b/plugins/archiv/service.py new file mode 100644 index 0000000..2ecbb25 --- /dev/null +++ b/plugins/archiv/service.py @@ -0,0 +1,494 @@ +"""Archivierungs-Logik des Plugins „archiv“. + +Regeln (dokumentiert, von Tests mit eingefrorener Uhr abgedeckt): + +- **Stichtag:** „jetzt minus 12 Monate“ in UTC. Archiviert wird, wer + *streng älter* als der Stichtag ist — genau 12 Monate gilt noch nicht als + „älter als 12 Monate“. +- **Referenzdatum:** + - Neuheiten: das Erscheinungsdatum. BoardGameGeek liefert nur das Jahr, + daher großzügig das Ende des Erscheinungsjahres (31.12., 23:59:59) — ein + Titel des Jahres J wird also erst nach dem 31.12. von J+1 archiviert. + Fehlt das Jahr, zählt das Eintragsdatum (`erstellt_am`). + - Planungseinträge: das Eintragsdatum (`erstellt_am`), unabhängig vom Status. +- **Zeitzonen:** Verglichen wird durchgängig in UTC. Zeitstempel ohne + Zeitzone aus der Datenbank gelten als UTC (SQLite CURRENT_TIMESTAMP ist + UTC; für Postgres werden naive Werte equally behandelt). Bewohnte + Zeitzonen werden vor dem Vergleich nach UTC verschoben. +- **Sicherungsdatei-Prinzip:** Zeilen werden vollständig in die Spiegel- + Tabellen kopiert und aus den aktiven Listen entfernt — nichts geht verloren. +""" +from __future__ import annotations + +import calendar +import logging +from collections.abc import Callable +from dataclasses import dataclass, field +from datetime import datetime, timezone + +from sqlalchemy import delete, insert, select +from sqlalchemy.exc import SQLAlchemyError +from sqlalchemy.orm import Session, sessionmaker + +from redaktionskern.db import Base + +from .models import ArchivNeuheit, ArchivPlanung + +_logger = logging.getLogger("plugins.archiv") + +QUELLE_NEUHEITEN = "neuheiten" +QUELLE_PLANUNG = "planung" +GUELTIGE_QUELLEN: tuple[str, ...] = (QUELLE_NEUHEITEN, QUELLE_PLANUNG) + +#: Objekttypen für Audit-Log-Einträge (deckungsgleich mit den Partner-Plugins). +OBJEKT_TYPEN: dict[str, str] = { + QUELLE_NEUHEITEN: "neuheit", + QUELLE_PLANUNG: "planungseintrag", +} + +ARCHIV_ZIEL = "archiv" +GRUND_TEXT = "älter als 12 Monate" + +STANDARD_MONATE = 12 + + +def vor_monaten(moment: datetime, monate: int) -> datetime: + """Zieht `monate` Kalendermonate ab; Monatstage werden geklemmt + (z. B. 29. Februar − 12 Monate → 28. Februar des Vorjahres).""" + monat = moment.month - monate + jahr = moment.year + while monat <= 0: + monat += 12 + jahr -= 1 + tag = min(moment.day, calendar.monthrange(jahr, monat)[1]) + return moment.replace(year=jahr, month=monat, day=tag) + + +def als_utc(moment: datetime | None) -> datetime | None: + """Naive Zeitstempel gelten als UTC; zeitzonenbewusste → UTC.""" + if moment is None: + return None + if moment.tzinfo is None: + return moment.replace(tzinfo=timezone.utc) + return moment.astimezone(timezone.utc) + + +def naiv_utc(moment: datetime | None) -> datetime | None: + """Nach UTC normalisiert und ohne Zeitzone (Speicherformat der App).""" + utc = als_utc(moment) + return None if utc is None else utc.replace(tzinfo=None) + + +@dataclass +class ArchivLauf: + """Zusammenfassung eines automatischen Archivierungs-Laufs.""" + + neuheiten_archiviert: int = 0 + planung_archiviert: int = 0 + + @property + def gesamt(self) -> int: + return self.neuheiten_archiviert + self.planung_archiviert + + def als_text(self) -> str: + return ( + f"{self.neuheiten_archiviert} Neuheiten und " + f"{self.planung_archiviert} Planungseinträge archiviert" + ) + + +@dataclass +class ArchivEreignis: + """Ein archivierter Datensatz — Grundlage für je einen Audit-Log-Eintrag.""" + + objekt_typ: str + objekt_id: int + details: dict = field(default_factory=dict) + + +@dataclass +class Wiederherstellung: + """Ergebnis eines Wiederherstellen-Versuchs.""" + + ok: bool + meldung: str + ziel: str = "" + neuer_id: int | None = None + rezensent_ersetzt: bool = False + titel: str = "" + + +class ArchivService: + """Verschiebt veraltete Titel in die Archiv-Tabellen und zurück.""" + + def __init__( + self, + session_factory: sessionmaker, + jetzt: Callable[[], datetime] | None = None, + monate: int = STANDARD_MONATE, + ) -> None: + self.session_factory = session_factory + # Einfrierbare Uhr (Tests); Standard: aktuelle UTC-Zeit. + self._jetzt_fabrik = jetzt or (lambda: datetime.now(timezone.utc)) + self.monate = max(int(monate), 0) + + # ---------- Zeitlogik ---------- + + def jetzt(self) -> datetime: + return als_utc(self._jetzt_fabrik()) # type: ignore[arg-type] + + def stichtag(self) -> datetime: + """„jetzt − 12 Monate“; streng älter heißt archiviert.""" + return vor_monaten(self.jetzt(), self.monate) + + def referenz_neuheit(self, zeile) -> datetime: + """Erscheinungsdatum (Ende des bekannten Jahres) oder Eintragsdatum.""" + jahr = zeile["erscheinungsjahr"] + if jahr is not None: + return datetime(int(jahr), 12, 31, 23, 59, 59, tzinfo=timezone.utc) + erstellt = als_utc(zeile.get("erstellt_am")) + return erstellt or self.jetzt() + + def referenz_planung(self, zeile) -> datetime: + """Eintragsdatum des Planungseintrags.""" + erstellt = als_utc(zeile.get("erstellt_am")) + return erstellt or self.jetzt() + + def _grund_neuheit(self, zeile, referenz: datetime) -> str: + if zeile["erscheinungsjahr"] is not None: + return f"{GRUND_TEXT} (Erscheinungsjahr {zeile['erscheinungsjahr']})" + return f"{GRUND_TEXT} (eingetragen am {referenz:%d.%m.%Y})" + + def _grund_planung(self, zeile, referenz: datetime) -> str: + return f"{GRUND_TEXT} (eingetragen am {referenz:%d.%m.%Y})" + + # ---------- Automatischer Lauf ---------- + + def lauf(self) -> tuple[ArchivLauf, list[ArchivEreignis]]: + """Verschiebt alle fälligen Titel in die Archiv-Tabellen. + + Die Audit-Log-Einträge schreibt der Aufrufer (Route oder Hintergrund- + Job) über die öffentliche API des audit-log-Plugins — ein Ereignis + pro archiviertem Titel. + """ + stichtag = self.stichtag() + zeitpunkt = naiv_utc(self.jetzt()) + lauf = ArchivLauf() + ereignisse: list[ArchivEreignis] = [] + + with self.session_factory() as db: + db: Session + neuheiten = Base.metadata.tables.get("neuheiten") + if neuheiten is not None: + try: + zeilen = db.execute(select(neuheiten)).mappings().all() + except SQLAlchemyError as exc: + _logger.warning( + "archiv: Tabelle 'neuheiten' nicht lesbar (%s) — " + "Quelle übersprungen.", + exc, + ) + zeilen = [] + for zeile in zeilen: + referenz = self.referenz_neuheit(zeile) + if referenz >= stichtag: + continue # noch keine 12 Monate alt (oder jünger) + db.execute( + insert(ArchivNeuheit).values( + quell_id=zeile["id"], + titel=zeile["titel"], + verlag=zeile["verlag"], + autor=zeile["autor"], + erscheinungsjahr=zeile["erscheinungsjahr"], + bgg_id=zeile["bgg_id"], + status=zeile["status"], + quelle=zeile["quelle"], + erstellt_am=zeile["erstellt_am"], + aktualisiert_am=zeile["aktualisiert_am"], + archiviert_am=zeitpunkt, + referenz_am=naiv_utc(referenz), + grund=self._grund_neuheit(zeile, referenz), + ) + ) + db.execute( + delete(neuheiten).where(neuheiten.c.id == zeile["id"]) + ) + lauf.neuheiten_archiviert += 1 + ereignisse.append( + ArchivEreignis( + objekt_typ=OBJEKT_TYPEN[QUELLE_NEUHEITEN], + objekt_id=zeile["id"], + details={ + "titel": zeile["titel"], + "herkunft": QUELLE_NEUHEITEN, + "ziel": ARCHIV_ZIEL, + "grund": self._grund_neuheit(zeile, referenz), + "referenz": f"{naiv_utc(referenz):%Y-%m-%d}", + }, + ) + ) + + planungsliste = Base.metadata.tables.get("planungsliste") + if planungsliste is not None: + try: + zeilen = db.execute(select(planungsliste)).mappings().all() + except SQLAlchemyError as exc: + _logger.warning( + "archiv: Tabelle 'planungsliste' nicht lesbar (%s) — " + "Quelle übersprungen.", + exc, + ) + zeilen = [] + for zeile in zeilen: + referenz = self.referenz_planung(zeile) + if referenz >= stichtag: + continue + db.execute( + insert(ArchivPlanung).values( + quell_id=zeile["id"], + titel=zeile["titel"], + verlag=zeile["verlag"], + autor=zeile["autor"], + bgg_id=zeile["bgg_id"], + ausgabe=zeile["ausgabe"], + rezensent_id=zeile["rezensent_id"], + status=zeile["status"], + notizen=zeile["notizen"], + quelle=zeile["quelle"], + pruefung=zeile["pruefung"], + erstellt_am=zeile["erstellt_am"], + aktualisiert_am=zeile["aktualisiert_am"], + archiviert_am=zeitpunkt, + referenz_am=naiv_utc(referenz), + grund=self._grund_planung(zeile, referenz), + ) + ) + db.execute( + delete(planungsliste).where(planungsliste.c.id == zeile["id"]) + ) + lauf.planung_archiviert += 1 + ereignisse.append( + ArchivEreignis( + objekt_typ=OBJEKT_TYPEN[QUELLE_PLANUNG], + objekt_id=zeile["id"], + details={ + "titel": zeile["titel"], + "herkunft": QUELLE_PLANUNG, + "ziel": ARCHIV_ZIEL, + "grund": self._grund_planung(zeile, referenz), + "referenz": f"{naiv_utc(referenz):%Y-%m-%d}", + }, + ) + ) + + db.commit() + + if lauf.gesamt: + _logger.info("Automatische Archivierung: %s", lauf.als_text()) + return lauf, ereignisse + + # ---------- Ansicht ---------- + + def ansicht(self, db: Session) -> list[dict]: + """Vereinigte Archiv-Ansicht beider Tabellen, neueste zuerst.""" + eintraege: list[dict] = [] + for a in db.scalars( + select(ArchivNeuheit).order_by( + ArchivNeuheit.archiviert_am.desc(), ArchivNeuheit.id.desc() + ) + ): + eintraege.append( + { + "quelle": QUELLE_NEUHEITEN, + "id": a.id, + "titel": a.titel, + "verlag": a.verlag, + "autor": a.autor, + "bgg_id": a.bgg_id, + "jahr": a.erscheinungsjahr, + "ausgabe": "", + "status": a.status, + "rezensent_id": None, + "erstellt_am": a.erstellt_am, + "archiviert_am": a.archiviert_am, + "referenz_am": a.referenz_am, + "grund": a.grund, + } + ) + for a in db.scalars( + select(ArchivPlanung).order_by( + ArchivPlanung.archiviert_am.desc(), ArchivPlanung.id.desc() + ) + ): + eintraege.append( + { + "quelle": QUELLE_PLANUNG, + "id": a.id, + "titel": a.titel, + "verlag": a.verlag, + "autor": a.autor, + "bgg_id": a.bgg_id, + "jahr": None, + "ausgabe": a.ausgabe, + "status": a.status, + "rezensent_id": a.rezensent_id, + "erstellt_am": a.erstellt_am, + "archiviert_am": a.archiviert_am, + "referenz_am": a.referenz_am, + "grund": a.grund, + } + ) + # Neueste Archivierung zuerst; bei Gleichstand die höhere ID zuerst. + return sorted( + eintraege, + key=lambda e: (e["archiviert_am"] or datetime.min, e["id"]), + reverse=True, + ) + + # ---------- Wiederherstellen ---------- + + def wiederherstellen( + self, + db: Session, + quelle: str, + archiv_id: int, + *, + ersatz_rezensent_id: int | None = None, + jetzt: datetime | None = None, + ) -> Wiederherstellung: + """Schiebt einen archivierten Eintrag zurück in die aktive Liste. + + Konflikte werden abgelehnt (keine Datenänderung im Archiv): existiert + in der Neuheitenliste bereits ein aktiver Eintrag mit derselben BGG-ID, + wird nicht wiederhergestellt. Existiert der zugeordnete Rezensent + eines Planungseintrags nicht mehr, übernimmt `ersatz_rezensent_id` + die Zuordnung (wird gemeldet und auditiert). + """ + zeitpunkt = naiv_utc(jetzt or self.jetzt()) + + if quelle == QUELLE_NEUHEITEN: + return self._wiederherstelle_neuheit(db, archiv_id, zeitpunkt) + if quelle == QUELLE_PLANUNG: + return self._wiederherstelle_planung( + db, archiv_id, zeitpunkt, ersatz_rezensent_id + ) + return Wiederherstellung(False, f"Unbekannte Herkunft „{quelle}“.") + + def _wiederherstelle_neuheit( + self, db: Session, archiv_id: int, zeitpunkt: datetime + ) -> Wiederherstellung: + archiv = db.get(ArchivNeuheit, archiv_id) + if archiv is None: + return Wiederherstellung( + False, "Der archivierte Eintrag wurde nicht gefunden." + ) + aktiv = Base.metadata.tables.get("neuheiten") + if aktiv is None: + return Wiederherstellung( + False, + "Die Neuheitenliste ist nicht verfügbar — " + "Wiederherstellen nicht möglich.", + ) + if archiv.bgg_id is not None: + vorhanden = db.execute( + select(aktiv.c.id, aktiv.c.titel).where( + aktiv.c.bgg_id == archiv.bgg_id + ) + ).first() + if vorhanden is not None: + return Wiederherstellung( + False, + f"Es gibt bereits einen aktiven Neuheiten-Eintrag mit der " + f"BGG-ID {archiv.bgg_id} („{vorhanden.titel}“) — " + "Wiederherstellen abgebrochen.", + ) + ergebnis = db.execute( + insert(aktiv).values( + titel=archiv.titel, + verlag=archiv.verlag, + autor=archiv.autor, + erscheinungsjahr=archiv.erscheinungsjahr, + bgg_id=archiv.bgg_id, + status=archiv.status, + quelle=archiv.quelle, + erstellt_am=archiv.erstellt_am, + aktualisiert_am=zeitpunkt, + ) + ) + neuer_id = ergebnis.inserted_primary_key[0] + titel = archiv.titel + db.delete(archiv) + db.commit() + return Wiederherstellung( + True, + f"„{titel}“ wurde zurück in die Neuheitenliste verschoben.", + ziel=QUELLE_NEUHEITEN, + neuer_id=int(neuer_id), + titel=titel, + ) + + def _wiederherstelle_planung( + self, + db: Session, + archiv_id: int, + zeitpunkt: datetime, + ersatz_rezensent_id: int | None, + ) -> Wiederherstellung: + archiv = db.get(ArchivPlanung, archiv_id) + if archiv is None: + return Wiederherstellung( + False, "Der archivierte Eintrag wurde nicht gefunden." + ) + aktiv = Base.metadata.tables.get("planungsliste") + if aktiv is None: + return Wiederherstellung( + False, + "Die Planungsliste ist nicht verfügbar — " + "Wiederherstellen nicht möglich.", + ) + + rezensent_id = archiv.rezensent_id + rezensent_ersetzt = False + benutzer = Base.metadata.tables.get("users") + if rezensent_id is not None and benutzer is not None: + existiert = db.execute( + select(benutzer.c.id).where(benutzer.c.id == rezensent_id) + ).first() + if existiert is None and ersatz_rezensent_id is not None: + rezensent_id = ersatz_rezensent_id + rezensent_ersetzt = True + + ergebnis = db.execute( + insert(aktiv).values( + titel=archiv.titel, + verlag=archiv.verlag, + autor=archiv.autor, + bgg_id=archiv.bgg_id, + ausgabe=archiv.ausgabe, + rezensent_id=rezensent_id, + status=archiv.status, + notizen=archiv.notizen, + quelle=archiv.quelle, + pruefung=archiv.pruefung, + erstellt_am=archiv.erstellt_am, + aktualisiert_am=zeitpunkt, + ) + ) + neuer_id = ergebnis.inserted_primary_key[0] + titel = archiv.titel + db.delete(archiv) + db.commit() + meldung = f"„{titel}“ wurde zurück in die Planungsliste verschoben." + if rezensent_ersetzt: + meldung += ( + " Der ursprüngliche Rezensent existierte nicht mehr — " + "der Eintrag wurde dir zugeordnet." + ) + return Wiederherstellung( + True, + meldung, + ziel=QUELLE_PLANUNG, + neuer_id=int(neuer_id), + rezensent_ersetzt=rezensent_ersetzt, + titel=titel, + ) diff --git a/plugins/archiv/templates/archiv/index.html b/plugins/archiv/templates/archiv/index.html index cc09663..48215e6 100644 --- a/plugins/archiv/templates/archiv/index.html +++ b/plugins/archiv/templates/archiv/index.html @@ -2,9 +2,120 @@ {% block titel %}{{ titel }} — Spiele-Redaktion{% endblock %} {% block inhalt %}

{{ titel }}

-

- Plugin {{ name }} - in Version {{ version }} ist geladen. - Diese Seite ist ein Platzhalter — die Funktion wird in einer späteren Phase implementiert. +

+ Sicherungsdatei der Redaktion: Titel, deren Erscheinungs- bzw. Eintragsdatum + mehr als 12 Monate zurückliegt, werden täglich automatisch aus den + aktiven Listen hierher verschoben. Es geht nichts verloren — jeder + Eintrag lässt sich unverändert zurück in die aktive Liste verschieben.

+ +
+ + + Täglicher Archiv-Job: {{ 'aktiv' if job_aktiv else 'aus' }} + + {{ anzahl_neuheiten }} archivierte Neuheiten + {{ anzahl_planung }} archivierte Planungseinträge +
+ +{% if meldung %} +
{{ meldung }}
+{% endif %} +{% if fehler %} +
{{ fehler }}
+{% endif %} + +
+
+
+ + +
+
+ + +
+ + {% if q or quelle_filter %} + Zurücksetzen + {% endif %} +
+
+ +

{{ anzahl }} Einträge im Archiv

+ +
+ + + + + + + + + + + + + + + + {% for eintrag in eintraege %} + + + + + + + + + + + + {% else %} + + + + {% endfor %} + +
SpieltitelHerkunftVerlagJahr / AusgabeStatusRezensentArchiviert amGrundAktionen
+ {{ eintrag.titel }} + {% if eintrag.bgg_id %} + BGG {{ eintrag.bgg_id }} + {% endif %} + + {% if eintrag.quelle == 'neuheiten' %} + Neuheiten + {% else %} + Planung + {% endif %} + {{ eintrag.verlag or '—' }}{{ eintrag.jahr if eintrag.jahr else (eintrag.ausgabe or '—') }}{{ status_anzeige.get(eintrag.status, eintrag.status) }}{{ namen.get(eintrag.rezensent_id, '—') if eintrag.rezensent_id else '—' }} + {{ eintrag.archiviert_am.strftime('%d.%m.%Y %H:%M') if eintrag.archiviert_am else '—' }} + {{ eintrag.grund }} +
+ +
+
+ {% if q or quelle_filter %} + Keine Treffer für diese Suche. + {% else %} + Das Archiv ist leer — nichts zu sehen. + {% endif %} +
+
{% endblock %} diff --git a/src/redaktionskern/app.py b/src/redaktionskern/app.py index b09aeab..810d06a 100644 --- a/src/redaktionskern/app.py +++ b/src/redaktionskern/app.py @@ -82,6 +82,7 @@ def create_app(settings: Settings | None = None) -> FastAPI: session_factory=session_factory, settings=settings, templates=templates, + registry=registry, ) with engine.begin() as conn: diff --git a/src/redaktionskern/contracts.py b/src/redaktionskern/contracts.py index 35ee402..5f6b5db 100644 --- a/src/redaktionskern/contracts.py +++ b/src/redaktionskern/contracts.py @@ -22,6 +22,7 @@ if TYPE_CHECKING: from sqlalchemy.orm import sessionmaker from redaktionskern.config import Settings + from redaktionskern.plugin_loader import PluginRegistry @dataclass(frozen=True) @@ -49,11 +50,16 @@ class PluginContext: session_factory: "sessionmaker", settings: "Settings", templates: "Jinja2Templates", + registry: "PluginRegistry | None" = None, ) -> None: self.engine = engine self.session_factory = session_factory self.settings = settings self.templates = templates + #: Registry aller geladenen Plugins — damit auch Hintergrund-Jobs + #: eines Plugins die öffentliche API anderer Plugins erreichen + #: (in Anfragen geht das üblicherweise über request.app.state.registry). + self.registry = registry class BasePlugin: diff --git a/tests/conftest.py b/tests/conftest.py index b728807..15e2960 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -22,6 +22,8 @@ def _hintergrundjobs_deaktiviert(monkeypatch): monkeypatch.setenv("SPIELE_BGG_SYNC_AKTIV", "1"). """ monkeypatch.setenv("SPIELE_BGG_SYNC_AKTIV", "0") + # Auch der tägliche Archiv-Job bleibt in Tests aus. + monkeypatch.setenv("SPIELE_ARCHIV_JOB_AKTIV", "0") # Auch der BGG-Hilfsclient des dedup-Plugins bleibt in Tests offline. monkeypatch.setenv("SPIELE_DEDUP_BGG_AKTIV", "0") diff --git a/tests/test_archiv.py b/tests/test_archiv.py new file mode 100644 index 0000000..a7fcda2 --- /dev/null +++ b/tests/test_archiv.py @@ -0,0 +1,675 @@ +"""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 diff --git a/tests/test_plugin_loader.py b/tests/test_plugin_loader.py index 12ebd3f..6666e1e 100644 --- a/tests/test_plugin_loader.py +++ b/tests/test_plugin_loader.py @@ -41,7 +41,7 @@ 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", "archiv"} PLATZHALTER_STUBS = ERWARTETE_STUBS - VOLL_IMPLEMENTIERT