diff --git a/src/auth/db.py b/src/auth/db.py index 75babdf..c6fe420 100644 --- a/src/auth/db.py +++ b/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,)) diff --git a/tests/auth/test_db.py b/tests/auth/test_db.py index 9180e1f..0d321b7 100644 --- a/tests/auth/test_db.py +++ b/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"