chicken_babies_site/tests/test_admin_routes.py

"""End-to-end HTTP tests for the admin auth flow.

Exercises the real :class:`FastAPI` app constructed by
:func:`app.main.create_app` against a temp-file SQLite database and a
monkeypatched ``EmailService`` so we can capture the magic-link URL
without depending on Resend.

These tests intentionally do NOT use the session-scoped ``db_engine``
fixture — each builds its own fresh app so rate-limit state and DB
rows don't leak between tests.
"""

from __future__ import annotations

import importlib
import re
from pathlib import Path
from typing import Iterator

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import text


# Regex used to extract the raw token from the dev-fallback log URL.
_URL_RE = re.compile(r"http[s]?://[^\s]+/admin/auth/consume/[A-Za-z0-9_-]+")


def _csrf_from_response(resp) -> str:
    """Extract the CSRF token from a rendered admin page.

    The admin base template emits a ``<meta name="csrf-token">`` tag
    populated by the CSRFCookieMiddleware. Tests pull it from there
    rather than the cookie value directly so the assertion also
    proves the template wiring is intact.
    """
    match = re.search(
        r'<meta name="csrf-token" content="([^"]*)"', resp.text
    )
    assert match, "csrf-token meta tag missing"
    return match.group(1)


@pytest.fixture
def admin_app(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> Iterator[tuple[TestClient, dict]]:
    """Build a fresh FastAPI app wired to a tmp DB + captured email URL."""
    # Point the settings loader at a clean DB and a known allowlist.
    monkeypatch.setenv("DATABASE_URL", f"sqlite:///{tmp_path}/admin.db")
    monkeypatch.setenv("ADMIN_EMAILS", "headhen@example.com")
    monkeypatch.setenv("APP_ENV", "development")
    monkeypatch.setenv(
        "SECRET_KEY", "test-only-secret-key-0123456789abcdef-XYZ"
    )
    monkeypatch.setenv("RESEND_API_KEY", "")
    monkeypatch.setenv("PUBLIC_BASE_URL", "http://testserver")

    # Reload config + main so the new env is picked up.
    from app import config as _config

    _config.get_settings.cache_clear()

    import app.main as main_module

    importlib.reload(main_module)
    app = main_module.app

    # Capture the magic-link URL instead of sending via Resend.
    captured: dict[str, str] = {}

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

    app.state.email_service.send_magic_link = _capture  # type: ignore[assignment]

    # Reset the in-memory rate limiter between tests (module-level
    # singleton would otherwise leak state across tests).
    from app.services.rate_limit import limiter

    limiter.reset()

    with TestClient(app) as client:
        yield client, captured

    _config.get_settings.cache_clear()


def _auth_events(app, event_type: str | None = None) -> list[dict]:
    """Return rows from ``auth_events`` (optionally filtered by type)."""
    with app.state.engine.connect() as conn:
        if event_type:
            rows = conn.execute(
                text(
                    "SELECT event_type, email, detail FROM auth_events"
                    " WHERE event_type = :t ORDER BY id"
                ),
                {"t": event_type},
            ).mappings().all()
        else:
            rows = conn.execute(
                text(
                    "SELECT event_type, email, detail FROM auth_events"
                    " ORDER BY id"
                )
            ).mappings().all()
    return [dict(r) for r in rows]


def test_full_login_flow(admin_app) -> None:
    """GET login, POST, consume, GET /admin, POST logout."""
    client, captured = admin_app

    # 1. GET login page.
    resp = client.get("/admin/login")
    assert resp.status_code == 200
    assert "Admin log in" in resp.text
    assert 'name="email"' in resp.text

    # 2. POST login with allowlisted email.
    resp = client.post("/admin/login", data={"email": "headhen@example.com"})
    assert resp.status_code == 200
    assert "Check your inbox" in resp.text
    assert "url" in captured
    raw_url = captured["url"]
    assert "/admin/auth/consume/" in raw_url

    # 3. Consume the magic link.
    consume_path = raw_url.replace("http://testserver", "")
    resp = client.get(consume_path, follow_redirects=False)
    assert resp.status_code == 303
    assert resp.headers["location"] == "/admin"
    # cb_session cookie set.
    assert "cb_session" in resp.cookies

    # 4. GET /admin renders the Phase 4 dashboard.
    resp = client.get("/admin")
    assert resp.status_code == 200
    assert "Dashboard" in resp.text
    assert "headhen@example.com" in resp.text
    csrf_token = _csrf_from_response(resp)

    # 5. POST /admin/logout clears cookie + redirects. CSRF required
    # now that Phase 4 has enabled the double-submit cookie.
    resp = client.post(
        "/admin/logout",
        data={"csrf_token": csrf_token},
        follow_redirects=False,
    )
    assert resp.status_code == 303
    assert resp.headers["location"] == "/admin/login"

    # 6. Subsequent /admin redirects back to login.
    # Manually drop the cookie from the test client, mirroring a browser
    # that honored the delete_cookie directive.
    client.cookies.clear()
    resp = client.get("/admin", follow_redirects=False)
    assert resp.status_code == 303
    assert resp.headers["location"] == "/admin/login"

    # Audit rows: link_requested, link_consumed, session_created, session_revoked.
    events = [e["event_type"] for e in _auth_events(client.app)]
    for required in (
        "link_requested",
        "link_consumed",
        "session_created",
        "session_revoked",
    ):
        assert required in events, f"missing audit event: {required}"


def test_non_allowlisted_returns_same_page_no_token(admin_app) -> None:
    """Non-allowlisted email: same 200 + 'Check your inbox', no token row."""
    client, captured = admin_app
    resp = client.post("/admin/login", data={"email": "stranger@example.com"})
    assert resp.status_code == 200
    assert "Check your inbox" in resp.text
    assert "url" not in captured  # email not sent

    # No token row for this email.
    with client.app.state.engine.connect() as conn:
        count = conn.execute(
            text(
                "SELECT COUNT(*) AS c FROM magic_link_tokens"
                " WHERE email = :e"
            ),
            {"e": "stranger@example.com"},
        ).mappings().first()
    assert count is not None and int(count["c"]) == 0

    # Audit trail still records the attempt.
    events = _auth_events(client.app, "link_requested")
    assert len(events) == 1
    assert "\"allowlisted\": false" in events[0]["detail"]


def test_invalid_token_returns_400(admin_app) -> None:
    """A random token on the consume endpoint returns the generic 400 page."""
    client, _ = admin_app
    resp = client.get("/admin/auth/consume/not-a-real-token", follow_redirects=False)
    assert resp.status_code == 400
    assert "Login link invalid" in resp.text


def test_login_form_validation(admin_app) -> None:
    """Empty / malformed email surfaces the inline validation message."""
    client, _ = admin_app
    resp = client.post("/admin/login", data={"email": ""})
    assert resp.status_code == 400
    assert "valid email" in resp.text

    resp = client.post("/admin/login", data={"email": "not-an-email"})
    assert resp.status_code == 400
    assert "valid email" in resp.text


def test_replay_attack_fails(admin_app) -> None:
    """A consumed magic link cannot be used a second time."""
    client, captured = admin_app
    client.post("/admin/login", data={"email": "headhen@example.com"})
    url = captured["url"].replace("http://testserver", "")

    first = client.get(url, follow_redirects=False)
    assert first.status_code == 303

    # Clear cookies to simulate a replay from a fresh browser context.
    client.cookies.clear()
    second = client.get(url, follow_redirects=False)
    assert second.status_code == 400