feat(admin): add admin_actions audit table and log/list functions
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
@@ -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()
|
||||
)
|
||||
@@ -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)",
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
|
||||
@@ -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()
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user