modbison/backend/tests/test_auth_expired_token.py

"""Tests for expired-token rejection across auth endpoints.

Covers two distinct expiry paths:

1. An *access* JWT whose ``exp`` claim is in the past must be rejected
   with 401 by ``GET /api/auth/me`` (jose's built-in ``exp`` validation).
2. A *refresh* token whose persisted ``expires_at`` is in the past must be
   rejected with 401 by ``POST /api/auth/refresh`` even though its JWT
   signature is still valid.

Requires migrations already applied (alembic upgrade head) before running.
"""

import asyncio
from datetime import datetime, timedelta, timezone

import pytest
from fastapi.testclient import TestClient
from jose import jwt
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.pool import NullPool

from app.config import settings
from app.main import app
from app.database import get_db
from app.models.session import RefreshToken


# ---------------------------------------------------------------------------
# Test fixtures
# ---------------------------------------------------------------------------

@pytest.fixture(autouse=True)
def _override_db():
    """Override get_db so the app uses a fresh session per test."""
    test_engine = create_async_engine(settings.DATABASE_URL, poolclass=NullPool)
    test_session_factory = async_sessionmaker(test_engine, expire_on_commit=False)

    async def _get_db():
        async_session = test_session_factory()
        try:
            yield async_session
            await async_session.commit()
        except Exception:
            await async_session.rollback()
            raise
        finally:
            await async_session.close()

    app.dependency_overrides[get_db] = _get_db
    yield
    app.dependency_overrides.clear()


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

def _mint_access_token(exp: datetime) -> str:
    """Mint an RS256 access token with an explicit ``exp`` claim."""
    return jwt.encode(
        {"sub": "1", "exp": exp},
        settings.private_pem,
        algorithm=settings.ALGORITHM,
    )


def _expire_refresh_token_in_db(token: str) -> None:
    """Set the persisted refresh token's ``expires_at`` into the past."""

    async def _do():
        engine = create_async_engine(settings.DATABASE_URL, poolclass=NullPool)
        session_factory = async_sessionmaker(engine, expire_on_commit=False)
        async with session_factory() as session:
            from sqlalchemy import update

            await session.execute(
                update(RefreshToken)
                .where(RefreshToken.token == token)
                .values(expires_at=datetime.now(timezone.utc) - timedelta(hours=1))
            )
            await session.commit()
        await engine.dispose()

    asyncio.run(_do())


def _register_user(client: TestClient, username: str) -> str:
    """Register a user and return its refresh token."""
    resp = client.post(
        "/api/auth/register",
        json={
            "username": username,
            "email": f"{username}@example.com",
            "password": "secret123",
            "date_of_birth": "1990-01-01",
        },
    )
    assert resp.status_code == 201
    return resp.json()["refresh_token"]


# ---------------------------------------------------------------------------
# Expired access token (GET /api/auth/me)
# ---------------------------------------------------------------------------

class TestExpiredAccessToken:
    def test_expired_access_token_is_rejected_by_me(self):
        """An access JWT with a past exp claim returns 401 on /me."""
        client = TestClient(app)
        expired_token = _mint_access_token(
            datetime.now(timezone.utc) - timedelta(minutes=5)
        )
        resp = client.get(
            "/api/auth/me",
            headers={"Authorization": f"Bearer {expired_token}"},
        )
        assert resp.status_code == 401
        assert "Could not validate credentials" in resp.json()["detail"]

    def test_garbage_access_token_is_rejected_by_me(self):
        """A non-JWT bearer token returns 401 on /me."""
        client = TestClient(app)
        resp = client.get(
            "/api/auth/me",
            headers={"Authorization": "Bearer not-a-real-jwt"},
        )
        assert resp.status_code == 401

    def test_missing_bearer_token_returns_401(self):
        """No Authorization header at all returns 401 on /me."""
        client = TestClient(app)
        resp = client.get("/api/auth/me")
        assert resp.status_code == 401


# ---------------------------------------------------------------------------
# Expired refresh token (POST /api/auth/refresh)
# ---------------------------------------------------------------------------

class TestExpiredRefreshToken:
    def test_expired_refresh_token_is_rejected(self):
        """A refresh token whose DB expires_at is in the past returns 401."""
        client = TestClient(app)
        refresh_token = _register_user(client, "expired_rt_user")

        # Force the persisted record to be expired while its JWT signature
        # is still valid.
        _expire_refresh_token_in_db(refresh_token)

        resp = client.post(
            "/api/auth/refresh",
            json={"refresh_token": refresh_token},
        )
        assert resp.status_code == 401
        assert "Invalid or expired refresh token" in resp.json()["detail"]

    def test_expired_refresh_token_jwt_is_rejected(self):
        """A refresh JWT with a past exp claim is rejected before DB lookup."""
        client = TestClient(app)
        expired_jwt = jwt.encode(
            {"sub": "1", "exp": datetime.now(timezone.utc) - timedelta(minutes=5)},
            settings.private_pem,
            algorithm=settings.ALGORITHM,
        )
        resp = client.post(
            "/api/auth/refresh",
            json={"refresh_token": expired_jwt},
        )
        assert resp.status_code == 401
        assert "Invalid or expired refresh token" in resp.json()["detail"]