modbison/backend/tests/test_refresh_endpoint.py

"""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