281 lines
10 KiB
Python
281 lines
10 KiB
Python
"""Tests for M4-T04: AuthUser TOTP fields and auth_recovery_code migration."""
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from alembic import command
|
|
from alembic.config import Config
|
|
from sqlalchemy import create_engine, inspect, text
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.auth import AuthUser, RecoveryCode
|
|
|
|
|
|
def _make_app_alembic_config(database_url: str) -> Config:
|
|
config = Config("alembic_app.ini")
|
|
config.set_main_option("sqlalchemy.url", database_url)
|
|
return config
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture()
|
|
def totp_db(tmp_path: Path):
|
|
"""Temporary SQLite DB upgraded to the current Alembic head."""
|
|
db_path = tmp_path / "totp_test.db"
|
|
db_url = f"sqlite:///{db_path}"
|
|
alembic_cfg = _make_app_alembic_config(db_url)
|
|
command.upgrade(alembic_cfg, "head")
|
|
engine = create_engine(db_url, connect_args={"check_same_thread": False})
|
|
yield engine
|
|
engine.dispose()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Migration shape tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_auth_users_has_totp_columns(totp_db):
|
|
"""After upgrade to head, auth_users must have totp_secret and totp_enabled columns."""
|
|
inspector = inspect(totp_db)
|
|
columns = {col["name"]: col for col in inspector.get_columns("auth_users")}
|
|
|
|
assert "totp_secret" in columns, "totp_secret column missing from auth_users"
|
|
assert "totp_enabled" in columns, "totp_enabled column missing from auth_users"
|
|
|
|
# totp_secret must be nullable
|
|
assert columns["totp_secret"]["nullable"], "totp_secret should be nullable"
|
|
# totp_enabled must be NOT NULL
|
|
assert not columns["totp_enabled"]["nullable"], "totp_enabled should be NOT NULL"
|
|
|
|
|
|
def test_auth_recovery_code_table_exists(totp_db):
|
|
"""auth_recovery_code table must exist after upgrade to head."""
|
|
inspector = inspect(totp_db)
|
|
assert "auth_recovery_code" in inspector.get_table_names()
|
|
|
|
|
|
def test_auth_recovery_code_has_expected_columns(totp_db):
|
|
"""auth_recovery_code must have id, user_id, code_hash, used_at with correct nullability."""
|
|
inspector = inspect(totp_db)
|
|
columns = {col["name"]: col for col in inspector.get_columns("auth_recovery_code")}
|
|
|
|
expected_non_nullable = {"id", "user_id", "code_hash"}
|
|
expected_nullable = {"used_at"}
|
|
|
|
for col_name in expected_non_nullable:
|
|
assert col_name in columns, f"Missing column: {col_name}"
|
|
assert not columns[col_name]["nullable"], f"Column {col_name} should be NOT NULL"
|
|
|
|
for col_name in expected_nullable:
|
|
assert col_name in columns, f"Missing column: {col_name}"
|
|
assert columns[col_name]["nullable"], f"Column {col_name} should be nullable"
|
|
|
|
|
|
def test_auth_recovery_code_has_fk_to_auth_users(totp_db):
|
|
"""auth_recovery_code.user_id must have a FK constraint pointing to auth_users.id."""
|
|
inspector = inspect(totp_db)
|
|
fks = inspector.get_foreign_keys("auth_recovery_code")
|
|
assert len(fks) == 1, f"Expected 1 FK on auth_recovery_code, got {len(fks)}"
|
|
fk = fks[0]
|
|
assert fk["referred_table"] == "auth_users"
|
|
assert "user_id" in fk["constrained_columns"]
|
|
assert "id" in fk["referred_columns"]
|
|
|
|
|
|
def test_auth_recovery_code_user_id_index_exists(totp_db):
|
|
"""ix_auth_recovery_code_user_id index must exist."""
|
|
inspector = inspect(totp_db)
|
|
indexes = inspector.get_indexes("auth_recovery_code")
|
|
index_names = {idx["name"] for idx in indexes}
|
|
assert "ix_auth_recovery_code_user_id" in index_names
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Default value tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_totp_enabled_defaults_to_false(totp_db):
|
|
"""A newly inserted AuthUser must have totp_enabled=False and totp_secret=None."""
|
|
now = datetime.now(tz=timezone.utc)
|
|
with Session(totp_db) as session:
|
|
user = AuthUser(
|
|
username="testuser",
|
|
password_hash="$argon2id$v=19$m=65536,t=3,p=4$fakehash",
|
|
is_active=True,
|
|
force_password_change=False,
|
|
created_at=now,
|
|
)
|
|
session.add(user)
|
|
session.commit()
|
|
user_id = user.id
|
|
|
|
with Session(totp_db) as session:
|
|
fetched = session.get(AuthUser, user_id)
|
|
assert fetched is not None
|
|
assert fetched.totp_enabled is False, "totp_enabled should default to False"
|
|
assert fetched.totp_secret is None, "totp_secret should default to None"
|
|
|
|
|
|
def test_totp_secret_can_be_set(totp_db):
|
|
"""totp_secret can be set to a base32 string and retrieved correctly."""
|
|
now = datetime.now(tz=timezone.utc)
|
|
secret = "JBSWY3DPEHPK3PXP"
|
|
with Session(totp_db) as session:
|
|
user = AuthUser(
|
|
username="totp_user",
|
|
password_hash="$argon2id$v=19$m=65536,t=3,p=4$fakehash",
|
|
is_active=True,
|
|
force_password_change=False,
|
|
created_at=now,
|
|
totp_secret=secret,
|
|
totp_enabled=True,
|
|
)
|
|
session.add(user)
|
|
session.commit()
|
|
user_id = user.id
|
|
|
|
with Session(totp_db) as session:
|
|
fetched = session.get(AuthUser, user_id)
|
|
assert fetched is not None
|
|
assert fetched.totp_secret == secret
|
|
assert fetched.totp_enabled is True
|
|
|
|
|
|
def test_recovery_code_insert_and_used_at_nullable(totp_db):
|
|
"""A RecoveryCode can be inserted with used_at=None and later marked consumed."""
|
|
now = datetime.now(tz=timezone.utc)
|
|
with Session(totp_db) as session:
|
|
user = AuthUser(
|
|
username="rc_user",
|
|
password_hash="$argon2id$v=19$m=65536,t=3,p=4$fakehash",
|
|
is_active=True,
|
|
force_password_change=False,
|
|
created_at=now,
|
|
)
|
|
session.add(user)
|
|
session.flush()
|
|
|
|
code = RecoveryCode(
|
|
user_id=user.id,
|
|
code_hash="$argon2id$v=19$m=65536,t=3,p=4$rcfakehash",
|
|
used_at=None,
|
|
)
|
|
session.add(code)
|
|
session.commit()
|
|
code_id = code.id
|
|
|
|
with Session(totp_db) as session:
|
|
fetched = session.get(RecoveryCode, code_id)
|
|
assert fetched is not None
|
|
assert fetched.used_at is None
|
|
|
|
# Mark as consumed
|
|
fetched.used_at = now
|
|
session.commit()
|
|
|
|
with Session(totp_db) as session:
|
|
fetched = session.get(RecoveryCode, code_id)
|
|
assert fetched is not None
|
|
assert fetched.used_at is not None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Existing user migration test (not-break-login guarantee)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_existing_user_defaults_to_totp_disabled_after_upgrade(tmp_path: Path):
|
|
"""An existing user row inserted before the TOTP migration must have totp_enabled=0 after upgrade.
|
|
|
|
Simulates the real-world scenario: a pre-existing row in auth_users gets migrated
|
|
to the TOTP revision and the server_default ensures totp_enabled=false (0) for it.
|
|
"""
|
|
db_path = tmp_path / "existing_user_test.db"
|
|
db_url = f"sqlite:///{db_path}"
|
|
alembic_cfg = _make_app_alembic_config(db_url)
|
|
|
|
# Upgrade to the TOTP revision's predecessor (the throttle revision).
|
|
command.upgrade(alembic_cfg, "20260621_07_auth_login_throttle")
|
|
|
|
# Insert a user row directly (simulates a user that existed before the TOTP migration).
|
|
engine = create_engine(db_url, connect_args={"check_same_thread": False})
|
|
with engine.connect() as conn:
|
|
conn.execute(
|
|
text(
|
|
"INSERT INTO auth_users (username, password_hash, is_active, force_password_change, created_at)"
|
|
" VALUES ('legacy_admin', 'hash', 1, 0, '2026-01-01T00:00:00+00:00')"
|
|
)
|
|
)
|
|
conn.commit()
|
|
row = conn.execute(
|
|
text("SELECT id FROM auth_users WHERE username = 'legacy_admin'")
|
|
).fetchone()
|
|
legacy_user_id = row[0]
|
|
engine.dispose()
|
|
|
|
# Now upgrade to head (applies the TOTP migration).
|
|
command.upgrade(alembic_cfg, "head")
|
|
|
|
engine = create_engine(db_url, connect_args={"check_same_thread": False})
|
|
with engine.connect() as conn:
|
|
row = conn.execute(
|
|
text(
|
|
"SELECT totp_enabled, totp_secret FROM auth_users WHERE id = :uid"
|
|
),
|
|
{"uid": legacy_user_id},
|
|
).fetchone()
|
|
engine.dispose()
|
|
|
|
assert row is not None
|
|
# totp_enabled should be 0 (false) for the legacy row — server_default="0" ensures this.
|
|
assert row[0] == 0, f"Legacy user should have totp_enabled=0 after migration, got {row[0]}"
|
|
# totp_secret should be NULL for the legacy row.
|
|
assert row[1] is None, f"Legacy user should have totp_secret=NULL after migration, got {row[1]}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Downgrade / upgrade reversibility tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_downgrade_removes_totp_columns_and_table(tmp_path: Path):
|
|
"""downgrade -1 from head must drop auth_recovery_code and remove TOTP columns from auth_users."""
|
|
db_path = tmp_path / "down_totp_test.db"
|
|
db_url = f"sqlite:///{db_path}"
|
|
alembic_cfg = _make_app_alembic_config(db_url)
|
|
|
|
command.upgrade(alembic_cfg, "head")
|
|
|
|
engine = create_engine(db_url, connect_args={"check_same_thread": False})
|
|
inspector = inspect(engine)
|
|
assert "auth_recovery_code" in inspector.get_table_names()
|
|
auth_user_cols = {col["name"] for col in inspector.get_columns("auth_users")}
|
|
assert "totp_secret" in auth_user_cols
|
|
assert "totp_enabled" in auth_user_cols
|
|
engine.dispose()
|
|
|
|
# Downgrade one step (removes TOTP migration).
|
|
command.downgrade(alembic_cfg, "-1")
|
|
|
|
engine = create_engine(db_url, connect_args={"check_same_thread": False})
|
|
inspector = inspect(engine)
|
|
assert "auth_recovery_code" not in inspector.get_table_names(), (
|
|
"auth_recovery_code table should be gone after downgrade"
|
|
)
|
|
auth_user_cols_after = {col["name"] for col in inspector.get_columns("auth_users")}
|
|
assert "totp_secret" not in auth_user_cols_after, (
|
|
"totp_secret column should be removed after downgrade"
|
|
)
|
|
assert "totp_enabled" not in auth_user_cols_after, (
|
|
"totp_enabled column should be removed after downgrade"
|
|
)
|
|
engine.dispose()
|