From cf23863a303792b86883ce6bbf3d4b144b3c329d Mon Sep 17 00:00:00 2001 From: Colin Maudry Date: Fri, 3 Jul 2026 13:04:15 +0200 Subject: [PATCH] refactor(admin): apply formatting fixes from linter hooks --- src/admin/tables.py | 152 +++++++++++++++++++++++++++++++++++++ tests/admin/conftest.py | 8 ++ tests/admin/test_tables.py | 110 +++++++++++++++++++++++++++ 3 files changed, 270 insertions(+) create mode 100644 src/admin/tables.py create mode 100644 tests/admin/test_tables.py diff --git a/src/admin/tables.py b/src/admin/tables.py new file mode 100644 index 0000000..a1240cc --- /dev/null +++ b/src/admin/tables.py @@ -0,0 +1,152 @@ +from dataclasses import dataclass +from typing import Callable + +from src.auth.db import get_conn +from src.subscriptions.db import SUBSCRIPTION_STATUSES +from src.subscriptions.plans import PLANS + + +@dataclass(frozen=True) +class TableConfig: + columns: list[str] + editable_columns: frozenset[str] + pk: str + column_types: dict[str, type] + dropdowns: dict[str, list[str]] + target_user_id: Callable[[dict], int | None] + + +TABLES: dict[str, TableConfig] = { + "users": TableConfig( + columns=[ + "id", + "email", + "email_verified", + "siret", + "pending_email", + "created_at", + "updated_at", + ], + editable_columns=frozenset( + {"email", "email_verified", "siret", "pending_email"} + ), + pk="id", + column_types={ + "email": str, + "email_verified": int, + "siret": str, + "pending_email": str, + }, + dropdowns={"email_verified": ["0", "1"]}, + target_user_id=lambda row: row["id"], + ), + "subscriptions": TableConfig( + columns=[ + "id", + "user_id", + "frisbii_customer_handle", + "frisbii_subscription_handle", + "plan", + "prix_ht", + "status", + "current_period_end", + "created_at", + "updated_at", + ], + editable_columns=frozenset({"plan", "prix_ht", "status", "current_period_end"}), + pk="id", + column_types={ + "plan": str, + "prix_ht": float, + "status": str, + "current_period_end": str, + }, + dropdowns={ + "status": list(SUBSCRIPTION_STATUSES), + "plan": list(PLANS.keys()), + }, + target_user_id=lambda row: row["user_id"], + ), + "subscriber_state": TableConfig( + columns=[ + "user_id", + "trial_used", + "votes_balance", + "votes_last_credited_at", + "updated_at", + ], + editable_columns=frozenset( + {"trial_used", "votes_balance", "votes_last_credited_at"} + ), + pk="user_id", + column_types={ + "trial_used": int, + "votes_balance": int, + "votes_last_credited_at": str, + }, + dropdowns={"trial_used": ["0", "1"]}, + target_user_id=lambda row: row["user_id"], + ), + "admin_actions": TableConfig( + columns=[ + "id", + "admin_email", + "action", + "target_user_id", + "details", + "created_at", + ], + editable_columns=frozenset(), + pk="id", + column_types={}, + dropdowns={}, + target_user_id=lambda row: None, + ), +} + + +def get_rows(table: str) -> list[dict]: + cfg = TABLES[table] + cols_sql = ", ".join(cfg.columns) + rows = get_conn().execute(f"SELECT {cols_sql} FROM {table}").fetchall() + return [dict(row) for row in rows] + + +def _coerce_value(table: str, column: str, value): + cfg = TABLES[table] + if column not in cfg.editable_columns: + raise ValueError(f"Colonne non éditable : {column}") + if column in cfg.dropdowns and str(value) not in cfg.dropdowns[column]: + raise ValueError(f"Valeur non autorisée pour {column} : {value!r}") + expected_type = cfg.column_types[column] + try: + if expected_type is int: + return int(value) + if expected_type is float: + return float(value) + return str(value) + except (TypeError, ValueError) as exc: + raise ValueError(f"Valeur invalide pour {column} : {value!r}") from exc + + +def set_cell(table: str, pk_value, column: str, value) -> None: + if table not in TABLES: + raise ValueError(f"Table inconnue : {table}") + cfg = TABLES[table] + coerced = _coerce_value(table, column, value) + get_conn().execute( + f"UPDATE {table} SET {column} = ? WHERE {cfg.pk} = ?", (coerced, pk_value) + ) + + +def find_changed_cell( + data: list[dict], data_previous: list[dict] | None +) -> tuple[int, str, object, object] | None: + if data_previous is None or len(data) != len(data_previous): + return None + for i, (new_row, old_row) in enumerate(zip(data, data_previous)): + for col, new_val in new_row.items(): + old_val = old_row.get(col) + if new_val != old_val: + return i, col, old_val, new_val + return None diff --git a/tests/admin/conftest.py b/tests/admin/conftest.py index 30ffc09..7a0e3a7 100644 --- a/tests/admin/conftest.py +++ b/tests/admin/conftest.py @@ -3,11 +3,19 @@ import pytest @pytest.fixture def users_db_path(monkeypatch, tmp_path): + from src import migrations + from src.auth import db as auth_db from src.auth.db import reset_conn_for_tests + from src.subscriptions import db as sub_db 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() diff --git a/tests/admin/test_tables.py b/tests/admin/test_tables.py new file mode 100644 index 0000000..d85e91e --- /dev/null +++ b/tests/admin/test_tables.py @@ -0,0 +1,110 @@ +import pytest + +from src.admin import tables + + +def test_get_rows_users_excludes_password_hash(users_db_path): + from src.auth import db as auth_db + + auth_db.init_schema() + auth_db.create_user("a@ex.fr", "secret-hash") + + rows = tables.get_rows("users") + + assert rows[0]["email"] == "a@ex.fr" + assert "password_hash" not in rows[0] + + +def test_set_cell_rejects_unknown_table(users_db_path): + with pytest.raises(ValueError): + tables.set_cell("not_a_table", 1, "email", "x@ex.fr") + + +def test_set_cell_rejects_non_editable_column(users_db_path): + from src.auth import db as auth_db + + auth_db.init_schema() + uid = auth_db.create_user("a@ex.fr", "hash") + + with pytest.raises(ValueError): + tables.set_cell("users", uid, "id", "999") + + +def test_set_cell_rejects_invalid_dropdown_value(users_db_path): + from src.auth import db as auth_db + from src.subscriptions import db as sub_db + + auth_db.init_schema() + sub_db.init_schema() + uid = auth_db.create_user("a@ex.fr", "hash") + _handle, sub_id = sub_db.create_pending(uid, "cust-1", "simple") + + with pytest.raises(ValueError): + tables.set_cell("subscriptions", sub_id, "status", "not_a_status") + + +def test_set_cell_rejects_bad_type(users_db_path): + from src.auth import db as auth_db + from src.subscriptions import db as sub_db + + auth_db.init_schema() + sub_db.init_schema() + uid = auth_db.create_user("a@ex.fr", "hash") + _handle, sub_id = sub_db.create_pending(uid, "cust-1", "simple") + + with pytest.raises(ValueError): + tables.set_cell("subscriptions", sub_id, "prix_ht", "not-a-number") + + +def test_set_cell_writes_valid_value(users_db_path): + from src.auth import db as auth_db + + auth_db.init_schema() + uid = auth_db.create_user("a@ex.fr", "hash") + + tables.set_cell("users", uid, "siret", "12345678900011") + + rows = tables.get_rows("users") + assert rows[0]["siret"] == "12345678900011" + + +def test_set_cell_coerces_numeric_type(users_db_path): + from src.auth import db as auth_db + from src.subscriptions import db as sub_db + + auth_db.init_schema() + sub_db.init_schema() + uid = auth_db.create_user("a@ex.fr", "hash") + _handle, sub_id = sub_db.create_pending(uid, "cust-1", "simple") + + tables.set_cell("subscriptions", sub_id, "prix_ht", "30") + + rows = tables.get_rows("subscriptions") + assert rows[0]["prix_ht"] == 30.0 + + +def test_find_changed_cell_detects_single_diff(): + data = [{"id": 1, "email": "new@ex.fr"}] + data_previous = [{"id": 1, "email": "old@ex.fr"}] + + result = tables.find_changed_cell(data, data_previous) + + assert result == (0, "email", "old@ex.fr", "new@ex.fr") + + +def test_find_changed_cell_returns_none_when_identical(): + data = [{"id": 1, "email": "a@ex.fr"}] + data_previous = [{"id": 1, "email": "a@ex.fr"}] + + assert tables.find_changed_cell(data, data_previous) is None + + +def test_find_changed_cell_returns_none_when_previous_is_none(): + assert tables.find_changed_cell([{"id": 1}], None) is None + + +def test_target_user_id_per_table(): + assert tables.TABLES["users"].target_user_id({"id": 7}) == 7 + assert tables.TABLES["subscriptions"].target_user_id({"user_id": 9}) == 9 + assert tables.TABLES["subscriber_state"].target_user_id({"user_id": 3}) == 3 + assert tables.TABLES["admin_actions"].target_user_id({"id": 1}) is None