src/auth/db.py : schéma SQLite et connexion (#73)

src/auth/db.py

@@ -0,0 +1,66 @@
+import os
+import sqlite3
+from pathlib import Path
+from threading import Lock
+
+_conn: sqlite3.Connection | None = None
+_conn_lock = Lock()
+
+USERS_SCHEMA = """
+CREATE TABLE IF NOT EXISTS users (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    email TEXT NOT NULL UNIQUE,
+    password_hash TEXT NOT NULL,
+    email_verified INTEGER NOT NULL DEFAULT 0,
+    created_at TEXT NOT NULL,
+    updated_at TEXT NOT NULL
+);
+
+CREATE UNIQUE INDEX IF NOT EXISTS idx_users_email ON users(email);
+
+CREATE TABLE IF NOT EXISTS email_verification_tokens (
+    token_hash TEXT PRIMARY KEY,
+    user_id INTEGER NOT NULL,
+    expires_at TEXT NOT NULL,
+    created_at TEXT NOT NULL,
+    FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
+);
+
+CREATE TABLE IF NOT EXISTS password_reset_tokens (
+    token_hash TEXT PRIMARY KEY,
+    user_id INTEGER NOT NULL,
+    expires_at TEXT NOT NULL,
+    created_at TEXT NOT NULL,
+    FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
+);
+"""
+
+
+def _db_path() -> Path:
+    return Path(os.getenv("USERS_DB_PATH", "users.sqlite"))
+
+
+def get_conn() -> sqlite3.Connection:
+    global _conn
+    with _conn_lock:
+        if _conn is None:
+            _conn = sqlite3.connect(
+                str(_db_path()), check_same_thread=False, isolation_level=None
+            )
+            _conn.row_factory = sqlite3.Row
+            _conn.execute("PRAGMA foreign_keys = ON")
+            _conn.execute("PRAGMA journal_mode = WAL")
+        return _conn
+
+
+def reset_conn_for_tests() -> None:
+    global _conn
+    with _conn_lock:
+        if _conn is not None:
+            _conn.close()
+        _conn = None
+
+
+def init_schema() -> None:
+    conn = get_conn()
+    conn.executescript(USERS_SCHEMA)

tests/auth/conftest.py

@@ -0,0 +1,12 @@
+import pytest
+
+
+@pytest.fixture
+def users_db_path(monkeypatch, tmp_path):
+    from src.auth.db import reset_conn_for_tests
+
+    db_path = tmp_path / "users.test.sqlite"
+    monkeypatch.setenv("USERS_DB_PATH", str(db_path))
+    reset_conn_for_tests()
+    yield db_path
+    reset_conn_for_tests()

tests/auth/test_db.py

@@ -0,0 +1,31 @@
+from src.auth.db import get_conn, init_schema
+
+
+def test_init_schema_creates_tables(users_db_path):
+    init_schema()
+    conn = get_conn()
+    tables = {
+        row[0]
+        for row in conn.execute(
+            "SELECT name FROM sqlite_master WHERE type='table'"
+        ).fetchall()
+    }
+    assert {"users", "email_verification_tokens", "password_reset_tokens"} <= tables
+
+
+def test_init_schema_is_idempotent(users_db_path):
+    init_schema()
+    init_schema()
+    conn = get_conn()
+    tables = conn.execute(
+        "SELECT name FROM sqlite_master WHERE type='table'"
+    ).fetchall()
+    names = [r[0] for r in tables]
+    assert names.count("users") == 1
+
+
+def test_pragmas_active(users_db_path):
+    init_schema()
+    conn = get_conn()
+    assert conn.execute("PRAGMA foreign_keys").fetchone()[0] == 1
+    assert conn.execute("PRAGMA journal_mode").fetchone()[0].lower() == "wal"