196 lines
6.4 KiB
Python
196 lines
6.4 KiB
Python
"""Tests for M4-T01: LoginThrottle model and auth_login_throttle 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
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.auth_throttle import LoginThrottle
|
|
|
|
|
|
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 throttle_db(tmp_path: Path):
|
|
"""Temporary SQLite DB upgraded to the current Alembic head."""
|
|
db_path = tmp_path / "throttle_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_table_exists_after_upgrade(throttle_db):
|
|
"""auth_login_throttle table must be present after upgrade to head."""
|
|
inspector = inspect(throttle_db)
|
|
assert "auth_login_throttle" in inspector.get_table_names()
|
|
|
|
|
|
def test_table_has_expected_columns(throttle_db):
|
|
"""All required columns must be present with correct nullability."""
|
|
inspector = inspect(throttle_db)
|
|
columns = {col["name"]: col for col in inspector.get_columns("auth_login_throttle")}
|
|
|
|
expected_non_nullable = {"id", "key", "scope", "failures", "first_failed_at", "last_failed_at"}
|
|
expected_nullable = {"next_allowed_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_unique_constraint_scope_key(throttle_db):
|
|
"""(scope, key) pair must be unique — inserting a duplicate must raise."""
|
|
now = datetime.now(tz=timezone.utc)
|
|
with Session(throttle_db) as session:
|
|
session.add(
|
|
LoginThrottle(
|
|
key="192.168.1.1",
|
|
scope="ip",
|
|
failures=1,
|
|
first_failed_at=now,
|
|
last_failed_at=now,
|
|
)
|
|
)
|
|
session.commit()
|
|
|
|
with Session(throttle_db) as session:
|
|
session.add(
|
|
LoginThrottle(
|
|
key="192.168.1.1",
|
|
scope="ip",
|
|
failures=2,
|
|
first_failed_at=now,
|
|
last_failed_at=now,
|
|
)
|
|
)
|
|
with pytest.raises(IntegrityError):
|
|
session.commit()
|
|
|
|
|
|
def test_same_key_different_scope_allowed(throttle_db):
|
|
"""Same key string but different scopes are independent rows — both should insert."""
|
|
now = datetime.now(tz=timezone.utc)
|
|
with Session(throttle_db) as session:
|
|
session.add(
|
|
LoginThrottle(
|
|
key="admin",
|
|
scope="ip",
|
|
failures=1,
|
|
first_failed_at=now,
|
|
last_failed_at=now,
|
|
)
|
|
)
|
|
session.add(
|
|
LoginThrottle(
|
|
key="admin",
|
|
scope="user",
|
|
failures=3,
|
|
first_failed_at=now,
|
|
last_failed_at=now,
|
|
)
|
|
)
|
|
session.commit() # must not raise
|
|
|
|
with Session(throttle_db) as session:
|
|
rows = session.query(LoginThrottle).filter_by(key="admin").all()
|
|
assert len(rows) == 2
|
|
scopes = {r.scope for r in rows}
|
|
assert scopes == {"ip", "user"}
|
|
|
|
|
|
def test_next_allowed_at_nullable(throttle_db):
|
|
"""next_allowed_at is nullable — a row without it must persist fine."""
|
|
now = datetime.now(tz=timezone.utc)
|
|
with Session(throttle_db) as session:
|
|
row = LoginThrottle(
|
|
key="10.0.0.1",
|
|
scope="ip",
|
|
failures=1,
|
|
first_failed_at=now,
|
|
last_failed_at=now,
|
|
next_allowed_at=None,
|
|
)
|
|
session.add(row)
|
|
session.commit()
|
|
session.refresh(row)
|
|
assert row.next_allowed_at is None
|
|
|
|
|
|
def test_next_allowed_at_persists_when_set(throttle_db):
|
|
"""next_allowed_at can be set and retrieved correctly."""
|
|
now = datetime.now(tz=timezone.utc)
|
|
with Session(throttle_db) as session:
|
|
row = LoginThrottle(
|
|
key="10.0.0.2",
|
|
scope="ip",
|
|
failures=5,
|
|
first_failed_at=now,
|
|
last_failed_at=now,
|
|
next_allowed_at=now,
|
|
)
|
|
session.add(row)
|
|
session.commit()
|
|
row_id = row.id
|
|
|
|
with Session(throttle_db) as session:
|
|
fetched = session.get(LoginThrottle, row_id)
|
|
assert fetched is not None
|
|
assert fetched.next_allowed_at is not None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Downgrade / upgrade reversibility test
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_downgrade_removes_table(tmp_path: Path):
|
|
"""downgrade -1 from head must cleanly drop auth_login_throttle."""
|
|
db_path = tmp_path / "down_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})
|
|
assert "auth_login_throttle" in inspect(engine).get_table_names()
|
|
engine.dispose()
|
|
|
|
command.downgrade(alembic_cfg, "-1")
|
|
engine = create_engine(db_url, connect_args={"check_same_thread": False})
|
|
assert "auth_login_throttle" not in inspect(engine).get_table_names()
|
|
engine.dispose()
|
|
|
|
|
|
def test_index_exists_on_scope_key(throttle_db):
|
|
"""ix_auth_login_throttle_scope_key composite index must exist."""
|
|
inspector = inspect(throttle_db)
|
|
indexes = inspector.get_indexes("auth_login_throttle")
|
|
index_names = {idx["name"] for idx in indexes}
|
|
assert "ix_auth_login_throttle_scope_key" in index_names
|