M4-T01: add auth_login_throttle table + LoginThrottle model

tests/test_auth_throttle_model.py

@@ -0,0 +1,195 @@
+"""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