From fb62c28e100fc89b10c7ad6de65741cfb4823831 Mon Sep 17 00:00:00 2001
From: Colin Maudry <colin@maudry.com>
Date: Fri, 3 Jul 2026 09:38:33 +0200
Subject: [PATCH] feat(admin): add admin_actions audit table and log/list
 functions

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
---
 src/admin/__init__.py   |  0
 src/admin/db.py         | 26 +++++++++++++++++++++++++
 src/migrations.py       | 10 ++++++++++
 tests/admin/__init__.py |  0
 tests/admin/conftest.py | 12 ++++++++++++
 tests/admin/test_db.py  | 43 +++++++++++++++++++++++++++++++++++++++++
 6 files changed, 91 insertions(+)
 create mode 100644 src/admin/__init__.py
 create mode 100644 src/admin/db.py
 create mode 100644 tests/admin/__init__.py
 create mode 100644 tests/admin/conftest.py
 create mode 100644 tests/admin/test_db.py

diff --git a/src/admin/__init__.py b/src/admin/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/src/admin/db.py b/src/admin/db.py
new file mode 100644
index 0000000..b4531c3
--- /dev/null
+++ b/src/admin/db.py
@@ -0,0 +1,26 @@
+import sqlite3
+from datetime import datetime, timezone
+
+from src.auth.db import get_conn
+
+
+def _now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def log_action(
+    admin_email: str, action: str, target_user_id: int | None, details: str | None
+) -> None:
+    get_conn().execute(
+        "INSERT INTO admin_actions (admin_email, action, target_user_id, details, created_at) "
+        "VALUES (?, ?, ?, ?, ?)",
+        (admin_email, action, target_user_id, details, _now()),
+    )
+
+
+def list_actions(limit: int = 200) -> list[sqlite3.Row]:
+    return (
+        get_conn()
+        .execute("SELECT * FROM admin_actions ORDER BY id DESC LIMIT ?", (limit,))
+        .fetchall()
+    )
diff --git a/src/migrations.py b/src/migrations.py
index 05cb933..dd44b6a 100644
--- a/src/migrations.py
+++ b/src/migrations.py
@@ -32,6 +32,16 @@ _MIGRATIONS: list[tuple[str, str]] = [
         "0005_rename_votes_credited_until_to_votes_last_credited_at",
         "ALTER TABLE subscriptions RENAME COLUMN votes_credited_until TO votes_last_credited_at",
     ),
+    (
+        "0006_create_admin_actions",
+        "CREATE TABLE IF NOT EXISTS admin_actions ("
+        "id INTEGER PRIMARY KEY AUTOINCREMENT, "
+        "admin_email TEXT NOT NULL, "
+        "action TEXT NOT NULL, "
+        "target_user_id INTEGER, "
+        "details TEXT, "
+        "created_at TEXT NOT NULL)",
+    ),
 ]
 
 
diff --git a/tests/admin/__init__.py b/tests/admin/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/admin/conftest.py b/tests/admin/conftest.py
new file mode 100644
index 0000000..ce77c3b
--- /dev/null
+++ b/tests/admin/conftest.py
@@ -0,0 +1,12 @@
+import pytest
+
+
+@pytest.fixture
+def users_db_path(monkeypatch, tmp_path):
+    from src.auth.db import reset_conn_for_tests
+
+    db_path = tmp_path / "users.test.sqlite"
+    monkeypatch.setenv("USERS_DB_PATH", str(db_path))
+    reset_conn_for_tests()
+    yield db_path
+    reset_conn_for_tests()
diff --git a/tests/admin/test_db.py b/tests/admin/test_db.py
new file mode 100644
index 0000000..8e6052d
--- /dev/null
+++ b/tests/admin/test_db.py
@@ -0,0 +1,43 @@
+from src.admin import db as admin_db
+from src.auth.db import init_schema
+from src.migrations import apply_pending
+from src.subscriptions.db import init_schema as init_subscriptions_schema
+
+
+def _setup():
+    init_schema()
+    init_subscriptions_schema()
+    apply_pending()
+
+
+def test_log_action_then_list_actions_returns_it(users_db_path):
+    _setup()
+
+    admin_db.log_action("admin@ex.fr", "subscription_status_change", 42, "active → cancelled")
+
+    rows = admin_db.list_actions()
+    assert len(rows) == 1
+    assert rows[0]["admin_email"] == "admin@ex.fr"
+    assert rows[0]["action"] == "subscription_status_change"
+    assert rows[0]["target_user_id"] == 42
+    assert rows[0]["details"] == "active → cancelled"
+
+
+def test_list_actions_most_recent_first(users_db_path):
+    _setup()
+
+    admin_db.log_action("admin@ex.fr", "action_one", None, None)
+    admin_db.log_action("admin@ex.fr", "action_two", None, None)
+
+    rows = admin_db.list_actions()
+    assert [r["action"] for r in rows] == ["action_two", "action_one"]
+
+
+def test_list_actions_respects_limit(users_db_path):
+    _setup()
+
+    for i in range(3):
+        admin_db.log_action("admin@ex.fr", f"action_{i}", None, None)
+
+    rows = admin_db.list_actions(limit=2)
+    assert len(rows) == 2