Add admin user seeding functionality and corresponding tests
This commit is contained in:
@@ -75,3 +75,26 @@ def _run_migrations(conn: duckdb.DuckDBPyConnection) -> None:
|
|||||||
created_at TIMESTAMP DEFAULT now()
|
created_at TIMESTAMP DEFAULT now()
|
||||||
)
|
)
|
||||||
""")
|
""")
|
||||||
|
_seed_admin(conn)
|
||||||
|
|
||||||
|
|
||||||
|
def _seed_admin(conn: duckdb.DuckDBPyConnection) -> None:
|
||||||
|
"""Insert the default admin user if it doesn't already exist."""
|
||||||
|
from passlib.context import CryptContext
|
||||||
|
_pwd = CryptContext(schemes=["bcrypt"], deprecated="auto")
|
||||||
|
|
||||||
|
email = os.getenv("ADMIN_EMAIL", "ai@allucanget.biz")
|
||||||
|
password = os.getenv("ADMIN_PASSWORD", "admin123")
|
||||||
|
|
||||||
|
existing = conn.execute(
|
||||||
|
"SELECT id FROM users WHERE email = ?", [email]
|
||||||
|
).fetchone()
|
||||||
|
if existing is None:
|
||||||
|
password_hash = _pwd.hash(password)
|
||||||
|
conn.execute(
|
||||||
|
"""
|
||||||
|
INSERT INTO users (email, password_hash, role)
|
||||||
|
VALUES (?, ?, 'admin')
|
||||||
|
""",
|
||||||
|
[email, password_hash],
|
||||||
|
)
|
||||||
|
|||||||
@@ -192,3 +192,39 @@ async def test_write_lock_serialises_concurrent_writes():
|
|||||||
# Each writer's start and end must be adjacent (no interleaving)
|
# Each writer's start and end must be adjacent (no interleaving)
|
||||||
assert order.index("A-start") + 1 == order.index("A-end") or \
|
assert order.index("A-start") + 1 == order.index("A-end") or \
|
||||||
order.index("B-start") + 1 == order.index("B-end")
|
order.index("B-start") + 1 == order.index("B-end")
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Admin seed user
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
def test_seed_admin_user_created_on_init():
|
||||||
|
import os
|
||||||
|
conn = db_module.init_db(":memory:")
|
||||||
|
row = conn.execute(
|
||||||
|
"SELECT email, role FROM users WHERE email = 'ai@allucanget.biz'"
|
||||||
|
).fetchone()
|
||||||
|
assert row is not None
|
||||||
|
assert row[0] == "ai@allucanget.biz"
|
||||||
|
assert row[1] == "admin"
|
||||||
|
|
||||||
|
|
||||||
|
def test_seed_admin_is_idempotent():
|
||||||
|
conn = db_module.init_db(":memory:")
|
||||||
|
# Simulate re-running seed (second init_db call reuses connection, so call _seed_admin directly)
|
||||||
|
db_module._seed_admin(conn)
|
||||||
|
count = conn.execute(
|
||||||
|
"SELECT COUNT(*) FROM users WHERE email = 'ai@allucanget.biz'"
|
||||||
|
).fetchone()[0]
|
||||||
|
assert count == 1
|
||||||
|
|
||||||
|
|
||||||
|
def test_seed_admin_email_env_override(monkeypatch):
|
||||||
|
monkeypatch.setenv("ADMIN_EMAIL", "custom@example.com")
|
||||||
|
monkeypatch.setenv("ADMIN_PASSWORD", "custompass")
|
||||||
|
conn = db_module.init_db(":memory:")
|
||||||
|
row = conn.execute(
|
||||||
|
"SELECT email, role FROM users WHERE email = 'custom@example.com'"
|
||||||
|
).fetchone()
|
||||||
|
assert row is not None
|
||||||
|
assert row[1] == "admin"
|
||||||
|
|||||||
Reference in New Issue
Block a user