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

"""Auth endpoint tests, plus — the one that matters most — an empirical proof that the
repository-layer scope actually enforces isolation between users.

Per CLAUDE.md invariant #4: since D15 removed Postgres RLS, `Scope` in db.py is not one layer of
isolation among two — it's the only one. test_scoped_session_blocks_cross_user_reads below doesn't
trust that `Scope.select()` filters correctly because the code reads correctly; it proves it by
registering two real users and confirming a scoped read for user A never returns user B's row,
even when both exist in the same table.
"""

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

import httpx
import pytest
from sqlalchemy import select, 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
from velodrome.models import Invite, 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.

    UUID values are passed as `.hex` (32 hex chars, no hyphens), not `str()` (36 chars, hyphenated)
    or a raw UUID object — a `text()` query has no ORM-level column-type awareness, so this is
    fixture setup working around two separate things confirmed empirically, not assumed:
    (1) aiosqlite's driver has no built-in adapter for a Python `UUID` object at all (raises
    "type 'UUID' is not supported"); (2) SQLAlchemy's `Uuid(as_uuid=True)` column type stores as
    the 32-char hex form on SQLite, NOT the hyphenated `str()` form — inserting the hyphenated
    form via raw SQL left rows the ORM's own later `UPDATE ... WHERE id = ?` (implicitly issued
    by `invite.used_count += 1` in auth/service.py) could never match, silently updating zero
    rows. `.hex` is what the ORM itself writes, so raw-SQL-inserted rows are indistinguishable
    from ORM-inserted ones.

    `creator_id` is a fixed sentinel, not a fresh `uuid4()` per call — confirmed the hard way: a
    test calling this twice (two invite codes) with a fresh id each time but the same hardcoded
    seed email hit the `ON CONFLICT DO NOTHING` on email on the second call, which silently
    no-ops, leaving that call's fresh id never actually inserted — so its invite's `created_by`
    pointed at a user row that was never created, and the FK constraint on the invites insert
    failed. A stable id makes repeat calls genuinely idempotent instead of just quiet about it.
    """
    # datetime binds use .isoformat() explicitly, not a raw datetime object — Python's sqlite3
    # module has its own implicit datetime adapter, which is deprecated as of 3.12 and warns on
    # every use (confirmed against this exact fixture); the ORM's own UTCDateTime type never hits
    # this because SQLAlchemy's dialect handles the conversion itself rather than delegating to
    # sqlite3's adapter, but a bare text() bind param has no such handling.
    creator_id = UUID(int=0)
    await db_auth.execute(
        text(
            "INSERT INTO users (id, email, display_name, password_hash, created_at) "
            "VALUES (:id, 'seed@example.com', 'Seed', :ph, :now) ON CONFLICT DO NOTHING"
        ),
        {"id": creator_id.hex, "ph": hash_password("unused"), "now": datetime.now(UTC).isoformat()},
    )
    await db_auth.execute(
        text(
            "INSERT INTO invites (id, code_hash, created_by, expires_at, max_uses, used_count) "
            "VALUES (:id, :hash, :creator, :expires, 1, 0)"
        ),
        {
            "id": uuid4().hex,
            "hash": hash_invite_code(code),
            "creator": creator_id.hex,
            "expires": (datetime.now(UTC) + timedelta(days=1)).isoformat(),
        },
    )
    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_scoped_session_blocks_cross_user_reads(
    client: httpx.AsyncClient, db_auth: AsyncSession
) -> None:
    """The load-bearing test: prove the repository-layer scope actually enforces isolation, not
    just that `Scope.select()` reads like it should.

    Two users register (creating two session rows). We then open `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 `Scope.select(Session)` returns ONLY user A's row, even though both exist
    in the same table. If this test ever passes with more than one row, the scope is not isolating
    users and every other invariant in this codebase is resting on nothing (see CLAUDE.md
    invariant #4 — there is no database-level backstop anymore; this IS the isolation boundary).
    """
    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 pre-identity bootstrap path.
    total = (await db_auth.execute(text("SELECT count(*) FROM sessions"))).scalar_one()
    assert total == 2
    # Close out the transaction this read just autobegan. db.py's BEGIN IMMEDIATE (see its
    # docstring) takes SQLite's exclusive write lock the instant ANY transaction starts, read
    # included — confirmed the hard way: without this commit, the db_auth fixture's session
    # (which the pytest fixture keeps open for the whole test, not just this block) holds that
    # lock indefinitely, and the scoped_session() calls below then fail with "database is
    # locked" trying to acquire their own. Every write in this file already commits promptly;
    # a read needs the same discipline once BEGIN IMMEDIATE is in play.
    await db_auth.commit()

    # Now the real test: Scope.select(), which is the only way application code is meant to
    # query a user-owned table once identity is known.
    async with scoped_session(user_a_id) as scope:
        rows = (await scope.session.execute(scope.select(Session))).scalars().all()

    assert len(rows) == 1, (
        f"expected exactly 1 row (user A's own session), got {len(rows)} — "
        "the repository-layer scope is not isolating users"
    )
    assert rows[0].user_id == 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 scope:
        rows_b = (await scope.session.execute(scope.select(Session))).scalars().all()
    assert len(rows_b) == 1
    assert rows_b[0].user_id == user_b_id


async def test_scope_select_rejects_models_without_user_id() -> None:
    """`Scope.select()` refuses to build a query for a model that has no `user_id` column,
    instead of silently returning an unfiltered (and therefore unsafe) query. `Invite` is a real
    example, not a contrived one: it's scoped by `created_by`, not `user_id`."""
    async with scoped_session(uuid4()) as scope:
        with pytest.raises(TypeError, match="has no user_id column"):
            scope.select(Invite)


async def test_unscoped_session_can_see_every_user_when_misused(
    client: httpx.AsyncClient, db_auth: AsyncSession
) -> None:
    """The deliberately alarming counterpart to the isolation test above: `unscoped_session()`
    plus a raw `select(Session)` returns EVERY user's rows, with no filtering at all — there is no
    database-level backstop to catch this mistake anymore (docs/DECISIONS.md D15). This is exactly
    why CLAUDE.md invariant #4 treats reaching for `unscoped_session()` on a user-owned table as a
    review-blocking mistake, not a style preference: the code review IS the isolation boundary
    for this specific failure mode, `Scope` is the boundary for the query-construction one.
    """
    await _seed_invite(db_auth, code="CODE-FOR-A")
    await _seed_invite(db_auth, code="CODE-FOR-B")
    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",
        },
    )
    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",
        },
    )

    async with unscoped_session() as db:
        rows = (await db.execute(select(Session))).scalars().all()
    assert len(rows) == 2, (
        "unscoped_session with a raw query sees every user's rows — by design, this is the "
        "unsafe path Scope exists to replace"
    )