colin/colibre
1
0
Fork 0
Code Issues 30 Pull Requests Actions Packages Projects Releases 33 Activity

feat(admin): replace dedicated pages with a generic table editor at /admin

Browse Source
This commit is contained in:
Colin Maudry
2026-07-03 14:19:01 +02:00
parent cf23863a30
commit d229b58f3b
9 changed files with 99 additions and 594 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/admin/routes.py
-42
View File
@@ -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")
src/auth/setup.py
-4
View File
@@ -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)
src/pages/admin/_shell.py
-12
View File
@@ -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",
)
src/pages/admin/detail.py
-203
View File
@@ -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/<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["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",
)
src/pages/admin/journal.py
-58
View File
@@ -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",
)
src/pages/admin/liste.py
+99 -31
View File
@@ -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),
)
tests/admin/conftest.py
-40
View File
@@ -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
tests/admin/test_pages.py
-128
View File
@@ -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)
tests/admin/test_routes.py
-76
View File
@@ -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"