chicken_babies_site/tests/test_auth_service.py

"""Tests for :class:`app.services.auth.AuthService`.

Covers:
- Allowlisted email → token inserted, email "sent" (dev fallback), audit
  rows emitted.
- Non-allowlisted email → no token row, audit row with allowlisted=False.
- Consume happy path (creates user on first login, bumps last_login_at
  on subsequent logins, issues session).
- Consume replay → second attempt returns None + consume_failed/used audit.
- Consume expired token → None + consume_failed/expired.
- Per-email rate-limit threshold trips on the 6th request.
"""

from __future__ import annotations

from datetime import datetime, timedelta, timezone

import pytest
from fastapi.templating import Jinja2Templates
from itsdangerous import URLSafeTimedSerializer
from pathlib import Path
from sqlalchemy import Engine, text

from app.config import Settings
from app.services.audit import AuditService
from app.services.auth import AuthService, RateLimitedError
from app.services.email import EmailService
from app.services.sessions import SessionService


_TEMPLATES_DIR = Path(__file__).resolve().parent.parent / "app" / "templates"


def _build(engine: Engine, *, admin_emails: str = "headhen@example.com") -> tuple[AuthService, Settings]:
    """Wire up an AuthService against a real SQLite engine."""
    settings = Settings(
        app_env="development",
        secret_key="a-very-long-test-secret-key-1234567890",
        admin_emails=admin_emails,
        resend_api_key=None,
        resend_from=None,
        public_base_url="http://localhost:8000",
        magic_link_ttl_min=15,
    )  # type: ignore[call-arg]
    templates = Jinja2Templates(directory=_TEMPLATES_DIR)
    signer = URLSafeTimedSerializer(settings.secret_key, salt="session")
    email = EmailService(settings, templates)
    audit = AuditService(engine)
    sessions = SessionService(engine, signer, settings)
    auth = AuthService(engine, email, sessions, audit, settings)
    return auth, settings


def _count(engine: Engine, table: str, where: str = "1=1", **params) -> int:
    with engine.connect() as conn:
        row = conn.execute(
            text(f"SELECT COUNT(*) AS c FROM {table} WHERE {where}"), params
        ).mappings().first()
    return int(row["c"]) if row is not None else 0


def _latest_detail(engine: Engine, event_type: str) -> str:
    with engine.connect() as conn:
        row = conn.execute(
            text(
                "SELECT detail FROM auth_events WHERE event_type = :t"
                " ORDER BY id DESC LIMIT 1"
            ),
            {"t": event_type},
        ).mappings().first()
    return str(row["detail"]) if row is not None else ""


def test_allowlisted_request_inserts_token_and_audit(
    clean_db_engine: Engine,
) -> None:
    """An allowlisted email mints a token row and a link_requested audit."""
    auth, _ = _build(clean_db_engine)
    auth.request_link(email="HeadHen@example.com", ip="1.2.3.4", user_agent="ua")

    assert _count(clean_db_engine, "magic_link_tokens") == 1
    # Stored email is lowercased.
    with clean_db_engine.connect() as conn:
        row = conn.execute(
            text("SELECT email, token_hash FROM magic_link_tokens")
        ).mappings().first()
    assert row["email"] == "headhen@example.com"
    # Hex SHA-256 is 64 chars.
    assert len(row["token_hash"]) == 64

    assert _count(clean_db_engine, "auth_events", "event_type='link_requested'") == 1
    assert "\"allowlisted\": true" in _latest_detail(
        clean_db_engine, "link_requested"
    )


def test_non_allowlisted_request_inserts_no_token(
    clean_db_engine: Engine,
) -> None:
    """A non-allowlisted email gets an audit row and no token."""
    auth, _ = _build(clean_db_engine, admin_emails="headhen@example.com")
    auth.request_link(email="intruder@example.com", ip="1.2.3.4", user_agent="ua")

    assert _count(clean_db_engine, "magic_link_tokens") == 0
    assert _count(clean_db_engine, "auth_events", "event_type='link_requested'") == 1
    assert "\"allowlisted\": false" in _latest_detail(
        clean_db_engine, "link_requested"
    )


def test_consume_happy_path_creates_user_and_session(
    clean_db_engine: Engine,
    monkeypatch,
) -> None:
    """Consuming a fresh token upserts a user and mints a session."""
    auth, _ = _build(clean_db_engine)

    # Intercept the raw token via the dev-fallback log. Easiest path:
    # patch EmailService.send_magic_link on this instance.
    captured: dict[str, str] = {}

    def _capture(**kw):
        captured["url"] = kw["url"]

    monkeypatch.setattr(auth._email, "send_magic_link", _capture)

    auth.request_link(email="headhen@example.com", ip="1.2.3.4", user_agent="ua")
    assert "url" in captured
    raw_token = captured["url"].rsplit("/", 1)[-1]

    result = auth.consume(
        raw_token=raw_token, ip="1.2.3.4", user_agent="ua"
    )
    assert result is not None
    user, session, cookie = result

    assert user.email == "headhen@example.com"
    assert user.display_name == "Headhen"
    assert user.active is True

    # User row persisted with last_login_at set.
    with clean_db_engine.connect() as conn:
        row = conn.execute(
            text("SELECT last_login_at FROM users WHERE id = :id"),
            {"id": user.id},
        ).mappings().first()
    assert row["last_login_at"] is not None

    # Token marked used.
    with clean_db_engine.connect() as conn:
        tok = conn.execute(
            text("SELECT used_at FROM magic_link_tokens")
        ).mappings().first()
    assert tok["used_at"] is not None

    # Session row exists, cookie is non-empty.
    assert session.user_id == user.id
    assert cookie

    # Audit trail: link_consumed + session_created.
    assert _count(
        clean_db_engine, "auth_events", "event_type='link_consumed'"
    ) == 1
    assert _count(
        clean_db_engine, "auth_events", "event_type='session_created'"
    ) == 1


def test_consume_replay_fails(clean_db_engine: Engine, monkeypatch) -> None:
    """The same raw token cannot be consumed twice."""
    auth, _ = _build(clean_db_engine)
    captured: dict[str, str] = {}
    monkeypatch.setattr(
        auth._email, "send_magic_link", lambda **kw: captured.update(url=kw["url"])
    )
    auth.request_link(email="headhen@example.com", ip="", user_agent="")
    raw = captured["url"].rsplit("/", 1)[-1]

    first = auth.consume(raw_token=raw, ip="", user_agent="")
    assert first is not None

    second = auth.consume(raw_token=raw, ip="", user_agent="")
    assert second is None

    # consume_failed audit with reason=used.
    assert "\"reason\": \"used\"" in _latest_detail(
        clean_db_engine, "consume_failed"
    )


def test_consume_expired_token_fails(clean_db_engine: Engine, monkeypatch) -> None:
    """A token past its expires_at cannot be consumed."""
    auth, _ = _build(clean_db_engine)
    captured: dict[str, str] = {}
    monkeypatch.setattr(
        auth._email, "send_magic_link", lambda **kw: captured.update(url=kw["url"])
    )
    auth.request_link(email="headhen@example.com", ip="", user_agent="")
    raw = captured["url"].rsplit("/", 1)[-1]

    # Push expires_at into the past.
    past = (datetime.now(timezone.utc) - timedelta(minutes=30)).isoformat()
    with clean_db_engine.begin() as conn:
        conn.execute(
            text("UPDATE magic_link_tokens SET expires_at = :p"), {"p": past}
        )

    assert auth.consume(raw_token=raw, ip="", user_agent="") is None
    assert "\"reason\": \"expired\"" in _latest_detail(
        clean_db_engine, "consume_failed"
    )


def test_consume_unknown_token_fails(clean_db_engine: Engine) -> None:
    """Random tokens result in not_found."""
    auth, _ = _build(clean_db_engine)
    assert auth.consume(
        raw_token="totally-random-nope", ip="", user_agent=""
    ) is None
    assert "\"reason\": \"not_found\"" in _latest_detail(
        clean_db_engine, "consume_failed"
    )


def test_per_email_rate_limit_trips_on_sixth(
    clean_db_engine: Engine, monkeypatch
) -> None:
    """Five requests in the window succeed; the sixth raises and audits."""
    auth, _ = _build(clean_db_engine)
    monkeypatch.setattr(auth._email, "send_magic_link", lambda **kw: None)

    for _ in range(5):
        auth.request_link(email="headhen@example.com", ip="", user_agent="")

    with pytest.raises(RateLimitedError):
        auth.request_link(email="headhen@example.com", ip="", user_agent="")

    # Exactly 5 tokens persisted; 6th blocked before insert.
    assert _count(clean_db_engine, "magic_link_tokens") == 5
    # rate_limited audit with scope=email.
    detail = _latest_detail(clean_db_engine, "rate_limited")
    assert "\"scope\": \"email\"" in detail


def test_existing_user_reactivated_on_consume(
    clean_db_engine: Engine, monkeypatch
) -> None:
    """Consume updates an existing (possibly deactivated) user row."""
    # Seed a deactivated existing admin.
    now_iso = datetime.now(timezone.utc).isoformat()
    with clean_db_engine.begin() as conn:
        conn.execute(
            text(
                "INSERT INTO users"
                " (email, display_name, created_at, last_login_at, active)"
                " VALUES (:e, :d, :c, NULL, 0)"
            ),
            {"e": "headhen@example.com", "d": "Old Name", "c": now_iso},
        )

    auth, _ = _build(clean_db_engine)
    captured: dict[str, str] = {}
    monkeypatch.setattr(
        auth._email, "send_magic_link", lambda **kw: captured.update(url=kw["url"])
    )
    auth.request_link(email="headhen@example.com", ip="", user_agent="")
    raw = captured["url"].rsplit("/", 1)[-1]

    result = auth.consume(raw_token=raw, ip="", user_agent="")
    assert result is not None
    user, _, _ = result

    # display_name preserved (we don't overwrite), active flipped back on.
    assert user.display_name == "Old Name"
    assert user.active is True