refactor(admin): apply formatting fixes from linter hooks

2026-07-03 13:04:15 +02:00
parent 19991dd22f
commit cf23863a30
3 changed files with 270 additions and 0 deletions
@@ -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
@@ -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()

@@ -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