WIP checkpoint: Write auth endpoint tests (pytest)

backend/tests/test_auth_expired_token.py

@@ -0,0 +1,171 @@
+"""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": "2000-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"]