From d229b58f3bd339e5ada3747c3a73cc3ac8297bbc Mon Sep 17 00:00:00 2001 From: Colin Maudry Date: Fri, 3 Jul 2026 14:19:01 +0200 Subject: [PATCH] feat(admin): replace dedicated pages with a generic table editor at /admin --- src/admin/routes.py | 42 -------- src/auth/setup.py | 4 - src/pages/admin/_shell.py | 12 --- src/pages/admin/detail.py | 203 ------------------------------------- src/pages/admin/journal.py | 58 ----------- src/pages/admin/liste.py | 130 ++++++++++++++++++------ tests/admin/conftest.py | 40 -------- tests/admin/test_pages.py | 128 ----------------------- tests/admin/test_routes.py | 76 -------------- 9 files changed, 99 insertions(+), 594 deletions(-) delete mode 100644 src/admin/routes.py delete mode 100644 src/pages/admin/detail.py delete mode 100644 src/pages/admin/journal.py delete mode 100644 tests/admin/test_pages.py delete mode 100644 tests/admin/test_routes.py diff --git a/src/admin/routes.py b/src/admin/routes.py deleted file mode 100644 index ba9ae2d..0000000 --- a/src/admin/routes.py +++ /dev/null @@ -1,42 +0,0 @@ -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") diff --git a/src/auth/setup.py b/src/auth/setup.py index 5e4197f..ea6c717 100644 --- a/src/auth/setup.py +++ b/src/auth/setup.py @@ -48,10 +48,6 @@ def init_auth(app: Flask) -> None: app.register_blueprint(auth_bp) - from src.admin.routes import admin_bp - - app.register_blueprint(admin_bp) - init_oauth(app) _csrf = CSRFProtect(app) diff --git a/src/pages/admin/_shell.py b/src/pages/admin/_shell.py index 0b724f0..d2ab837 100644 --- a/src/pages/admin/_shell.py +++ b/src/pages/admin/_shell.py @@ -1,4 +1,3 @@ -import dash_bootstrap_components as dbc from dash import html @@ -6,14 +5,3 @@ 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", - ) diff --git a/src/pages/admin/detail.py b/src/pages/admin/detail.py deleted file mode 100644 index c67d960..0000000 --- a/src/pages/admin/detail.py +++ /dev/null @@ -1,203 +0,0 @@ -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["frisbii_subscription_handle"] 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", - "Handle Frisbii", - "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", - ) diff --git a/src/pages/admin/journal.py b/src/pages/admin/journal.py deleted file mode 100644 index 0a0e4ea..0000000 --- a/src/pages/admin/journal.py +++ /dev/null @@ -1,58 +0,0 @@ -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", - ) diff --git a/src/pages/admin/liste.py b/src/pages/admin/liste.py index 2c0442b..7967d84 100644 --- a/src/pages/admin/liste.py +++ b/src/pages/admin/liste.py @@ -1,10 +1,21 @@ import dash_bootstrap_components as dbc -from dash import dash_table, html, register_page +from dash import ( + Input, + Output, + State, + callback, + ctx, + dash_table, + html, + no_update, + register_page, +) +from flask_login import current_user +from src.admin.db import log_action 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 +from src.admin.tables import TABLES, find_changed_cell, get_rows, set_cell +from src.pages.admin._shell import not_admin register_page( __name__, @@ -14,22 +25,28 @@ register_page( description="Panneau d'administration interne.", ) +DEFAULT_TABLE = "users" -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 _columns_for(table: str): + cfg = TABLES[table] + return [ + { + "name": col, + "id": col, + "editable": col in cfg.editable_columns, + **({"presentation": "dropdown"} if col in cfg.dropdowns else {}), + } + for col in cfg.columns + ] + + +def _dropdown_for(table: str): + cfg = TABLES[table] + return { + col: {"options": [{"label": v, "value": v} for v in values]} + for col, values in cfg.dropdowns.items() + } def layout(**_): @@ -38,25 +55,76 @@ def layout(**_): return dbc.Container( [ html.H2("Panneau admin"), - admin_nav("liste"), + html.Div(id="admin-alerts"), + dbc.Select( + id="admin-table-select", + options=[{"label": name, "value": name} for name in TABLES], + value=DEFAULT_TABLE, + className="mb-3", + style={"maxWidth": "300px"}, + ), 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(), + id="admin-table", + columns=_columns_for(DEFAULT_TABLE), + data=get_rows(DEFAULT_TABLE), + dropdown=_dropdown_for(DEFAULT_TABLE), + editable=True, filter_action="native", sort_action="native", page_action="native", page_size=20, - markdown_options={"link_target": "_self"}, ), ], fluid=True, className="py-4", ) + + +@callback( + Output("admin-table", "data"), + Output("admin-table", "columns"), + Output("admin-table", "dropdown"), + Output("admin-alerts", "children"), + Input("admin-table-select", "value"), + Input("admin-table", "data"), + State("admin-table", "data_previous"), + prevent_initial_call=True, +) +def _update_table(selected_table, data, data_previous): + if ctx.triggered_id == "admin-table-select": + return ( + get_rows(selected_table), + _columns_for(selected_table), + _dropdown_for(selected_table), + None, + ) + + change = find_changed_cell(data, data_previous) + if change is None: + return no_update, no_update, no_update, None + + row_index, column, old_value, new_value = change + pk_value = data[row_index][TABLES[selected_table].pk] + try: + set_cell(selected_table, pk_value, column, new_value) + except ValueError as exc: + return ( + no_update, + no_update, + no_update, + dbc.Alert(str(exc), color="danger", dismissable=True), + ) + + target_user_id = TABLES[selected_table].target_user_id(data[row_index]) + log_action( + current_user.email, + f"edit_{selected_table}", + target_user_id, + f"{column}: {old_value!r} → {new_value!r}", + ) + return ( + no_update, + no_update, + no_update, + dbc.Alert("Modification enregistrée.", color="success", dismissable=True), + ) diff --git a/tests/admin/conftest.py b/tests/admin/conftest.py index 7a0e3a7..6b38a8c 100644 --- a/tests/admin/conftest.py +++ b/tests/admin/conftest.py @@ -10,49 +10,9 @@ def users_db_path(monkeypatch, tmp_path): db_path = tmp_path / "users.test.sqlite" monkeypatch.setenv("USERS_DB_PATH", str(db_path)) - monkeypatch.setenv("FRISBII_PLAN_SIMPLE", "plan_simple") - monkeypatch.setenv("FRISBII_PLAN_SOUTIEN", "plan_soutien") reset_conn_for_tests() auth_db.init_schema() sub_db.init_schema() migrations.apply_pending() yield db_path reset_conn_for_tests() - - -@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 diff --git a/tests/admin/test_pages.py b/tests/admin/test_pages.py deleted file mode 100644 index 26b6dd5..0000000 --- a/tests/admin/test_pages.py +++ /dev/null @@ -1,128 +0,0 @@ -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,)) - # tests/users.test.sqlite is committed to git and shared by the whole - # Selenium session (USERS_DB_PATH is set globally in pyproject.toml). - # Row DELETEs alone leave the file byte-diffed: AUTOINCREMENT tables - # record their high-water mark in sqlite_sequence, and that table is - # never rolled back by DELETE (by SQLite design). Reset it explicitly so - # a full test run leaves the committed file untouched (`git status` - # clean), matching the pre-existing convention for this file where - # users/subscriptions already sit at seq=0. - conn.execute( - "UPDATE sqlite_sequence SET seq = 0 " - "WHERE name IN ('users', 'subscriptions', 'admin_actions')" - ) - - -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) - # Confirm login actually succeeded before relying on the admin guard. - # A failed login (bad credentials, unverified email, ...) redirects - # back to /connexion rather than raising, leaving the session - # anonymous — which would also yield a 404 at /admin and make this - # test indistinguishable from test_admin_anonymous_gets_404. Since - # this user has no subscription, a successful login redirects to - # /compte/abonnement (see _post_login_url in src/auth/routes.py). - dash_duo.wait_for_text_to_equal("h2", "Abonnement", timeout=8) - assert "/connexion" not in dash_duo.driver.current_url - - 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) diff --git a/tests/admin/test_routes.py b/tests/admin/test_routes.py deleted file mode 100644 index a105ba1..0000000 --- a/tests/admin/test_routes.py +++ /dev/null @@ -1,76 +0,0 @@ -import itertools - -from src.auth import db as auth_db -from src.subscriptions import db as sub_db - -_email_counter = itertools.count(1) - - -def _make_target_with_subscription(): - uid = auth_db.create_user(f"target{next(_email_counter)}@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"