kite/backend/tests/test_auth_deps.py

"""
Tests for backend/deps/auth.py — FastAPI dependency chain.

Tests verify:
  - get_current_user returns the User ORM object when a valid Bearer token is provided
  - get_current_user raises HTTP 401 for expired/tampered tokens
  - get_current_user raises HTTP 401 when user.is_active is False
  - get_current_admin raises HTTP 403 when user.role == "user"
  - get_current_admin returns user when user.role == "admin"

Phase 7.2 additions (Wave 0 scaffolding):
  - FakeRedis attached to app.state.redis so Wave 1 NBF check in get_current_user
    does not raise AttributeError: 'State' object has no attribute 'redis'
  - Three xfail stubs for NBF check behaviours (reject, allow, fail-open)
"""
import uuid

import pytest
import pytest_asyncio
from httpx import ASGITransport, AsyncClient
from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import AsyncSession

from tests.test_auth_api import FakeRedis
from tests.conftest import _TEST_USER_AGENT


# ── Minimal test app with /test/me and /test/admin routes ─────────────────────

def make_test_app():
    """Create a minimal FastAPI app that exercises the auth deps.

    Phase 7.2: app.state.redis is set to a FakeRedis() instance so that the
    NBF check in get_current_user (Wave 1) can call request.app.state.redis
    without raising AttributeError (Pitfall 4 from Phase 7.2 RESEARCH.md).
    """
    from deps.auth import get_current_user, get_current_admin
    from db.models import User

    test_app = FastAPI()

    @test_app.get("/test/me")
    async def get_me(current_user: User = Depends(get_current_user)):
        return {"id": str(current_user.id), "role": current_user.role}

    @test_app.get("/test/admin")
    async def admin_only(_admin: User = Depends(get_current_admin)):
        return {"role": _admin.role}

    # Phase 7.2: mount FakeRedis so NBF check in get_current_user works
    test_app.state.redis = FakeRedis()

    return test_app


@pytest_asyncio.fixture
async def auth_client(db_session: AsyncSession):
    """Async HTTP test client for the auth-dep test app.

    Phase 7.2: each invocation of make_test_app() creates a fresh FakeRedis
    instance (instantiated inside make_test_app). The NBF check in
    get_current_user reads request.app.state.redis — this fixture provides it.
    """
    from deps.db import get_db

    app = make_test_app()
    app.dependency_overrides[get_db] = lambda: db_session

    # Phase 7.4: send _TEST_USER_AGENT so fgp check in get_current_user succeeds
    # for tokens created with the same user-agent below.
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test",
        headers={"User-Agent": _TEST_USER_AGENT},
    ) as c:
        yield c

    app.dependency_overrides.clear()


async def _create_user(db_session, role: str = "user", is_active: bool = True):
    """Helper: insert a minimal User row into the test DB."""
    from db.models import User
    from services.auth import hash_password

    user = User(
        id=uuid.uuid4(),
        handle=f"user_{uuid.uuid4().hex[:6]}",
        email=f"{uuid.uuid4().hex[:6]}@example.com",
        password_hash=hash_password("testpassword"),
        role=role,
        is_active=is_active,
    )
    db_session.add(user)
    await db_session.commit()
    return user


# ── Tests ────────────────────────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_get_current_user_returns_user(auth_client, db_session):
    """A valid Bearer token should return the user's id and role."""
    from services.auth import create_access_token

    user = await _create_user(db_session, role="user")
    token = create_access_token(str(user.id), "user", user_agent=_TEST_USER_AGENT)

    resp = await auth_client.get(
        "/test/me", headers={"Authorization": f"Bearer {token}"}
    )
    assert resp.status_code == 200
    data = resp.json()
    assert data["id"] == str(user.id)
    assert data["role"] == "user"


@pytest.mark.asyncio
async def test_get_current_user_rejects_tampered_token(auth_client):
    """A tampered Bearer token should return 401."""
    tampered = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJmYWtlIn0.invalidsig"
    resp = await auth_client.get(
        "/test/me", headers={"Authorization": f"Bearer {tampered}"}
    )
    assert resp.status_code == 401


@pytest.mark.asyncio
async def test_get_current_user_rejects_inactive_user(auth_client, db_session):
    """A valid token for an inactive user should return 401."""
    from services.auth import create_access_token

    user = await _create_user(db_session, role="user", is_active=False)
    token = create_access_token(str(user.id), "user", user_agent=_TEST_USER_AGENT)

    resp = await auth_client.get(
        "/test/me", headers={"Authorization": f"Bearer {token}"}
    )
    assert resp.status_code == 401


@pytest.mark.asyncio
async def test_get_current_admin_rejects_non_admin(auth_client, db_session):
    """A regular user (role='user') should get 403 on admin endpoint."""
    from services.auth import create_access_token

    user = await _create_user(db_session, role="user")
    token = create_access_token(str(user.id), "user", user_agent=_TEST_USER_AGENT)

    resp = await auth_client.get(
        "/test/admin", headers={"Authorization": f"Bearer {token}"}
    )
    assert resp.status_code == 403
    assert "Admin" in resp.json()["detail"]


@pytest.mark.asyncio
async def test_get_current_admin_allows_admin(auth_client, db_session):
    """An admin user (role='admin') should get 200 on admin endpoint."""
    from services.auth import create_access_token

    admin_user = await _create_user(db_session, role="admin")
    token = create_access_token(str(admin_user.id), "admin", user_agent=_TEST_USER_AGENT)

    resp = await auth_client.get(
        "/test/admin", headers={"Authorization": f"Bearer {token}"}
    )
    assert resp.status_code == 200
    assert resp.json()["role"] == "admin"


@pytest.mark.asyncio
async def test_get_current_user_missing_token(auth_client):
    """Missing Authorization header should return 401 or 403 (HTTPBearer default)."""
    resp = await auth_client.get("/test/me")
    assert resp.status_code in (401, 403)  # HTTPBearer raises 403 in older fastapi, 401 in newer


def test_deps_auth_has_http_403():
    """deps/auth.py must contain an HTTPException(status.HTTP_403_FORBIDDEN) check."""
    import re
    import os
    path = os.path.join(os.path.dirname(__file__), "..", "deps", "auth.py")
    with open(path) as f:
        source = f.read()
    assert re.search(r"HTTP_403_FORBIDDEN", source), \
        "deps/auth.py must raise HTTP 403 for non-admin access"


# ── Phase 7.2 Wave 0 stubs: NBF check behaviours ─────────────────────────────
# These three tests are xfailed with strict=False. When Wave 1 (Plan 02) adds
# the NBF check to get_current_user, the stubs are promoted to real assertions
# by replacing `assert False, "stub"` with the actual assertion logic.


@pytest.mark.asyncio
async def test_get_current_user_rejects_token_when_iat_before_user_nbf(auth_client, db_session):
    """Token with iat < user_nbf in Redis should return 401 'Session invalidated'."""
    from services.auth import create_access_token
    import time

    user = await _create_user(db_session, role="user")
    token = create_access_token(str(user.id), "user", user_agent=_TEST_USER_AGENT)

    # Pre-populate Redis with a future nbf (token's iat will be < this value)
    future_ts = int(time.time()) + 3600
    fake_redis = auth_client._transport.app.state.redis
    await fake_redis.set(f"user_nbf:{user.id}", str(future_ts).encode())

    resp = await auth_client.get(
        "/test/me", headers={"Authorization": f"Bearer {token}"}
    )
    assert resp.status_code == 401
    assert resp.json()["detail"] == "Session invalidated"
    assert resp.headers.get("WWW-Authenticate") == "Bearer"


@pytest.mark.asyncio
async def test_get_current_user_allows_token_when_iat_after_user_nbf(auth_client, db_session):
    """Token with iat > user_nbf in Redis should return 200 (token is valid)."""
    from services.auth import create_access_token
    import time

    user = await _create_user(db_session, role="user")
    token = create_access_token(str(user.id), "user", user_agent=_TEST_USER_AGENT)

    # Pre-populate Redis with a past nbf (token's iat will be > this value)
    past_ts = int(time.time()) - 3600
    fake_redis = auth_client._transport.app.state.redis
    await fake_redis.set(f"user_nbf:{user.id}", str(past_ts).encode())

    resp = await auth_client.get(
        "/test/me", headers={"Authorization": f"Bearer {token}"}
    )
    assert resp.status_code == 200


@pytest.mark.asyncio
async def test_get_current_user_failopen_on_redis_error(auth_client, db_session):
    """Redis error during NBF check must fail-open (return 200, not crash)."""
    from services.auth import create_access_token

    class _BrokenRedis:
        async def get(self, key):
            raise RuntimeError("simulated redis down")

    user = await _create_user(db_session, role="user")
    token = create_access_token(str(user.id), "user", user_agent=_TEST_USER_AGENT)

    # Override app.state.redis with a broken redis for this test
    auth_client._transport.app.state.redis = _BrokenRedis()

    resp = await auth_client.get(
        "/test/me", headers={"Authorization": f"Bearer {token}"}
    )
    # D-04: Redis outage must fail-open; request proceeds normally
    assert resp.status_code == 200