home-automation/tests/test_auth_throttle_model.py

"""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):
    """Downgrading the auth_login_throttle revision must cleanly drop the table.

    We upgrade to the specific throttle revision (not necessarily head, since later
    revisions may have been added), then downgrade one step to verify the table is gone.
    """
    db_path = tmp_path / "down_test.db"
    db_url = f"sqlite:///{db_path}"
    alembic_cfg = _make_app_alembic_config(db_url)
    throttle_revision = "20260621_07_auth_login_throttle"

    command.upgrade(alembic_cfg, throttle_revision)
    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