bike-app/apps/api/tests/test_auth.py

"""Auth endpoint tests, plus — the one that matters most — an empirical proof of RLS isolation.

Per CLAUDE.md: 'auth/, RLS policies — security, and a mistake exposes another user's data.' The
whole point of test_rls_blocks_cross_user_session_reads below is that it doesn't trust the SQL in
the migration is correct because it reads correctly — it proves it by actually trying to read
another user's row through the RLS-scoped role and confirming zero rows come back.
"""

from datetime import UTC, datetime, timedelta
from uuid import UUID

import httpx
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession

from velodrome.auth.security import hash_invite_code, hash_password
from velodrome.db import scoped_session, unscoped_session

_REGISTER_PASSWORD = "correct horse battery staple"


async def _seed_invite(db_auth: AsyncSession, *, code: str = "TESTCODE123") -> UUID:
    """Insert a usable invite directly, bypassing the API — this is fixture setup, not the
    thing under test."""
    creator_id = UUID(int=0)
    await db_auth.execute(
        text(
            "INSERT INTO users (id, email, display_name, password_hash) "
            "VALUES (:id, 'seed@example.com', 'Seed', :ph) ON CONFLICT DO NOTHING"
        ),
        {"id": str(creator_id), "ph": hash_password("unused")},
    )
    await db_auth.execute(
        text(
            "INSERT INTO invites (id, code_hash, created_by, expires_at, max_uses, used_count) "
            "VALUES (gen_random_uuid(), :hash, :creator, :expires, 1, 0)"
        ),
        {
            "hash": hash_invite_code(code),
            "creator": str(creator_id),
            "expires": datetime.now(UTC) + timedelta(days=1),
        },
    )
    await db_auth.commit()
    return creator_id


async def test_register_with_valid_invite_creates_session(
    client: httpx.AsyncClient, db_auth: AsyncSession
) -> None:
    await _seed_invite(db_auth)
    resp = await client.post(
        "/api/v1/auth/register",
        json={
            "email": "rider@example.com",
            "password": _REGISTER_PASSWORD,
            "display_name": "Rider",
            "invite_code": "TESTCODE123",
        },
    )
    assert resp.status_code == 201, resp.text
    assert resp.json()["email"] == "rider@example.com"
    assert "vd_session" in resp.cookies


async def test_register_rejects_unknown_invite_code(client: httpx.AsyncClient) -> None:
    resp = await client.post(
        "/api/v1/auth/register",
        json={
            "email": "rider@example.com",
            "password": _REGISTER_PASSWORD,
            "display_name": "Rider",
            "invite_code": "no-such-code",
        },
    )
    assert resp.status_code == 400


async def test_register_rejects_reused_invite(
    client: httpx.AsyncClient, db_auth: AsyncSession
) -> None:
    await _seed_invite(db_auth)
    body = {
        "password": _REGISTER_PASSWORD,
        "display_name": "Rider",
        "invite_code": "TESTCODE123",
    }
    first = await client.post("/api/v1/auth/register", json={**body, "email": "first@example.com"})
    assert first.status_code == 201

    second = await client.post(
        "/api/v1/auth/register", json={**body, "email": "second@example.com"}
    )
    assert second.status_code == 400


async def test_login_with_correct_password_succeeds(
    client: httpx.AsyncClient, db_auth: AsyncSession
) -> None:
    await _seed_invite(db_auth)
    await client.post(
        "/api/v1/auth/register",
        json={
            "email": "rider@example.com",
            "password": _REGISTER_PASSWORD,
            "display_name": "Rider",
            "invite_code": "TESTCODE123",
        },
    )
    resp = await client.post(
        "/api/v1/auth/login",
        json={"email": "rider@example.com", "password": _REGISTER_PASSWORD},
    )
    assert resp.status_code == 200
    assert "vd_session" in resp.cookies


async def test_login_with_wrong_password_fails(
    client: httpx.AsyncClient, db_auth: AsyncSession
) -> None:
    await _seed_invite(db_auth)
    await client.post(
        "/api/v1/auth/register",
        json={
            "email": "rider@example.com",
            "password": _REGISTER_PASSWORD,
            "display_name": "Rider",
            "invite_code": "TESTCODE123",
        },
    )
    resp = await client.post(
        "/api/v1/auth/login",
        json={"email": "rider@example.com", "password": "wrong password entirely"},
    )
    assert resp.status_code == 401


async def test_me_requires_authentication(client: httpx.AsyncClient) -> None:
    resp = await client.get("/api/v1/auth/me")
    assert resp.status_code == 401


async def test_me_returns_current_user_with_valid_session(
    client: httpx.AsyncClient, db_auth: AsyncSession
) -> None:
    await _seed_invite(db_auth)
    await client.post(
        "/api/v1/auth/register",
        json={
            "email": "rider@example.com",
            "password": _REGISTER_PASSWORD,
            "display_name": "Rider",
            "invite_code": "TESTCODE123",
        },
    )
    resp = await client.get("/api/v1/auth/me")
    assert resp.status_code == 200
    assert resp.json()["email"] == "rider@example.com"


async def test_logout_revokes_the_session(client: httpx.AsyncClient, db_auth: AsyncSession) -> None:
    await _seed_invite(db_auth)
    await client.post(
        "/api/v1/auth/register",
        json={
            "email": "rider@example.com",
            "password": _REGISTER_PASSWORD,
            "display_name": "Rider",
            "invite_code": "TESTCODE123",
        },
    )
    logout_resp = await client.post(
        "/api/v1/auth/logout", headers={"Origin": "http://localhost:5173"}
    )
    assert logout_resp.status_code == 204

    me_resp = await client.get("/api/v1/auth/me")
    assert me_resp.status_code == 401


async def test_logout_without_matching_origin_is_rejected(
    client: httpx.AsyncClient, db_auth: AsyncSession
) -> None:
    await _seed_invite(db_auth)
    await client.post(
        "/api/v1/auth/register",
        json={
            "email": "rider@example.com",
            "password": _REGISTER_PASSWORD,
            "display_name": "Rider",
            "invite_code": "TESTCODE123",
        },
    )
    resp = await client.post("/api/v1/auth/logout", headers={"Origin": "https://evil.example"})
    assert resp.status_code == 403


async def test_rls_blocks_cross_user_session_reads(
    client: httpx.AsyncClient, db_auth: AsyncSession
) -> None:
    """The load-bearing test: prove RLS actually enforces isolation, not just that the migration
    ran without a syntax error.

    Two users register (creating two session rows). We then open a `scoped_session` as user A —
    the exact code path every real request handler will use once Phase 1 adds user-owned domain
    tables — and confirm the raw SQL result set contains ONLY user A's session, even though it
    runs no WHERE clause on user_id at all. If this test ever passes with more than one row, RLS
    is not doing its job and every other invariant in this codebase is resting on nothing.
    """
    await _seed_invite(db_auth, code="CODE-FOR-A")
    await _seed_invite(db_auth, code="CODE-FOR-B")

    resp_a = await client.post(
        "/api/v1/auth/register",
        json={
            "email": "user-a@example.com",
            "password": _REGISTER_PASSWORD,
            "display_name": "User A",
            "invite_code": "CODE-FOR-A",
        },
    )
    user_a_id = UUID(resp_a.json()["id"])

    resp_b = await client.post(
        "/api/v1/auth/register",
        json={
            "email": "user-b@example.com",
            "password": _REGISTER_PASSWORD,
            "display_name": "User B",
            "invite_code": "CODE-FOR-B",
        },
    )
    user_b_id = UUID(resp_b.json()["id"])
    assert user_a_id != user_b_id

    # Sanity check first: BOTH sessions genuinely exist, via the bypass role.
    total = (await db_auth.execute(text("SELECT count(*) FROM sessions"))).scalar_one()
    assert total == 2

    # Now the real test: as user A, scoped through the RLS-subject role, with NO WHERE clause.
    async with scoped_session(user_a_id) as scoped_db:
        rows = (await scoped_db.execute(text("SELECT user_id FROM sessions"))).scalars().all()

    assert len(rows) == 1, (
        f"expected exactly 1 row (user A's own session) via RLS, got {len(rows)} — "
        "RLS is not isolating users"
    )
    assert UUID(str(rows[0])) == user_a_id

    # And the mirror image, as user B, proving this isn't a coincidence of row ordering.
    async with scoped_session(user_b_id) as scoped_db:
        rows_b = (await scoped_db.execute(text("SELECT user_id FROM sessions"))).scalars().all()
    assert len(rows_b) == 1
    assert UUID(str(rows_b[0])) == user_b_id


async def test_unscoped_session_sees_zero_rows_of_user_owned_tables(
    client: httpx.AsyncClient, db_auth: AsyncSession
) -> None:
    """The default-deny half of the same proof: with app.user_id unset entirely (the state the
    health check and any other unauthenticated code path runs in), RLS denies everything."""
    await _seed_invite(db_auth)
    await client.post(
        "/api/v1/auth/register",
        json={
            "email": "rider@example.com",
            "password": _REGISTER_PASSWORD,
            "display_name": "Rider",
            "invite_code": "TESTCODE123",
        },
    )

    async with unscoped_session() as db:
        rows = (await db.execute(text("SELECT * FROM sessions"))).all()
    assert rows == []