src/auth/db.py : CRUD users (#73)

src/auth/db.py

@@ -1,5 +1,6 @@
 import os
 import sqlite3
+from datetime import datetime, timezone
 from pathlib import Path
 from threading import Lock
 
@@ -64,3 +65,48 @@ def reset_conn_for_tests() -> None:
 def init_schema() -> None:
     conn = get_conn()
     conn.executescript(USERS_SCHEMA)
+
+
+def _now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def create_user(email: str, password_hash: str) -> int:
+    conn = get_conn()
+    now = _now()
+    cur = conn.execute(
+        "INSERT INTO users (email, password_hash, email_verified, created_at, updated_at) "
+        "VALUES (?, ?, 0, ?, ?)",
+        (email.lower(), password_hash, now, now),
+    )
+    return cur.lastrowid
+
+
+def get_user_by_email(email: str) -> sqlite3.Row | None:
+    return (
+        get_conn()
+        .execute("SELECT * FROM users WHERE email = ?", (email.lower(),))
+        .fetchone()
+    )
+
+
+def get_user_by_id(user_id: int) -> sqlite3.Row | None:
+    return get_conn().execute("SELECT * FROM users WHERE id = ?", (user_id,)).fetchone()
+
+
+def set_email_verified(user_id: int) -> None:
+    get_conn().execute(
+        "UPDATE users SET email_verified = 1, updated_at = ? WHERE id = ?",
+        (_now(), user_id),
+    )
+
+
+def update_password_hash(user_id: int, password_hash: str) -> None:
+    get_conn().execute(
+        "UPDATE users SET password_hash = ?, updated_at = ? WHERE id = ?",
+        (password_hash, _now(), user_id),
+    )
+
+
+def delete_user(user_id: int) -> None:
+    get_conn().execute("DELETE FROM users WHERE id = ?", (user_id,))

tests/auth/test_db.py

@@ -1,4 +1,14 @@
-from src.auth.db import get_conn, init_schema
+import pytest
+
+from src.auth.db import (
+    create_user,
+    get_conn,
+    get_user_by_email,
+    get_user_by_id,
+    init_schema,
+    set_email_verified,
+    update_password_hash,
+)
 
 
 def test_init_schema_creates_tables(users_db_path):
@@ -29,3 +39,53 @@ def test_pragmas_active(users_db_path):
     conn = get_conn()
     assert conn.execute("PRAGMA foreign_keys").fetchone()[0] == 1
     assert conn.execute("PRAGMA journal_mode").fetchone()[0].lower() == "wal"
+
+
+def test_create_user_and_get_by_email(users_db_path):
+    init_schema()
+    user_id = create_user("alice@example.com", "hash-bidon")
+    assert user_id > 0
+    row = get_user_by_email("alice@example.com")
+    assert row is not None
+    assert row["email"] == "alice@example.com"
+    assert row["email_verified"] == 0
+
+
+def test_email_is_lowercased(users_db_path):
+    init_schema()
+    create_user("Alice@Example.COM", "hash")
+    row = get_user_by_email("alice@example.com")
+    assert row is not None
+    row_upper = get_user_by_email("ALICE@example.com")
+    assert row_upper is not None
+    assert row["id"] == row_upper["id"]
+
+
+def test_duplicate_email_raises(users_db_path):
+    init_schema()
+    create_user("alice@example.com", "hash")
+    with pytest.raises(Exception):
+        create_user("alice@example.com", "autre")
+
+
+def test_get_user_by_id(users_db_path):
+    init_schema()
+    uid = create_user("a@b.c", "h")
+    row = get_user_by_id(uid)
+    assert row["email"] == "a@b.c"
+    assert get_user_by_id(999999) is None
+
+
+def test_set_email_verified(users_db_path):
+    init_schema()
+    uid = create_user("a@b.c", "h")
+    assert get_user_by_id(uid)["email_verified"] == 0
+    set_email_verified(uid)
+    assert get_user_by_id(uid)["email_verified"] == 1
+
+
+def test_update_password_hash(users_db_path):
+    init_schema()
+    uid = create_user("a@b.c", "old")
+    update_password_hash(uid, "new")
+    assert get_user_by_id(uid)["password_hash"] == "new"