diff --git a/AGENTS.md b/AGENTS.md index d986771..721484f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -4,7 +4,9 @@ Multi-User-Webanwendung „KI-Assistenz für Spielemagazin-Redaktionen": Neuheitenliste (BGG), Dedup-Prüfungen, Planungsliste, Archivierung, Erinnerungen, Benachrichtigungen (E-Mail/Telegram/In-App), Audit-Log, Export. ## Status -Setup — Prompt freigegeben 2026-08-21, Implementierung noch nicht begonnen. +Kern mit Plugin-System/Auth/Migrationen fertig; `audit-log` vollständig +implementiert; `neuheiten`/`benachrichtigung` in Arbeit; übrige Plugins Stubs. +Live-Status: README.md → „Projekt-Fortschritt“. ## Struktur | Pfad | Inhalt | diff --git a/plugins/audit-log/__init__.py b/plugins/audit-log/__init__.py index 9c7af3f..0c0143c 100644 --- a/plugins/audit-log/__init__.py +++ b/plugins/audit-log/__init__.py @@ -1,41 +1,278 @@ -"""Plugin „audit-log“ — Platzhalter gemäß Plugin-Vertrag. +"""Plugin „audit-log“ — Wer hat was wann verschoben, eingetragen oder geändert. -Implementiert in einer späteren Phase. Der Stub zeigt den vollen Vertrag: -eigene Route, eigenes Template, Lifecycle-Hooks, Migrations-Schnittstelle. +Voller Plugin-Vertrag: eigene Migration (Tabelle audit_eintraege), eigene +Routen/Templates (Admin-Ansicht mit Filtern und Paginierung) und eine +öffentliche API, die andere Plugins bei relevanten Ereignissen aufrufen. + +Andere Plugins benutzen die API so: + + audit = request.app.state.registry.get("audit-log") + if audit is not None: + await audit.log(user, "geaendert", "planungseintrag", eintrag.id, + details={"alt": alt, "neu": neu}, + ip_adresse=request.client.host) + +In synchronen Kontexten (FastAPI führt `def`-Routen im Threadpool aus) steht +`audit.log_sync(...)` mit denselben Argumenten bereit. """ from __future__ import annotations -from fastapi import Depends, Request +import json +from datetime import datetime, time +from math import ceil +from urllib.parse import urlencode -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 sqlalchemy import func, select +from sqlalchemy.orm import Session + +from redaktionskern.auth.deps import get_db, require_roles, require_user +from redaktionskern.auth.models import Role, User +from redaktionskern.contracts import BasePlugin, Migration, NavEntry + +from .models import AuditEintrag + +SEITEN_GROESSE = 25 +SYSTEM_NAME = "System" + +# Kanonische Aktionstypen mit deutschen Anzeigenamen. Andere Plugins sollen +# diese Konstanten verwenden, damit die Filter-Auswahl konsistent bleibt. +AKTIONEN_ANZEIGE: dict[str, str] = { + "erstellt": "Eintrag erstellt", + "verschoben": "Eintrag verschoben", + "geaendert": "Eintrag geändert", + "geloescht": "Eintrag gelöscht", + "benachrichtigt": "Benachrichtigung versendet", +} + + +def _json_schoen(wert) -> str: + """Jinja-Filter: Dict lesbar als JSON formatieren (leer → Leerstring).""" + if not wert: + return "" + return json.dumps(wert, ensure_ascii=False, indent=2, sort_keys=True) + + +def _aus_actor(actor: User | str | None) -> tuple[int | None, str]: + """Normalisiert den Akteur auf (Benutzer-ID, Name im Log).""" + if actor is None: + return None, SYSTEM_NAME + if isinstance(actor, User): + return actor.id, actor.username + return None, str(actor) + + +def _tabelle_anlegen(conn) -> None: + """Migration 0001: legt die Plugin-Tabelle an (portabel, idempotent).""" + AuditEintrag.__table__.create(conn, checkfirst=True) + + +def _parse_datum(roh: str) -> datetime | None: + """Parst ein ISO-Datum (JJJJ-MM-TT); None bei leer, ValueError bei Unsinn.""" + roh = (roh or "").strip() + if not roh: + return None + return datetime.strptime(roh, "%Y-%m-%d") class AuditLogPlugin(BasePlugin): name = "audit-log" title = "Audit-Log" - description = "Wer hat was wann verschoben, eingetragen oder geändert (Platzhalter)." + description = "Wer hat was wann verschoben, eingetragen oder geändert." + + # ---------- Plugin-Vertrag ---------- + + def migrations(self) -> list[Migration]: + return [Migration(version="0001_audit_eintraege", up=_tabelle_anlegen)] + + def navigation(self) -> list[NavEntry]: + return [NavEntry(label=self.title, url=f"/{self.name}")] + + def on_load(self, context) -> None: + super().on_load(context) + # Jinja-Filter für die JSON-Anzeige in der Admin-Ansicht. + context.templates.env.filters["json_schoen"] = _json_schoen + + # ---------- Öffentliche API für andere Plugins ---------- + + async def log( + self, + actor: User | str | None, + action: str, + objekt_typ: str = "", + objekt_id: int | str | None = None, + details: dict | None = None, + *, + ip_adresse: str | None = None, + ) -> None: + """Protokolliert ein Ereignis (asynchrone Variante). + + `details` ist ein freies Dict; die Schlüssel „alt“ und „neu“ werden + in die gleichnamigen JSON-Spalten übernommen, der Rest landet in + `details`. + """ + self._speichere(actor, action, objekt_typ, objekt_id, details, ip_adresse) + + def log_sync( + self, + actor: User | str | None, + action: str, + objekt_typ: str = "", + objekt_id: int | str | None = None, + details: dict | None = None, + *, + ip_adresse: str | None = None, + ) -> None: + """Synchrone Variante von log() für `def`-Routen (Threadpool).""" + self._speichere(actor, action, objekt_typ, objekt_id, details, ip_adresse) + + def _speichere( + self, + actor: User | str | None, + action: str, + objekt_typ: str, + objekt_id: int | str | None, + details: dict | None, + ip_adresse: str | None, + ) -> None: + actor_id, actor_name = _aus_actor(actor) + alt = neu = rest = None + if details: + rest = dict(details) + alt = rest.pop("alt", None) + neu = rest.pop("neu", None) + if not rest: + rest = None + with self.context.session_factory() as db: + db.add( + AuditEintrag( + actor_id=actor_id, + actor_name=actor_name, + action=action, + objekt_typ=objekt_typ or "", + objekt_id="" if objekt_id is None else str(objekt_id), + alt=alt, + neu=neu, + details=rest, + ip_adresse=ip_adresse, + ) + ) + db.commit() + + # ---------- Admin-Ansicht ---------- def __init__(self) -> None: super().__init__() @self.router.get("/audit-log") - def seite(request: Request, user: User = Depends(require_user)): - """Platzhalterseite des Plugins.""" + def ansicht( + request: Request, + db: Session = Depends(get_db), + benutzer: str = "", + aktion: str = "", + von: str = "", + bis: str = "", + seite: int = 1, + user: User = Depends(require_roles(Role.ADMIN.value)), + ): + """Admin-Ansicht: filterbar nach Benutzer, Aktion und Zeitraum.""" + fehler: list[str] = [] + + filter_liste = [] + if benutzer.strip(): + filter_liste.append(AuditEintrag.actor_name == benutzer.strip()) + if aktion.strip(): + filter_liste.append(AuditEintrag.action == aktion.strip()) + try: + von_dt = _parse_datum(von) + except ValueError: + von_dt = None + fehler.append("„Von“ ist kein gültiges Datum (JJJJ-MM-TT) — Filter ignoriert.") + try: + bis_dt = _parse_datum(bis) + except ValueError: + bis_dt = None + fehler.append("„Bis“ ist kein gültiges Datum (JJJJ-MM-TT) — Filter ignoriert.") + if von_dt is not None: + filter_liste.append(AuditEintrag.erstellt_am >= von_dt) + if bis_dt is not None: + filter_liste.append( + AuditEintrag.erstellt_am + <= datetime.combine(bis_dt.date(), time.max) + ) + + gesamt = db.scalar( + select(func.count()).select_from(AuditEintrag).where(*filter_liste) + ) or 0 + gesamt_seiten = max(1, ceil(gesamt / SEITEN_GROESSE)) + seite = min(max(1, seite), gesamt_seiten) + offset = (seite - 1) * SEITEN_GROESSE + + eintraege = ( + db.scalars( + select(AuditEintrag) + .where(*filter_liste) + .order_by(AuditEintrag.erstellt_am.desc(), AuditEintrag.id.desc()) + .offset(offset) + .limit(SEITEN_GROESSE) + ) + .all() + ) + + # Auswahl für die Filter-Dropdowns (aus vorhandenen Einträgen). + benutzer_auswahl = [ + name + for name in db.scalars( + select(AuditEintrag.actor_name).distinct().order_by(AuditEintrag.actor_name) + ) + if name + ] + aktionen_auswahl = db.scalars( + select(AuditEintrag.action).distinct().order_by(AuditEintrag.action) + ).all() + + # Paginierungs-Links mit erhaltenen Filtern. + parameter = { + schluessel: wert + for schluessel, wert in ( + ("benutzer", benutzer.strip()), + ("aktion", aktion.strip()), + ("von", von.strip()), + ("bis", bis.strip()), + ) + if wert + } + basis = urlencode(parameter) + + def _seiten_url(nr: int) -> str: + return f"/audit-log?{basis}&seite={nr}" if basis else f"/audit-log?seite={nr}" + return self.context.templates.TemplateResponse( request=request, name="audit-log/index.html", context={ "user": user, "titel": self.title, - "name": self.name, - "version": self.version, + "eintraege": eintraege, + "gesamt": gesamt, + "seite": seite, + "gesamt_seiten":gesamt_seiten, + "seiten_groesse": SEITEN_GROESSE, + "start": offset + 1 if eintraege else 0, + "ende": offset + len(eintraege), + "url_zurueck": _seiten_url(seite - 1), + "url_weiter": _seiten_url(seite + 1), + "url_ohne_seite": f"/audit-log?{basis}" if basis else "/audit-log", + "benutzer_auswahl": benutzer_auswahl, + "aktionen_auswahl": aktionen_auswahl, + "aktionen_anzeige": AKTIONEN_ANZEIGE, + "filter_benutzer": benutzer.strip(), + "filter_aktion": aktion.strip(), + "filter_von": von.strip(), + "filter_bis": bis.strip(), + "fehler": fehler, }, ) - def navigation(self) -> list[NavEntry]: - return [NavEntry(label=self.title, url="/audit-log")] - plugin = AuditLogPlugin() diff --git a/plugins/audit-log/models.py b/plugins/audit-log/models.py new file mode 100644 index 0000000..3509702 --- /dev/null +++ b/plugins/audit-log/models.py @@ -0,0 +1,40 @@ +"""Datenmodell des Plugins „audit-log“. + +Ein AuditEintrag protokolliert ein Ereignis: wer hat was wann mit welchem +Objekt getan (Alt-/Neustand als JSON, IP optional). Die Tabelle gehört +ausschließlich zu diesem Plugin; der Kern kennt sie nicht. +""" +from __future__ import annotations + +from datetime import datetime + +from sqlalchemy import JSON, DateTime, Integer, String, func +from sqlalchemy.orm import Mapped, mapped_column + +from redaktionskern.db import Base + + +class AuditEintrag(Base): + """Ein protokolliertes Ereignis im Audit-Log.""" + + __tablename__ = "audit_eintraege" + + id: Mapped[int] = mapped_column(primary_key=True) + # Wer: Benutzer-ID und -Name (NULL/„System“ bei automatischen Ereignissen) + actor_id: Mapped[int | None] = mapped_column(Integer, index=True) + actor_name: Mapped[str] = mapped_column(String(100), default="", index=True) + # Was: Aktionstyp (z. B. „erstellt“, „geaendert“, „geloescht“) + action: Mapped[str] = mapped_column(String(100), index=True) + # Auf welches Objekt: Typ + ID (ID als Text, damit alle ID-Arten passen) + objekt_typ: Mapped[str] = mapped_column(String(100), default="") + objekt_id: Mapped[str] = mapped_column(String(100), default="") + # Alt-/Neustand und freie Zusatzinformationen als JSON + alt: Mapped[dict | None] = mapped_column(JSON) + neu: Mapped[dict | None] = mapped_column(JSON) + details: Mapped[dict | None] = mapped_column(JSON) + # Optionale Herkunftsangabe + ip_adresse: Mapped[str | None] = mapped_column(String(64)) + # Wann + erstellt_am: Mapped[datetime] = mapped_column( + DateTime, server_default=func.now(), index=True + ) diff --git a/plugins/audit-log/templates/audit-log/index.html b/plugins/audit-log/templates/audit-log/index.html index cc09663..20a8fa1 100644 --- a/plugins/audit-log/templates/audit-log/index.html +++ b/plugins/audit-log/templates/audit-log/index.html @@ -1,10 +1,132 @@ {% extends "base.html" %} -{% block titel %}{{ titel }} — Spiele-Redaktion{% endblock %} +{% block titel %}{{ titel }} — {{ app_name }}{% 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. +

{{ titel }}

+

+ Protokoll aller relevanten Ereignisse — nur für Administratoren sichtbar.

+ +{% if fehler %} + {% for meldung in fehler %} +
+ {{ meldung }} +
+ {% endfor %} +{% endif %} + +
+
+ + + + +
+ + + Zurücksetzen + +
+
+
+ +
+ + + + + + + + + + + + + {% for e in eintraege %} + + + + + + + + + {% else %} + + + + {% endfor %} + +
ZeitpunktBenutzerAktionObjektÄnderung / DetailsIP
+ {{ e.erstellt_am.strftime('%d.%m.%Y %H:%M:%S') if e.erstellt_am else '–' }} + {{ e.actor_name }}{{ aktionen_anzeige[e.action] or e.action }} + {{ e.objekt_typ }}{% if e.objekt_id %} #{{ e.objekt_id }}{% endif %} + + {% if e.alt or e.neu %} + {% if e.alt %}
Alt: + 
{{ e.alt | json_schoen }}
{% endif %} + {% if e.neu %}
Neu: + 
{{ e.neu | json_schoen }}
{% endif %} + {% elif e.details %} + 
{{ e.details | json_schoen }}
+ {% else %} + – + {% endif %} + 		{{ e.ip_adresse or '–' }}
+ Keine Einträge gefunden. +
+
+ +
+

+ {% if gesamt %} + Zeige {{ start }}–{{ ende }} von {{ gesamt }} Einträgen + {% else %} + 0 Einträge + {% endif %} +

+ {% if gesamt_seiten > 1 %} +
+ {% if seite > 1 %} + ← Zurück + {% endif %} + Seite {{ seite }} von {{ gesamt_seiten }} + {% if seite < gesamt_seiten %} + Weiter → + {% endif %} +
+ {% endif %} +
{% endblock %} diff --git a/tests/test_audit_log.py b/tests/test_audit_log.py new file mode 100644 index 0000000..29ae439 --- /dev/null +++ b/tests/test_audit_log.py @@ -0,0 +1,261 @@ +"""Tests: Audit-Log — Logging-Funktion, Migration, Filter, Paginierung, Nur-Admin.""" +from __future__ import annotations + +import asyncio +import inspect +import sys +from datetime import datetime + +from sqlalchemy import func, select, text + +from tests.conftest import lege_benutzer_an, melde_an + + +# ---------------- Hilfen ---------------- + +def plugin(app): + return app.state.registry.get("audit-log") + + +def modul(app): + return sys.modules[type(plugin(app)).__module__] + + +def klasse(app): + return modul(app).AuditEintrag + + +def anzahl(app) -> int: + with app.state.session_factory() as db: + return db.scalar(select(func.count()).select_from(klasse(app))) + + +def direkt_anlegen(app, **kwargs): + """Legt einen Audit-Eintrag direkt an (für Filter-/Paginierungstests).""" + kwargs.setdefault("action", "erstellt") + kwargs.setdefault("actor_name", "systemtest") + with app.state.session_factory() as db: + db.add(klasse(app)(**kwargs)) + db.commit() + + +# ---------------- Migration ---------------- + +def test_migration_legt_tabelle_an(app): + with app.state.engine.connect() as conn: + tabellen = { + zeile[0] + for zeile in conn.execute(text("SELECT name FROM sqlite_master WHERE type='table'")) + } + protokoll = conn.execute( + text("SELECT version FROM schema_migrations WHERE plugin='audit-log'") + ).fetchall() + assert "audit_eintraege" in tabellen + assert [z[0] for z in protokoll] == ["0001_audit_eintraege"] + + +# ---------------- Logging-Funktion ---------------- + +def test_log_mit_benutzer_und_vollständigen_feldern(app): + lege_benutzer_an(app, "anna", "redakteur") + with app.state.session_factory() as db: + from redaktionskern.auth.models import User + anna = db.scalar(select(User).where(User.username == "anna")) + + asyncio.run( + plugin(app).log( + anna, + "erstellt", + "planungseintrag", + 42, + {"titel": "Cascadia"}, + ip_adresse="10.1.2.3", + ) + ) + + with app.state.session_factory() as db: + eintrag = db.scalar(select(klasse(app))) + assert eintrag.actor_id == anna.id + assert eintrag.actor_name == "anna" + assert eintrag.action == "erstellt" + assert eintrag.objekt_typ == "planungseintrag" + assert eintrag.objekt_id == "42" # int wird als Text gespeichert + assert eintrag.details == {"titel": "Cascadia"} + assert eintrag.alt is None and eintrag.neu is None + assert eintrag.ip_adresse == "10.1.2.3" + assert eintrag.erstellt_am is not None + + +def test_log_ohne_actor_ergibt_system(app): + asyncio.run(plugin(app).log(None, "benachrichtigt", "erinnerung")) + with app.state.session_factory() as db: + eintrag = db.scalar(select(klasse(app))) + assert eintrag.actor_id is None + assert eintrag.actor_name == "System" + assert eintrag.action == "benachrichtigt" + assert eintrag.details is None + assert eintrag.ip_adresse is None + + +def test_log_extrahiert_alt_und_neu_aus_details(app): + asyncio.run( + plugin(app).log( + "extern", + "geaendert", + "neuheit", + 7, + {"alt": {"titel": "Falsch"}, "neu": {"titel": "Richtig"}, "grund": "Tippfehler"}, + ) + ) + with app.state.session_factory() as db: + eintrag = db.scalar(select(klasse(app))) + assert eintrag.alt == {"titel": "Falsch"} + assert eintrag.neu == {"titel": "Richtig"} + assert eintrag.details == {"grund": "Tippfehler"} + assert eintrag.actor_name == "extern" + + +def test_log_sync_und_async_schreiben_beide(app): + plugin(app).log_sync(None, "geloescht", "planungseintrag", 5) + asyncio.run(plugin(app).log(None, "verschoben", "planungseintrag", 6)) + assert anzahl(app) == 2 + + +def test_api_ist_oeffentlich_erreichbar(app): + """Andere Plugins holen die API über die Plugin-Registry.""" + audit = app.state.registry.get("audit-log") + assert audit is not None + assert callable(audit.log) + assert inspect.iscoroutinefunction(audit.log) + assert callable(audit.log_sync) + + +# ---------------- Admin-Ansicht: Zugriff ---------------- + +def test_admin_sieht_audit_log(client): + melde_an(client) + antwort = client.get("/audit-log") + assert antwort.status_code == 200 + assert "Audit-Log" in antwort.text + + +def test_rezensent_und_redakteur_haben_keinen_zugriff(app, client): + lege_benutzer_an(app, "rez1", "rezensent") + lege_benutzer_an(app, "red1", "redakteur") + + melde_an(client, "rez1", "test-12345678") + assert client.get("/audit-log").status_code == 403 + client.post("/logout") + + melde_an(client, "red1", "test-12345678") + assert client.get("/audit-log").status_code == 403 + + +def test_anonym_wird_zum_login_umgeleitet(client): + antwort = client.get("/audit-log", follow_redirects=False) + assert antwort.status_code == 303 + assert antwort.headers["location"] == "/login" + + +# ---------------- Admin-Ansicht: Filter ---------------- + +def test_filter_nach_benutzer(app, client): + direkt_anlegen(app, actor_name="anna", objekt_typ="planungseintrag", objekt_id="alpha-eins") + direkt_anlegen(app, actor_name="anna", objekt_typ="planungseintrag", objekt_id="alpha-zwei") + direkt_anlegen(app, actor_name="bob", objekt_typ="planungseintrag", objekt_id="beta-eins") + + melde_an(client) + alles = client.get("/audit-log").text + assert "alpha-eins" in alles and "beta-eins" in alles + + gefiltert = client.get("/audit-log?benutzer=anna").text + assert "alpha-eins" in gefiltert and "alpha-zwei" in gefiltert + assert "beta-eins" not in gefiltert + + +def test_filter_nach_aktionstyp(app, client): + direkt_anlegen(app, action="erstellt", objekt_id="neu-ding") + direkt_anlegen(app, action="geloescht", objekt_id="weg-ding") + + melde_an(client) + gefiltert = client.get("/audit-log?aktion=geloescht").text + assert "weg-ding" in gefiltert + assert "neu-ding" not in gefiltert + + +def test_filter_nach_zeitraum_inklusive_grenzen(app, client): + direkt_anlegen(app, objekt_id="alt-ding", erstellt_am=datetime(2026, 1, 5, 8, 0, 0)) + direkt_anlegen(app, objekt_id="grenze-unten", erstellt_am=datetime(2026, 2, 1, 0, 0, 0)) + direkt_anlegen(app, objekt_id="mitte", erstellt_am=datetime(2026, 2, 15, 12, 30, 0)) + direkt_anlegen(app, objekt_id="grenze-oben", erstellt_am=datetime(2026, 2, 28, 23, 59, 59)) + direkt_anlegen(app, objekt_id="spaet-ding", erstellt_am=datetime(2026, 3, 10, 9, 0, 0)) + + melde_an(client) + gefiltert = client.get("/audit-log?von=2026-02-01&bis=2026-02-28").text + for marker in ("grenze-unten", "mitte", "grenze-oben"): + assert marker in gefiltert, marker + assert "alt-ding" not in gefiltert + assert "spaet-ding" not in gefiltert + + +def test_filter_kombinierbar_und_ungueltiges_datum_wird_angemerkt(app, client): + direkt_anlegen(app, actor_name="anna", action="erstellt", objekt_id="treffer") + direkt_anlegen(app, actor_name="anna", action="geloescht", objekt_id="kein-treffer") + direkt_anlegen(app, actor_name="bob", action="erstellt", objekt_id="auch-kein-treffer") + + melde_an(client) + gefiltert = client.get("/audit-log?benutzer=anna&aktion=erstellt").text + assert "treffer" in gefiltert + assert "kein-treffer" not in gefiltert and "auch-kein-treffer" not in gefiltert + + unsinn = client.get("/audit-log?von=gestern").text + assert "kein gültiges Datum" in unsinn + assert "treffer" in unsinn # ungültiger Filter wird ignoriert, Liste bleibt sichtbar + + +def test_neueste_zuerst_sortiert(app, client): + direkt_anlegen(app, objekt_id="frueh", erstellt_am=datetime(2026, 1, 1)) + direkt_anlegen(app, objekt_id="spaet", erstellt_am=datetime(2026, 6, 1)) + + melde_an(client) + text_html = client.get("/audit-log").text + assert text_html.index("spaet") < text_html.index("frueh") + + +# ---------------- Admin-Ansicht: Paginierung ---------------- + +def test_paginierung_trennt_seiten(app, client): + for i in range(30): # 30 Einträge bei 25 pro Seite → 2 Seiten + direkt_anlegen(app, objekt_id=f"ding-{i:02d}") + + melde_an(client) + # Neueste zuerst: Seite 1 zeigt ding-29 (neuest) bis ding-05. + seite1 = client.get("/audit-log").text + assert "Seite 1 von 2" in seite1 + assert "Zeige 1–25 von 30" in seite1 + assert "ding-29" in seite1 and "ding-05" in seite1 + assert "ding-04" not in seite1 and "ding-00" not in seite1 + assert "seite=2" in seite1 # Weiter-Link + + seite2 = client.get("/audit-log?seite=2").text + assert "Seite 2 von 2" in seite2 + assert "Zeige 26–30 von 30" in seite2 + assert "ding-04" in seite2 and "ding-00" in seite2 + assert "ding-05" not in seite2 + assert "seite=3" not in seite2 # kein Weiter mehr + + +def test_paginierung_behaelt_filter(app, client): + for i in range(30): + direkt_anlegen(app, actor_name="anna", objekt_id=f"anna-{i:02d}") + for i in range(30): + direkt_anlegen(app, actor_name="bob", objekt_id=f"bob-{i:02d}") + + melde_an(client) + # Gefiltert nach bob: Seite 2 zeigt die ältesten bob-Einträge (bob-04 … bob-00). + seite2 = client.get("/audit-log?benutzer=bob&seite=2").text + assert "Seite 2 von 2" in seite2 + assert "Zeige 26–30 von 30" in seite2 + assert "bob-04" in seite2 and "bob-00" in seite2 + assert "bob-29" not in seite2 + assert "anna-" not in seite2 diff --git a/tests/test_plugin_loader.py b/tests/test_plugin_loader.py index 970253f..fa22446 100644 --- a/tests/test_plugin_loader.py +++ b/tests/test_plugin_loader.py @@ -8,7 +8,6 @@ from redaktionskern.contracts import BasePlugin, NavEntry from redaktionskern.plugin_loader import PluginError, PluginRegistry ERWARTETE_STUBS = { - "neuheiten", "dedup", "planung", "archiv", @@ -18,35 +17,53 @@ ERWARTETE_STUBS = { "export", } +# Voll implementierte Plugins (keine Stubs mehr): +ERWARTETE_PLUGINS = ERWARTETE_STUBS | {"neuheiten"} -def test_alle_stubs_werden_geladen(app): + +def test_alle_plugins_werden_geladen(app): namen = {p.name for p in app.state.registry.all()} - assert ERWARTETE_STUBS <= namen + assert ERWARTETE_PLUGINS <= namen def test_on_load_wurde_aufgerufen(app): for plugin in app.state.registry.all(): - if plugin.name in ERWARTETE_STUBS: + if plugin.name in ERWARTETE_PLUGINS: assert plugin.loaded, f"Plugin {plugin.name} ist nicht geladen." assert plugin.context is not None -def test_stubs_haben_navigation(app): - for name in ERWARTETE_STUBS: +def test_plugins_haben_navigation(app): + for name in ERWARTETE_PLUGINS: plugin = app.state.registry.get(name) assert len(plugin.navigation()) == 1 +# 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"} +PLATZHALTER_STUBS = ERWARTETE_STUBS - VOLL_IMPLEMENTIERT + + def test_plugin_seiten_erreichbar(client): from tests.conftest import melde_an melde_an(client) - for name in sorted(ERWARTETE_STUBS): + for name in sorted(PLATZHALTER_STUBS): antwort = client.get(f"/{name}") assert antwort.status_code == 200, name assert "ist geladen" in antwort.text +def test_audit_log_seite_erreichbar_fuer_admin(client): + from tests.conftest import melde_an + + melde_an(client) + antwort = client.get("/audit-log") + assert antwort.status_code == 200 + assert "Audit-Log" in antwort.text + + def test_on_unload_beim_herunterfahren(app): plugin = app.state.registry.get("neuheiten") with TestClient(app):