colibre/docs/superpowers/plans/2026-07-03-admin-ui-plan.md

# 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 `<h1>colibre</h1>` logo on every page, see `src/app.py:221`; a second `<h1>` 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 `<class 'dash_bootstrap_components._components.Container.Container'>` 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/<user_id>` 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/<user_id>",
    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/<user_id> 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/<id>`, `/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.