feat(auth): email en attente (pending_email) + migration (#73)

Adds pending_email column to users table with idempotent migration.
Implements set_pending_email() and promote_pending_email() for email-change workflow.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

src/auth/db.py

@@ -12,6 +12,7 @@ CREATE TABLE IF NOT EXISTS users (
     email TEXT NOT NULL UNIQUE,
     password_hash TEXT NOT NULL,
     email_verified INTEGER NOT NULL DEFAULT 0,
+    pending_email TEXT,
     created_at TEXT NOT NULL,
     updated_at TEXT NOT NULL
 );
@@ -61,6 +62,13 @@ def reset_conn_for_tests() -> None:
 def init_schema() -> None:
     conn = get_conn()
     conn.executescript(USERS_SCHEMA)
+    _migrate(conn)
+
+
+def _migrate(conn: sqlite3.Connection) -> None:
+    cols = {row["name"] for row in conn.execute("PRAGMA table_info(users)")}
+    if "pending_email" not in cols:
+        conn.execute("ALTER TABLE users ADD COLUMN pending_email TEXT")
 
 
 def _now() -> str:
@@ -104,6 +112,29 @@ def update_password_hash(user_id: int, password_hash: str) -> None:
     )
 
 
+def set_pending_email(user_id: int, email: str) -> None:
+    get_conn().execute(
+        "UPDATE users SET pending_email = ?, updated_at = ? WHERE id = ?",
+        (email.lower(), _now(), user_id),
+    )
+
+
+def promote_pending_email(user_id: int) -> str | None:
+    conn = get_conn()
+    row = conn.execute(
+        "SELECT pending_email FROM users WHERE id = ?", (user_id,)
+    ).fetchone()
+    if row is None or not row["pending_email"]:
+        return None
+    new_email = row["pending_email"]
+    conn.execute(
+        "UPDATE users SET email = ?, pending_email = NULL, "
+        "email_verified = 1, updated_at = ? WHERE id = ?",
+        (new_email, _now(), user_id),
+    )
+    return new_email
+
+
 def delete_user(user_id: int) -> None:
     get_conn().execute("DELETE FROM users WHERE id = ?", (user_id,))
 

tests/auth/test_db.py

@@ -165,3 +165,40 @@ def test_cascade_delete_on_user_delete(users_db_path):
     create_email_verification_token("t", uid, _future(1))
     delete_user(uid)
     assert find_email_verification_token("t") is None
+
+
+def test_pending_email_column_exists(users_db_path):
+    from src.auth import db
+
+    db.init_schema()
+    cols = {r["name"] for r in db.get_conn().execute("PRAGMA table_info(users)")}
+    assert "pending_email" in cols
+
+
+def test_set_and_promote_pending_email(users_db_path):
+    from werkzeug.security import generate_password_hash
+
+    from src.auth import db
+
+    db.init_schema()
+    uid = db.create_user("old@example.fr", generate_password_hash("password12"))
+    db.set_pending_email(uid, "new@example.fr")
+    assert db.get_user_by_id(uid)["pending_email"] == "new@example.fr"
+
+    promoted = db.promote_pending_email(uid)
+    assert promoted == "new@example.fr"
+    row = db.get_user_by_id(uid)
+    assert row["email"] == "new@example.fr"
+    assert row["pending_email"] is None
+    assert row["email_verified"] == 1
+
+
+def test_promote_pending_email_noop_when_empty(users_db_path):
+    from werkzeug.security import generate_password_hash
+
+    from src.auth import db
+
+    db.init_schema()
+    uid = db.create_user("a@b.c", generate_password_hash("password12"))
+    assert db.promote_pending_email(uid) is None
+    assert db.get_user_by_id(uid)["email"] == "a@b.c"