Plugin audit-log: Protokolltabelle, öffentliche API und Admin-Ansicht

- Eigene Migration 0001_audit_eintraege (Tabelle audit_eintraege: Akteur, Aktion, Objekttyp/-ID, Alt/Neu + Details als JSON, optionale IP, Zeitstempel) - Öffentliche Plugin-API: await log(actor, action, objekt_typ, objekt_id, details, ip_adresse) bzw. log_sync() für synchrone Routen; andere Plugins holen die API über request.app.state.registry.get("audit-log") - Admin-Ansicht /audit-log (nur Admin): Filter nach Benutzer, Aktionstyp und Zeitraum, Paginierung (25/Seite), deutsche UI, Alt/Neu-JSON-Darstellung - Tests: Logging-Funktion (Actor-Varianten, alt/neu-Extraktion), Migration, Filter-Querys, Paginierung, Nur-Admin-Zugriff (13 neue Tests) - Loader-Test angepasst: voll implementierte Plugins tragen keinen Platzhalter-Text mehr - AGENTS.md: Status aktualisiert Verifiziert: HEAD + diese Änderungen = 58 Tests grün (uv run pytest)
2026-08-21 19:02:50 +00:00
parent f51732bff1
commit 83d574d95c
6 changed files with 708 additions and 29 deletions
--- a/tests/test_audit_log.py
+++ 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&ndash;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&ndash;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&ndash;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
--- 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):