From 3e81d60268edfb259256f26f6e789e9fbcf53388 Mon Sep 17 00:00:00 2001 From: coder-bot Date: Wed, 19 Aug 2026 22:03:00 +0000 Subject: [PATCH] WIP checkpoint: Implement /auth/refresh endpoint --- backend/tests/test_refresh_endpoint.py | 201 +++++++++++++++++++++++++ 1 file changed, 201 insertions(+) create mode 100644 backend/tests/test_refresh_endpoint.py diff --git a/backend/tests/test_refresh_endpoint.py b/backend/tests/test_refresh_endpoint.py new file mode 100644 index 0000000..995bf0f --- /dev/null +++ b/backend/tests/test_refresh_endpoint.py @@ -0,0 +1,201 @@ +"""Tests for the POST /api/auth/refresh endpoint. + +These tests exercise the full stack: register → get refresh token → +refresh → new access token. Requires migrations already applied +(alembic upgrade head) before running. +""" + +import pytest +from fastapi.testclient import TestClient +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 + + +# --------------------------------------------------------------------------- +# 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() + + +# --------------------------------------------------------------------------- +# Refresh endpoint tests +# --------------------------------------------------------------------------- + +class TestRefresh: + """Tests for POST /api/auth/refresh.""" + + def _register_user(self, username: str, password: str, email: str = None): + """Helper to register a user and return the refresh token.""" + client = TestClient(app) + if email is None: + email = f"{username}@example.com" + resp = client.post( + "/api/auth/register", + json={ + "username": username, + "email": email, + "password": password, + "date_of_birth": "2000-01-01", + }, + ) + assert resp.status_code == 201 + return resp.json()["refresh_token"] + + def test_refresh_success(self): + """A valid refresh token returns a new access token.""" + refresh_token = self._register_user("refreshuser", "secret123") + client = TestClient(app) + resp = client.post( + "/api/auth/refresh", + json={"refresh_token": refresh_token}, + ) + assert resp.status_code == 200 + data = resp.json() + assert "access_token" in data + assert data["token_type"] == "bearer" + # The refresh token in the response is the same one we sent + assert data["refresh_token"] == refresh_token + + def test_refresh_returns_different_access_token(self): + """Refreshing produces a new access token (different from any previous).""" + refresh_token = self._register_user("refreshtoken", "secret123") + client = TestClient(app) + + # First refresh + resp1 = client.post( + "/api/auth/refresh", + json={"refresh_token": refresh_token}, + ) + assert resp1.status_code == 200 + access_token_1 = resp1.json()["access_token"] + + # Second refresh + resp2 = client.post( + "/api/auth/refresh", + json={"refresh_token": refresh_token}, + ) + assert resp2.status_code == 200 + access_token_2 = resp2.json()["access_token"] + + # Each refresh should produce a new access token + assert access_token_1 != access_token_2 + + def test_refresh_invalid_token(self): + """An invalid (garbage) refresh token returns 401.""" + client = TestClient(app) + resp = client.post( + "/api/auth/refresh", + json={"refresh_token": "not.a.jwt.token"}, + ) + assert resp.status_code == 401 + assert "Invalid or expired refresh token" in resp.json()["detail"] + + def test_refresh_empty_body(self): + """Missing refresh_token field returns 422 validation error.""" + client = TestClient(app) + resp = client.post( + "/api/auth/refresh", + json={}, + ) + assert resp.status_code == 422 + + def test_refresh_missing_field(self): + """No body at all returns 422.""" + client = TestClient(app) + resp = client.post( + "/api/auth/refresh", + content="{}", + headers={"Content-Type": "application/json"}, + ) + assert resp.status_code == 422 + + def test_refresh_with_login_token(self): + """A refresh token obtained from login also works for /refresh.""" + client = TestClient(app) + # Register first + client.post( + "/api/auth/register", + json={ + "username": "loginrefuser", + "email": "loginref@example.com", + "password": "secret123", + "date_of_birth": "2000-01-01", + }, + ) + # Login to get a refresh token + login_resp = client.post( + "/api/auth/login", + data={"username": "loginrefuser", "password": "secret123"}, + ) + assert login_resp.status_code == 200 + refresh_token = login_resp.json()["refresh_token"] + + # Use it to refresh + resp = client.post( + "/api/auth/refresh", + json={"refresh_token": refresh_token}, + ) + assert resp.status_code == 200 + assert "access_token" in resp.json() + + def test_refresh_access_token_is_valid_jwt(self): + """The access token returned by /refresh is a valid JWT.""" + from jose import jwt + + refresh_token = self._register_user("jwttoken", "secret123") + client = TestClient(app) + resp = client.post( + "/api/auth/refresh", + json={"refresh_token": refresh_token}, + ) + assert resp.status_code == 200 + access_token = resp.json()["access_token"] + + # Verify it's a 3-part JWT + parts = access_token.split(".") + assert len(parts) == 3 + + # Verify it decodes with the public key + payload = jwt.decode( + access_token, + settings.public_pem, + algorithms=[settings.ALGORITHM], + ) + assert "sub" in payload + assert "exp" in payload + + def test_refresh_expires_minutes_in_response(self): + """The response includes access_token_expires_minutes.""" + refresh_token = self._register_user("expiretest", "secret123") + client = TestClient(app) + resp = client.post( + "/api/auth/refresh", + json={"refresh_token": refresh_token}, + ) + assert resp.status_code == 200 + assert "access_token_expires_minutes" in resp.json() + assert resp.json()["access_token_expires_minutes"] == settings.ACCESS_TOKEN_EXPIRE_MINUTES