diff --git a/docs/superpowers/plans/2026-07-03-admin-ui-plan.md b/docs/superpowers/plans/2026-07-03-admin-ui-plan.md new file mode 100644 index 0000000..37b7a06 --- /dev/null +++ b/docs/superpowers/plans/2026-07-03-admin-ui-plan.md @@ -0,0 +1,1362 @@ +# Panneau admin interne (`/admin`) Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Build an internal admin panel at `/admin` for support/debugging — list users, view a user's full subscription history, correct a subscription status manually, all restricted to a single `ADMIN_EMAIL` and logged to an `admin_actions` audit table also viewable at `/admin/journal`. + +**Architecture:** Dash pages (`register_page`) for reads, one Flask blueprint (`src/admin/routes.py`) for the single write action, both guarded by a shared `is_admin()` check. Mirrors the existing `/compte/*` (Dash) + `/auth/*` (Flask blueprint) split. + +**Tech Stack:** Dash 3.4, dash-bootstrap-components, Flask, flask-login, flask-wtf (CSRFProtect, already global), sqlite3 (raw, no ORM). + +**Spec:** `docs/superpowers/specs/2026-07-03-admin-ui-design.md` + +## Global Constraints + +- Valid subscription statuses (from `src/subscriptions/webhooks.py:map_subscription`): `active`, `trial`, `cancelled`, `expired`, `pending`. +- `ADMIN_EMAIL` is a single email, compared case-insensitively. No multi-admin support in this lot. +- Access guard (`is_admin()` false) always returns/aborts **404**, never a redirect — applies uniformly to anonymous and authenticated-non-admin. +- CSRF protection is already global (`CSRFProtect(app)`, `src/auth/setup.py:53`) — no custom CSRF code, just the existing hidden `csrf_token` input pattern. +- `dash_table.DataTable` uses `filter_action="native"`, `sort_action="native"`, `page_action="native"`, `page_size=20` everywhere in this feature (no custom/server-side filtering). +- Run tests with `uv run pytest` (venv activation via the Bash tool doesn't reliably update `PATH` in this environment). +- `sqlite3` connections here are autocommit (`isolation_level=None`, see `src/auth/db.py:56`) — new DB functions must NOT call `conn.commit()`, matching every existing function in `src/auth/db.py` and `src/subscriptions/db.py`. + +--- + +### Task 1: `list_users()` in the auth DB layer + +**Files:** + +- Modify: `src/auth/db.py` (add after `get_user_by_id`, around line 148) +- Test: `tests/auth/test_db.py` + +**Interfaces:** + +- Produces: `list_users(limit: int = 1000) -> list[sqlite3.Row]` — all users, most recently created first, capped at `limit`. + +- [ ] **Step 1: Write the failing test** + +Add to `tests/auth/test_db.py`: + +```python +def test_list_users_orders_by_created_at_desc(users_db_path): + from src.auth import db + + db.init_schema() + db.create_user("first@ex.fr", "hash") + db.create_user("second@ex.fr", "hash") + + rows = db.list_users() + + assert [r["email"] for r in rows] == ["second@ex.fr", "first@ex.fr"] + + +def test_list_users_respects_limit(users_db_path): + from src.auth import db + + db.init_schema() + for i in range(3): + db.create_user(f"user{i}@ex.fr", "hash") + + rows = db.list_users(limit=2) + + assert len(rows) == 2 +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `uv run pytest tests/auth/test_db.py -k list_users -v` +Expected: FAIL with `AttributeError: module 'src.auth.db' has no attribute 'list_users'` + +- [ ] **Step 3: Implement `list_users`** + +In `src/auth/db.py`, immediately after `get_user_by_id` (line 148): + +```python +def list_users(limit: int = 1000) -> list[sqlite3.Row]: + return ( + get_conn() + .execute("SELECT * FROM users ORDER BY created_at DESC LIMIT ?", (limit,)) + .fetchall() + ) +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `uv run pytest tests/auth/test_db.py -k list_users -v` +Expected: 2 passed + +- [ ] **Step 5: Commit** + +```bash +git add src/auth/db.py tests/auth/test_db.py +git commit -m "feat(admin): add list_users() to auth DB layer" +``` + +--- + +### Task 2: Subscription DB additions — history, status override, statuses constant + +**Files:** + +- Modify: `src/subscriptions/db.py` (add after `get_current`, around line 124, and after `_get_state`, around line 210) +- Test: `tests/subscriptions/test_db.py` + +**Interfaces:** + +- Produces: `SUBSCRIPTION_STATUSES: tuple[str, ...]` = `("active", "trial", "cancelled", "expired", "pending")` +- Produces: `list_by_user(user_id: int) -> list[sqlite3.Row]` — all subscriptions for a user, most recent (`id DESC`) first. +- Produces: `set_status(subscription_id: int, status: str) -> None` — overwrites `status` and `updated_at`. +- Produces: `get_subscriber_state(user_id: int) -> sqlite3.Row | None` — public wrapper around the existing private `_get_state`. +- Consumes: `get_conn()` from `src.auth.db` (already imported), `_now()` (already defined in this module), `create_pending(user_id, customer_handle, plan, prix_ht=None) -> (handle, subscription_id)` (already exists, used by the test). + +- [ ] **Step 1: Write the failing tests** + +Add to `tests/subscriptions/test_db.py`: + +```python +def test_list_by_user_returns_most_recent_first(users_db_path): + uid = _make_user() + db.init_schema() + _handle1, sub_id1 = db.create_pending(uid, "cust-1", "simple") + _handle2, sub_id2 = db.create_pending(uid, "cust-1", "soutien") + + rows = db.list_by_user(uid) + + assert [r["id"] for r in rows] == [sub_id2, sub_id1] + + +def test_set_status_updates_status(users_db_path): + uid = _make_user() + db.init_schema() + _handle, sub_id = db.create_pending(uid, "cust-1", "simple") + + db.set_status(sub_id, "active") + + row = db.get_current(uid) + assert row["status"] == "active" + + +def test_get_subscriber_state_returns_row_after_create_pending(users_db_path): + uid = _make_user() + db.init_schema() + db.create_pending(uid, "cust-1", "simple") + + state = db.get_subscriber_state(uid) + + assert state is not None + assert state["user_id"] == uid + + +def test_get_subscriber_state_returns_none_for_unknown_user(users_db_path): + db.init_schema() + assert db.get_subscriber_state(999999) is None + + +def test_subscription_statuses_constant(): + assert db.SUBSCRIPTION_STATUSES == ( + "active", + "trial", + "cancelled", + "expired", + "pending", + ) +``` + +(`_make_user` and the `users_db_path` fixture already exist in this file/its `conftest.py` — see `tests/subscriptions/test_db.py:16`.) + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `uv run pytest tests/subscriptions/test_db.py -k "list_by_user or set_status or get_subscriber_state or subscription_statuses" -v` +Expected: FAIL with `AttributeError` for each missing symbol + +- [ ] **Step 3: Implement the additions** + +In `src/subscriptions/db.py`, immediately after `get_current` (after line 123, before `get_by_handle`): + +```python +SUBSCRIPTION_STATUSES = ("active", "trial", "cancelled", "expired", "pending") + + +def list_by_user(user_id: int) -> list[sqlite3.Row]: + return ( + get_conn() + .execute( + "SELECT * FROM subscriptions WHERE user_id = ? ORDER BY id DESC", + (user_id,), + ) + .fetchall() + ) + + +def set_status(subscription_id: int, status: str) -> None: + get_conn().execute( + "UPDATE subscriptions SET status = ?, updated_at = ? WHERE id = ?", + (status, _now(), subscription_id), + ) +``` + +Then, immediately after `_get_state` (after line 210, before `freeze_votes_cursor`): + +```python +def get_subscriber_state(user_id: int) -> sqlite3.Row | None: + return _get_state(user_id) +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `uv run pytest tests/subscriptions/test_db.py -k "list_by_user or set_status or get_subscriber_state or subscription_statuses" -v` +Expected: 5 passed + +- [ ] **Step 5: Commit** + +```bash +git add src/subscriptions/db.py tests/subscriptions/test_db.py +git commit -m "feat(admin): add subscription history, status override, and statuses constant" +``` + +--- + +### Task 3: Audit log table + `src/admin/db.py` + +**Files:** + +- Modify: `src/migrations.py` (append to `_MIGRATIONS`) +- Create: `src/admin/__init__.py` (empty) +- Create: `src/admin/db.py` +- Create: `tests/admin/__init__.py` (empty) +- Create: `tests/admin/conftest.py` +- Create: `tests/admin/test_db.py` + +**Interfaces:** + +- Produces: table `admin_actions(id, admin_email, action, target_user_id, details, created_at)`. +- Produces: `log_action(admin_email: str, action: str, target_user_id: int | None, details: str | None) -> None` +- Produces: `list_actions(limit: int = 200) -> list[sqlite3.Row]` — most recent (`id DESC`) first. +- Consumes: `get_conn()` from `src.auth.db`, `apply_pending()` from `src.migrations`. + +- [ ] **Step 1: Add the migration** + +In `src/migrations.py`, append to `_MIGRATIONS` (after the `0005_...` entry): + +```python + ( + "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)", + ), +``` + +- [ ] **Step 2: Create the test conftest** + +Create `tests/admin/__init__.py` (empty file). + +Create `tests/admin/conftest.py`: + +```python +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() +``` + +- [ ] **Step 3: Write the failing test** + +Create `tests/admin/test_db.py`: + +```python +from src.admin import db as admin_db +from src.auth.db import init_schema +from src.migrations import apply_pending + + +def _setup(): + init_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 +``` + +- [ ] **Step 4: Run test to verify it fails** + +Run: `uv run pytest tests/admin/test_db.py -v` +Expected: FAIL with `ModuleNotFoundError: No module named 'src.admin'` + +- [ ] **Step 5: Implement `src/admin/db.py`** + +Create `src/admin/__init__.py` (empty file). + +Create `src/admin/db.py`: + +```python +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() + ) +``` + +- [ ] **Step 6: Run tests to verify they pass** + +Run: `uv run pytest tests/admin/test_db.py -v` +Expected: 3 passed + +- [ ] **Step 7: Commit** + +```bash +git add src/migrations.py src/admin/__init__.py src/admin/db.py tests/admin/__init__.py tests/admin/conftest.py tests/admin/test_db.py +git commit -m "feat(admin): add admin_actions audit table and log/list functions" +``` + +--- + +### Task 4: Access guard — `is_admin()` + +**Files:** + +- Create: `src/admin/guard.py` +- Create: `tests/admin/test_guard.py` + +**Interfaces:** + +- Produces: `is_admin() -> bool` — `True` only if `ADMIN_EMAIL` is set, `current_user.is_authenticated`, and `current_user.email` matches case-insensitively. + +- [ ] **Step 1: Write the failing tests** + +Create `tests/admin/test_guard.py`: + +```python +from unittest.mock import patch + +from src.admin import guard + + +def _fake_user(email: str, authenticated: bool = True): + user = type("U", (), {})() + user.is_authenticated = authenticated + user.email = email + return user + + +def test_is_admin_true_for_matching_email(monkeypatch): + monkeypatch.setenv("ADMIN_EMAIL", "admin@ex.fr") + with patch("src.admin.guard.current_user", _fake_user("admin@ex.fr")): + assert guard.is_admin() is True + + +def test_is_admin_true_case_insensitive(monkeypatch): + monkeypatch.setenv("ADMIN_EMAIL", "Admin@Ex.fr") + with patch("src.admin.guard.current_user", _fake_user("admin@ex.fr")): + assert guard.is_admin() is True + + +def test_is_admin_false_for_different_email(monkeypatch): + monkeypatch.setenv("ADMIN_EMAIL", "admin@ex.fr") + with patch("src.admin.guard.current_user", _fake_user("someone@ex.fr")): + assert guard.is_admin() is False + + +def test_is_admin_false_for_anonymous(monkeypatch): + monkeypatch.setenv("ADMIN_EMAIL", "admin@ex.fr") + with patch( + "src.admin.guard.current_user", _fake_user("admin@ex.fr", authenticated=False) + ): + assert guard.is_admin() is False + + +def test_is_admin_false_when_admin_email_unset(monkeypatch): + monkeypatch.delenv("ADMIN_EMAIL", raising=False) + with patch("src.admin.guard.current_user", _fake_user("admin@ex.fr")): + assert guard.is_admin() is False +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `uv run pytest tests/admin/test_guard.py -v` +Expected: FAIL with `ModuleNotFoundError: No module named 'src.admin.guard'` + +- [ ] **Step 3: Implement `src/admin/guard.py`** + +```python +import os + +from flask_login import current_user + + +def is_admin() -> bool: + admin_email = os.getenv("ADMIN_EMAIL") + return bool( + admin_email + and current_user.is_authenticated + and current_user.email.lower() == admin_email.lower() + ) +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `uv run pytest tests/admin/test_guard.py -v` +Expected: 5 passed + +- [ ] **Step 5: Commit** + +```bash +git add src/admin/guard.py tests/admin/test_guard.py +git commit -m "feat(admin): add is_admin() access guard" +``` + +--- + +### Task 5: Mutation route — `POST /admin/actions/subscription-status` + +**Files:** + +- Create: `src/admin/routes.py` +- Modify: `src/auth/setup.py` (register blueprint) +- Modify: `.template.env` (add `ADMIN_EMAIL`) +- Modify: `tests/admin/conftest.py` (add Flask app/client fixtures) +- Create: `tests/admin/test_routes.py` + +**Interfaces:** + +- Produces: Flask blueprint `admin_bp` (`url_prefix="/admin/actions"`), route `POST /subscription-status`. +- Consumes: `is_admin()` (Task 4), `SUBSCRIPTION_STATUSES`, `get_current`, `set_status` (Task 2), `log_action` (Task 3). + +- [ ] **Step 1: Extend the test conftest** + +In `tests/admin/conftest.py`, add (after the existing `users_db_path` fixture): + +```python +@pytest.fixture +def admin_app(users_db_path, monkeypatch): + from flask import Flask + + from src.auth.setup import init_auth + from src.subscriptions.setup import init_subscriptions + + monkeypatch.setenv("SECRET_KEY", "test-secret-key") + monkeypatch.setenv("APP_BASE_URL", "http://localhost:8050") + monkeypatch.setenv("FRISBII_PLAN_SIMPLE", "plan_simple") + monkeypatch.setenv("FRISBII_PLAN_SOUTIEN", "plan_soutien") + monkeypatch.setenv("FRISBII_WEBHOOK_SECRET", "s3cr3t") + flask_app = Flask(__name__) + flask_app.config["WTF_CSRF_ENABLED"] = False + init_auth(flask_app) + init_subscriptions(flask_app) + return flask_app + + +@pytest.fixture +def admin_client(admin_app): + return admin_app.test_client() + + +@pytest.fixture +def logged_in_admin_client(admin_app, monkeypatch): + from src.auth import db as auth_db + + monkeypatch.setenv("ADMIN_EMAIL", "admin@ex.fr") + uid = auth_db.create_user("admin@ex.fr", "hash") + auth_db.set_email_verified(uid) + client = admin_app.test_client() + with client.session_transaction() as sess: + sess["_user_id"] = str(uid) + sess["_fresh"] = True + return client, uid +``` + +This mirrors `tests/subscriptions/conftest.py:62-106` (`sub_app`/`logged_in_client`). + +- [ ] **Step 2: Write the failing tests** + +Create `tests/admin/test_routes.py`: + +```python +from src.auth import db as auth_db +from src.subscriptions import db as sub_db + + +def _make_target_with_subscription(): + uid = auth_db.create_user("target@ex.fr", "hash") + _handle, sub_id = sub_db.create_pending(uid, "cust-1", "simple") + return uid, sub_id + + +def test_subscription_status_requires_admin(admin_client): + resp = admin_client.post( + "/admin/actions/subscription-status", + data={"user_id": "1", "subscription_id": "1", "status": "active"}, + ) + assert resp.status_code == 404 + + +def test_subscription_status_rejects_invalid_status(logged_in_admin_client): + client, _admin_uid = logged_in_admin_client + uid, sub_id = _make_target_with_subscription() + + resp = client.post( + "/admin/actions/subscription-status", + data={"user_id": str(uid), "subscription_id": str(sub_id), "status": "bogus"}, + ) + + assert resp.status_code == 302 + assert resp.headers["Location"] == f"/admin/user/{uid}?error=invalid_status" + assert sub_db.get_current(uid)["status"] == "pending" + + +def test_subscription_status_rejects_mismatched_subscription(logged_in_admin_client): + client, _admin_uid = logged_in_admin_client + uid, _sub_id = _make_target_with_subscription() + other_uid, other_sub_id = _make_target_with_subscription() + + resp = client.post( + "/admin/actions/subscription-status", + data={ + "user_id": str(uid), + "subscription_id": str(other_sub_id), + "status": "active", + }, + ) + + assert resp.status_code == 302 + assert resp.headers["Location"] == f"/admin/user/{uid}?error=invalid_status" + assert sub_db.get_current(other_uid)["status"] == "pending" + + +def test_subscription_status_success_updates_and_logs(logged_in_admin_client): + from src.admin.db import list_actions + + client, _admin_uid = logged_in_admin_client + uid, sub_id = _make_target_with_subscription() + + resp = client.post( + "/admin/actions/subscription-status", + data={"user_id": str(uid), "subscription_id": str(sub_id), "status": "active"}, + ) + + assert resp.status_code == 302 + assert resp.headers["Location"] == f"/admin/user/{uid}?status_changed=1" + assert sub_db.get_current(uid)["status"] == "active" + + actions = list_actions() + assert len(actions) == 1 + assert actions[0]["action"] == "subscription_status_change" + assert actions[0]["target_user_id"] == uid + assert actions[0]["details"] == "pending → active" + assert actions[0]["admin_email"] == "admin@ex.fr" +``` + +- [ ] **Step 3: Run tests to verify they fail** + +Run: `uv run pytest tests/admin/test_routes.py -v` +Expected: FAIL — `admin_client`/`logged_in_admin_client` fixtures raise `ModuleNotFoundError: No module named 'src.admin.routes'` (via `init_auth` once it tries to import it in Step 5) or `404`/`ImportError` depending on order; before Step 5, `test_subscription_status_requires_admin` will fail with a connection/404-mismatch since the blueprint doesn't exist yet (real 404 from Flask's default routing, not from `is_admin()` — acceptable coincidence, but the other three tests will fail). + +- [ ] **Step 4: Implement `src/admin/routes.py`** + +```python +from flask import Blueprint, abort, redirect, request +from flask_login import current_user + +from src.admin.db import log_action +from src.admin.guard import is_admin +from src.subscriptions.db import SUBSCRIPTION_STATUSES, get_current, set_status + +admin_bp = Blueprint("admin", __name__, url_prefix="/admin/actions") + + +@admin_bp.before_request +def _require_admin(): + if not is_admin(): + abort(404) + + +@admin_bp.route("/subscription-status", methods=["POST"]) +def subscription_status(): + user_id = request.form.get("user_id", type=int) + subscription_id = request.form.get("subscription_id", type=int) + status = request.form.get("status", "") + + if user_id is None or subscription_id is None: + abort(400) + + current = get_current(user_id) + if ( + status not in SUBSCRIPTION_STATUSES + or current is None + or current["id"] != subscription_id + ): + return redirect(f"/admin/user/{user_id}?error=invalid_status") + + old_status = current["status"] + set_status(subscription_id, status) + log_action( + current_user.email, + "subscription_status_change", + user_id, + f"{old_status} → {status}", + ) + return redirect(f"/admin/user/{user_id}?status_changed=1") +``` + +- [ ] **Step 5: Register the blueprint** + +In `src/auth/setup.py`, immediately after `app.register_blueprint(auth_bp)` (currently the line right after the `from src.auth.routes import auth_bp` block): + +```python + from src.auth.routes import auth_bp + + app.register_blueprint(auth_bp) + + from src.admin.routes import admin_bp + + app.register_blueprint(admin_bp) +``` + +- [ ] **Step 6: Add `ADMIN_EMAIL` to `.template.env`** + +In `.template.env`, add a new section after the "Comptes utilisateurs" block (after `APP_BASE_URL=http://localhost:8050`): + +``` +# Panneau admin interne (accès à /admin, protégé par cette adresse) +ADMIN_EMAIL= +``` + +- [ ] **Step 7: Run tests to verify they pass** + +Run: `uv run pytest tests/admin/test_routes.py -v` +Expected: 4 passed + +- [ ] **Step 8: Commit** + +```bash +git add src/admin/routes.py src/auth/setup.py .template.env tests/admin/conftest.py tests/admin/test_routes.py +git commit -m "feat(admin): add subscription-status mutation route" +``` + +--- + +### Task 6: Shared page chrome + `/admin` list page + +**Files:** + +- Create: `src/pages/admin/__init__.py` (empty) +- Create: `src/pages/admin/_shell.py` +- Create: `src/pages/admin/liste.py` + +**Interfaces:** + +- Produces: `not_admin() -> Component`, `admin_nav(active: str) -> Component` (in `_shell.py`), reused by Tasks 7 and 8. +- Consumes: `is_admin()` (Task 4), `list_users()` (Task 1), `get_current()` (existing). + +- [ ] **Step 1: Create the package and shared shell** + +Create `src/pages/admin/__init__.py` (empty file). + +Create `src/pages/admin/_shell.py`: + +```python +import dash_bootstrap_components as dbc +from dash import html + + +def not_admin(): + return html.Div( + html.H2("404", id="admin-404-heading"), className="py-5 text-center" + ) + + +def admin_nav(active: str): + return dbc.Nav( + [ + dbc.NavLink("Utilisateurs", href="/admin", active=(active == "liste")), + dbc.NavLink( + "Journal", href="/admin/journal", active=(active == "journal") + ), + ], + pills=True, + class_name="mb-4", + ) +``` + +(`html.H2` here, not `H1` — the navbar already renders a global `

colibre

` logo on every page, see `src/app.py:221`; a second `

` would make text-based Selenium assertions ambiguous.) + +- [ ] **Step 2: Create the list page** + +Create `src/pages/admin/liste.py`: + +```python +import dash_bootstrap_components as dbc +from dash import dash_table, html, register_page + +from src.admin.guard import is_admin +from src.auth.db import list_users +from src.pages.admin._shell import admin_nav, not_admin +from src.subscriptions.db import get_current + +register_page( + __name__, + path="/admin", + title="Panneau admin | colibre", + name="Admin", + description="Panneau d'administration interne.", +) + + +def _rows(): + rows = [] + for user in list_users(): + sub = get_current(user["id"]) + rows.append( + { + "email": user["email"], + "vérifié": "oui" if user["email_verified"] else "non", + "plan": sub["plan"] if sub else "", + "statut": sub["status"] if sub else "", + "créé le": user["created_at"], + "voir": f"[Voir](/admin/user/{user['id']})", + } + ) + return rows + + +def layout(**_): + if not is_admin(): + return not_admin() + return dbc.Container( + [ + html.H2("Panneau admin"), + admin_nav("liste"), + dash_table.DataTable( + id="admin-users-table", + columns=[ + {"name": "Email", "id": "email"}, + {"name": "Vérifié", "id": "vérifié"}, + {"name": "Plan", "id": "plan"}, + {"name": "Statut", "id": "statut"}, + {"name": "Créé le", "id": "créé le"}, + {"name": "", "id": "voir", "presentation": "markdown"}, + ], + data=_rows(), + filter_action="native", + sort_action="native", + page_action="native", + page_size=20, + markdown_options={"link_target": "_self"}, + ), + ], + fluid=True, + className="py-4", + ) +``` + +- [ ] **Step 3: Sanity-check the layout renders without a browser** + +Run: + +```bash +uv run python -c " +import os +os.environ.setdefault('USERS_DB_PATH', 'tests/users.test.sqlite') +os.environ.setdefault('SECRET_KEY', 'x') +os.environ['ADMIN_EMAIL'] = 'admin@ex.fr' +from unittest.mock import patch +from flask import Flask +from src.auth.setup import init_auth +app = Flask(__name__) +app.config['WTF_CSRF_ENABLED'] = False +init_auth(app) +with app.test_request_context(): + import src.pages.admin.liste as liste + admin = type('U', (), {'is_authenticated': True, 'email': 'admin@ex.fr'})() + with patch('src.admin.guard.current_user', admin): + component = liste.layout() + print(type(component)) +" +``` + +Expected: prints `` with no traceback. (This does not assert content — that's covered by Task 9's Selenium test — it just proves the module imports and `layout()` executes without error, catching typos/import errors before the browser test.) + +- [ ] **Step 4: Commit** + +```bash +git add src/pages/admin/__init__.py src/pages/admin/_shell.py src/pages/admin/liste.py +git commit -m "feat(admin): add /admin user list page" +``` + +--- + +### Task 7: `/admin/user/` detail page + +**Files:** + +- Create: `src/pages/admin/detail.py` + +**Interfaces:** + +- Consumes: `not_admin`, `admin_nav` (Task 6), `is_admin` (Task 4), `get_user_by_id` (existing), `get_current`, `list_by_user`, `get_subscriber_state`, `SUBSCRIPTION_STATUSES` (Task 2). + +- [ ] **Step 1: Create the detail page** + +Create `src/pages/admin/detail.py`: + +```python +import dash_bootstrap_components as dbc +from dash import dcc, html, register_page + +from src.admin.guard import is_admin +from src.auth.db import get_user_by_id +from src.pages.admin._shell import admin_nav, not_admin +from src.subscriptions.db import ( + SUBSCRIPTION_STATUSES, + get_current, + get_subscriber_state, + list_by_user, +) + +register_page( + __name__, + path_template="/admin/user/", + title="Détail utilisateur | colibre", + name="Détail utilisateur", + description="Panneau d'administration interne.", +) + +ERROR_MESSAGES = {"invalid_status": "Statut invalide ou abonnement introuvable."} +SUCCESS_MESSAGES = {"status_changed": "Statut de l'abonnement mis à jour."} + + +def _parse_user_id(user_id): + try: + return int(user_id) + except (TypeError, ValueError): + return None + + +def _not_found(): + return dbc.Container( + [ + html.H2("Panneau admin"), + admin_nav("liste"), + dbc.Alert("Utilisateur introuvable.", color="warning"), + ], + fluid=True, + className="py-4", + ) + + +def _csrf(): + return dcc.Input( + type="hidden", + id={"type": "csrf-input", "index": "admin-status"}, + name="csrf_token", + ) + + +def _account_section(user): + return html.Div( + [ + html.H4("Compte"), + html.P([html.Strong("Email : "), user["email"]]), + html.P( + [ + html.Strong("Vérifié : "), + "oui" if user["email_verified"] else "non", + ] + ), + html.P([html.Strong("SIRET : "), user["siret"] or "—"]), + html.P([html.Strong("Créé le : "), user["created_at"]]), + ] + ) + + +def _state_section(user_id): + state = get_subscriber_state(user_id) + return html.Div( + [ + html.H4("État abonné", className="mt-4"), + html.P( + [ + html.Strong("Solde de votes : "), + str(state["votes_balance"] if state else 0), + ] + ), + html.P( + [ + html.Strong("Essai utilisé : "), + "oui" if state and state["trial_used"] else "non", + ] + ), + ] + ) + + +def _history_section(user_id, current_id): + rows = [] + for sub in list_by_user(user_id): + badge = ( + dbc.Badge("actuel", color="primary", className="ms-2") + if sub["id"] == current_id + else None + ) + rows.append( + html.Tr( + [ + html.Td([sub["plan"] or "—", badge]), + html.Td(sub["status"] or "—"), + html.Td(sub["prix_ht"] if sub["prix_ht"] is not None else "—"), + html.Td(sub["current_period_end"] or "—"), + html.Td(sub["created_at"]), + ] + ) + ) + return html.Div( + [ + html.H4("Historique des abonnements", className="mt-4"), + dbc.Table( + [ + html.Thead( + html.Tr( + [ + html.Th(h) + for h in ( + "Plan", + "Statut", + "Prix HT", + "Fin de période", + "Créé le", + ) + ] + ) + ), + html.Tbody(rows), + ], + bordered=True, + size="sm", + ), + ] + ) + + +def _status_form(user_id, current): + if current is None: + return html.Div() + return html.Div( + [ + html.H4("Changer le statut de l'abonnement courant", className="mt-4"), + html.Form( + method="POST", + action="/admin/actions/subscription-status", + children=[ + _csrf(), + dcc.Input(type="hidden", name="user_id", value=str(user_id)), + dcc.Input( + type="hidden", name="subscription_id", value=str(current["id"]) + ), + dbc.Select( + id="admin-status-select", + name="status", + options=[ + {"label": s, "value": s} for s in SUBSCRIPTION_STATUSES + ], + value=current["status"], + className="mb-3", + style={"maxWidth": "300px"}, + ), + dbc.Button("Mettre à jour le statut", type="submit", color="primary"), + ], + ), + ] + ) + + +def layout(user_id=None, error=None, status_changed=None, **_): + if not is_admin(): + return not_admin() + + uid = _parse_user_id(user_id) + user = get_user_by_id(uid) if uid is not None else None + if user is None: + return _not_found() + + alerts = [] + if error in ERROR_MESSAGES: + alerts.append(dbc.Alert(ERROR_MESSAGES[error], color="danger")) + if status_changed == "1": + alerts.append(dbc.Alert(SUCCESS_MESSAGES["status_changed"], color="success")) + + current = get_current(user["id"]) + + return dbc.Container( + [ + html.H2("Panneau admin"), + admin_nav("liste"), + *alerts, + _account_section(user), + _state_section(user["id"]), + _history_section(user["id"], current["id"] if current else None), + _status_form(user["id"], current), + ], + fluid=True, + className="py-4", + ) +``` + +- [ ] **Step 2: Sanity-check the layout renders without a browser** + +Run: + +```bash +uv run python -c " +import os +os.environ.setdefault('USERS_DB_PATH', 'tests/users.test.sqlite') +os.environ.setdefault('SECRET_KEY', 'x') +os.environ['ADMIN_EMAIL'] = 'admin@ex.fr' +from unittest.mock import patch +from flask import Flask +from src.auth.setup import init_auth +app = Flask(__name__) +app.config['WTF_CSRF_ENABLED'] = False +init_auth(app) +with app.test_request_context(): + import src.pages.admin.detail as detail + admin = type('U', (), {'is_authenticated': True, 'email': 'admin@ex.fr'})() + with patch('src.admin.guard.current_user', admin): + print(type(detail.layout(user_id='999999'))) +" +``` + +Expected: prints the `Container` type for the not-found branch, no traceback. + +- [ ] **Step 3: Commit** + +```bash +git add src/pages/admin/detail.py +git commit -m "feat(admin): add /admin/user/ detail page" +``` + +--- + +### Task 8: `/admin/journal` audit log page + +**Files:** + +- Create: `src/pages/admin/journal.py` + +**Interfaces:** + +- Consumes: `not_admin`, `admin_nav` (Task 6), `is_admin` (Task 4), `list_actions` (Task 3). + +- [ ] **Step 1: Create the journal page** + +Create `src/pages/admin/journal.py`: + +```python +import dash_bootstrap_components as dbc +from dash import dash_table, html, register_page + +from src.admin.db import list_actions +from src.admin.guard import is_admin +from src.pages.admin._shell import admin_nav, not_admin + +register_page( + __name__, + path="/admin/journal", + title="Journal admin | colibre", + name="Journal admin", + description="Panneau d'administration interne.", +) + + +def _rows(): + rows = [] + for action in list_actions(): + target = action["target_user_id"] + rows.append( + { + "date": action["created_at"], + "admin": action["admin_email"], + "action": action["action"], + "user": f"[{target}](/admin/user/{target})" if target else "", + "détails": action["details"] or "", + } + ) + return rows + + +def layout(**_): + if not is_admin(): + return not_admin() + return dbc.Container( + [ + html.H2("Journal des actions admin"), + admin_nav("journal"), + dash_table.DataTable( + id="admin-journal-table", + columns=[ + {"name": "Date", "id": "date"}, + {"name": "Admin", "id": "admin"}, + {"name": "Action", "id": "action"}, + {"name": "User", "id": "user", "presentation": "markdown"}, + {"name": "Détails", "id": "détails"}, + ], + data=_rows(), + sort_action="native", + page_action="native", + page_size=20, + markdown_options={"link_target": "_self"}, + ), + ], + fluid=True, + className="py-4", + ) +``` + +- [ ] **Step 2: Sanity-check the layout renders without a browser** + +Run: + +```bash +uv run python -c " +import os +os.environ.setdefault('USERS_DB_PATH', 'tests/users.test.sqlite') +os.environ.setdefault('SECRET_KEY', 'x') +os.environ['ADMIN_EMAIL'] = 'admin@ex.fr' +from unittest.mock import patch +from flask import Flask +from src.auth.setup import init_auth +app = Flask(__name__) +app.config['WTF_CSRF_ENABLED'] = False +init_auth(app) +with app.test_request_context(): + import src.pages.admin.journal as journal + admin = type('U', (), {'is_authenticated': True, 'email': 'admin@ex.fr'})() + with patch('src.admin.guard.current_user', admin): + print(type(journal.layout())) +" +``` + +Expected: prints the `Container` type, no traceback. + +- [ ] **Step 3: Commit** + +```bash +git add src/pages/admin/journal.py +git commit -m "feat(admin): add /admin/journal audit log page" +``` + +--- + +### Task 9: End-to-end Selenium coverage + +**Files:** + +- Create: `tests/admin/test_pages.py` + +**Interfaces:** + +- Consumes: `src.app.app` (the full Dash/Flask app), `src.auth.db`, `src.subscriptions.db`. + +**Note on `tests/users.test.sqlite`:** this file is committed to git and shared by the whole Selenium test session (`USERS_DB_PATH=tests/users.test.sqlite` is set globally in `pyproject.toml`'s `[tool.pytest.ini_options] env`, unlike the per-test-isolated `users_db_path` fixture used in Tasks 1-8). Every row this test creates **must** be deleted in a `finally` block so the committed file is left unchanged (`git status` clean) after a test run. Emails are also uuid-suffixed so repeated runs never collide even if cleanup is skipped due to a crash. + +- [ ] **Step 1: Write the test file** + +Create `tests/admin/test_pages.py`: + +```python +import uuid + +from dash.testing.composite import DashComposite +from selenium.webdriver.support.ui import Select +from werkzeug.security import generate_password_hash + +from src.auth import db as auth_db +from src.subscriptions import db as sub_db + +PASSWORD = "s3cretpass!" + + +def _unique_email(prefix: str) -> str: + return f"{prefix}-{uuid.uuid4().hex[:8]}@ex.fr" + + +def _make_verified_user(email: str) -> int: + auth_db.init_schema() + uid = auth_db.create_user(email, generate_password_hash(PASSWORD)) + auth_db.set_email_verified(uid) + return uid + + +def _cleanup_user(user_id: int) -> None: + conn = auth_db.get_conn() + conn.execute("DELETE FROM admin_actions WHERE target_user_id = ?", (user_id,)) + conn.execute("DELETE FROM subscriptions WHERE user_id = ?", (user_id,)) + conn.execute("DELETE FROM subscriber_state WHERE user_id = ?", (user_id,)) + conn.execute("DELETE FROM users WHERE id = ?", (user_id,)) + + +def _login(dash_duo: DashComposite, email: str): + dash_duo.driver.get(dash_duo.server_url + "/connexion") + dash_duo.wait_for_element("input[name=email]", timeout=8).send_keys(email) + dash_duo.driver.find_element("css selector", "input[name=password]").send_keys( + PASSWORD + ) + dash_duo.driver.find_element("css selector", "button[type=submit]").click() + + +def test_admin_anonymous_gets_404(dash_duo: DashComposite): + from src.app import app + + dash_duo.start_server(app) + dash_duo.driver.get(dash_duo.server_url + "/admin") + dash_duo.wait_for_text_to_equal("#admin-404-heading", "404", timeout=8) + + +def test_admin_non_admin_gets_404(dash_duo: DashComposite, monkeypatch): + from src.app import app + + monkeypatch.setenv("ADMIN_EMAIL", "admin-only@ex.fr") + email = _unique_email("regular") + uid = _make_verified_user(email) + try: + dash_duo.start_server(app) + _login(dash_duo, email) + dash_duo.driver.get(dash_duo.server_url + "/admin") + dash_duo.wait_for_text_to_equal("#admin-404-heading", "404", timeout=8) + finally: + _cleanup_user(uid) + + +def test_admin_full_flow(dash_duo: DashComposite, monkeypatch): + from src.app import app + + admin_email = _unique_email("admin") + monkeypatch.setenv("ADMIN_EMAIL", admin_email) + admin_uid = _make_verified_user(admin_email) + + target_email = _unique_email("target") + target_uid = _make_verified_user(target_email) + sub_db.init_schema() + _handle, sub_id = sub_db.create_pending(target_uid, "cust-e2e", "simple", 20.0) + sub_db.set_status(sub_id, "active") + + try: + dash_duo.start_server(app) + _login(dash_duo, admin_email) + + dash_duo.driver.get(dash_duo.server_url + "/admin") + dash_duo.wait_for_text_to_equal("h2", "Panneau admin", timeout=8) + assert target_email in dash_duo.driver.page_source + + dash_duo.driver.get(f"{dash_duo.server_url}/admin/user/{target_uid}") + dash_duo.wait_for_text_to_equal("h2", "Panneau admin", timeout=8) + assert target_email in dash_duo.driver.page_source + + select = Select( + dash_duo.driver.find_element("css selector", "select[name=status]") + ) + select.select_by_value("cancelled") + dash_duo.driver.find_element("css selector", "button[type=submit]").click() + + dash_duo.wait_for_text_to_equal( + ".alert-success", "Statut de l'abonnement mis à jour.", timeout=8 + ) + assert "cancelled" in dash_duo.driver.page_source + + dash_duo.driver.get(dash_duo.server_url + "/admin/journal") + dash_duo.wait_for_text_to_equal("h2", "Journal des actions admin", timeout=8) + assert "subscription_status_change" in dash_duo.driver.page_source + assert "active → cancelled" in dash_duo.driver.page_source + finally: + _cleanup_user(target_uid) + _cleanup_user(admin_uid) +``` + +- [ ] **Step 2: Run the tests** + +Run: `uv run pytest tests/admin/test_pages.py -v` +Expected: 3 passed + +- [ ] **Step 3: Verify `tests/users.test.sqlite` is unchanged** + +Run: `git status --short tests/users.test.sqlite` +Expected: no output (empty — file unchanged) + +If it shows as modified, the cleanup in Step 1 missed a row (check `admin_actions`, `subscriptions`, `subscriber_state`, `users` in that order — foreign keys cascade `ON DELETE CASCADE` from `users`, but `admin_actions.target_user_id` has no FK constraint so it needs the explicit `DELETE` shown above). Fix and re-run before committing. + +- [ ] **Step 4: Commit** + +```bash +git add tests/admin/test_pages.py +git commit -m "test(admin): add end-to-end Selenium coverage for the admin panel" +``` + +--- + +## Self-Review Notes + +- **Spec coverage:** Contexte/architecture → Tasks 4-5; couche données (`list_users`, `list_by_user`, `set_status`, `get_subscriber_state`, `SUBSCRIPTION_STATUSES`, `admin_actions`) → Tasks 1-3; pages (`/admin`, `/admin/user/`, `/admin/journal`) → Tasks 6-8; route de mutation → Task 5; tests (unitaires, route, pages) → Tasks 1-9 respectively; pagination native 20/page → Task 6 and 8. Hors périmètre items (reset mot de passe, suppression de compte, édition subscriber_state, multi-admin, pagination SQL) are intentionally not implemented anywhere in this plan. +- **Type consistency checked:** `is_admin()` (Task 4) used identically in `admin_bp.before_request` (Task 5) and in every page's `layout()` (Tasks 6-8). `SUBSCRIPTION_STATUSES` (Task 2) used identically in the route's validation (Task 5) and the detail page's dropdown (Task 7). `list_actions()` / `log_action()` signatures (Task 3) match their use in the route (Task 5) and journal page (Task 8). +- **`_not_admin` naming:** unified as `not_admin()` (no leading underscore) in `_shell.py` since it's imported across three page modules — an underscore-prefixed name would be misleading as a cross-module import.