7fcb86ff51
- Add UserRepository and LinkedAccountRepository protocols to protocols.py - Add UserEntry and LinkedAccountEntry DTOs for service layer decoupling - Implement PostgresUserRepository and PostgresLinkedAccountRepository - Refactor UserService to use constructor-injected repositories - Add get_user_service factory and UserServiceDep to API deps - Update auth.py and users.py endpoints to use UserServiceDep - Rewrite tests to use FastAPI dependency overrides (no monkey patching) This follows the established repository pattern used by DeckService and CollectionService, enabling future offline fork support. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
293 lines
10 KiB
Python
293 lines
10 KiB
Python
"""Tests for auth API endpoints.
|
|
|
|
Tests the authentication endpoints including OAuth redirects,
|
|
token refresh, and logout.
|
|
|
|
Uses FastAPI's dependency override pattern for proper dependency injection testing.
|
|
"""
|
|
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
from uuid import UUID
|
|
|
|
import pytest
|
|
from fastapi import status
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.api.deps import get_user_service
|
|
|
|
|
|
class TestGoogleAuthRedirect:
|
|
"""Tests for GET /api/auth/google endpoint."""
|
|
|
|
def test_returns_501_when_not_configured(self, client: TestClient):
|
|
"""Test that endpoint returns 501 when Google OAuth is not configured.
|
|
|
|
Without client credentials, OAuth flow cannot proceed.
|
|
"""
|
|
with patch("app.api.auth.google_oauth") as mock_oauth:
|
|
mock_oauth.is_configured.return_value = False
|
|
|
|
response = client.get(
|
|
"/api/auth/google",
|
|
params={"redirect_uri": "http://localhost/callback"},
|
|
follow_redirects=False,
|
|
)
|
|
|
|
assert response.status_code == status.HTTP_501_NOT_IMPLEMENTED
|
|
assert "not configured" in response.json()["detail"]
|
|
|
|
|
|
class TestDiscordAuthRedirect:
|
|
"""Tests for GET /api/auth/discord endpoint."""
|
|
|
|
def test_returns_501_when_not_configured(self, client: TestClient):
|
|
"""Test that endpoint returns 501 when Discord OAuth is not configured."""
|
|
with patch("app.api.auth.discord_oauth") as mock_oauth:
|
|
mock_oauth.is_configured.return_value = False
|
|
|
|
response = client.get(
|
|
"/api/auth/discord",
|
|
params={"redirect_uri": "http://localhost/callback"},
|
|
follow_redirects=False,
|
|
)
|
|
|
|
assert response.status_code == status.HTTP_501_NOT_IMPLEMENTED
|
|
assert "not configured" in response.json()["detail"]
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_user_service_instance():
|
|
"""Create a mock UserService for dependency injection.
|
|
|
|
Returns a MagicMock with async methods configured.
|
|
"""
|
|
mock = MagicMock()
|
|
mock.get_by_id = AsyncMock()
|
|
mock.get_by_email = AsyncMock()
|
|
mock.get_by_oauth = AsyncMock()
|
|
mock.get_or_create_from_oauth = AsyncMock()
|
|
mock.update_last_login = AsyncMock()
|
|
return mock
|
|
|
|
|
|
class TestRefreshTokens:
|
|
"""Tests for POST /api/auth/refresh endpoint."""
|
|
|
|
def test_returns_new_access_token(
|
|
self,
|
|
app,
|
|
client: TestClient,
|
|
test_user,
|
|
refresh_token_data,
|
|
mock_get_redis,
|
|
mock_user_service_instance,
|
|
):
|
|
"""Test that refresh endpoint returns new access token for valid refresh token.
|
|
|
|
A valid, non-revoked refresh token should yield a new access token.
|
|
"""
|
|
# Store the refresh token in fake Redis
|
|
import asyncio
|
|
|
|
async def setup_token():
|
|
async with mock_get_redis() as redis:
|
|
key = f"refresh_token:{refresh_token_data['user_id']}:{refresh_token_data['jti']}"
|
|
await redis.setex(key, 86400, "1")
|
|
|
|
asyncio.get_event_loop().run_until_complete(setup_token())
|
|
|
|
# Configure mock to return test user
|
|
# Convert to UserEntry-like object
|
|
mock_user_entry = MagicMock()
|
|
mock_user_entry.id = test_user.id
|
|
mock_user_entry.email = test_user.email
|
|
mock_user_service_instance.get_by_id.return_value = mock_user_entry
|
|
|
|
# Override the dependency on the test app
|
|
app.dependency_overrides[get_user_service] = lambda: mock_user_service_instance
|
|
|
|
try:
|
|
response = client.post(
|
|
"/api/auth/refresh",
|
|
json={"refresh_token": refresh_token_data["token"]},
|
|
)
|
|
|
|
assert response.status_code == status.HTTP_200_OK
|
|
data = response.json()
|
|
assert "access_token" in data
|
|
assert data["refresh_token"] == refresh_token_data["token"]
|
|
assert data["token_type"] == "bearer"
|
|
assert "expires_in" in data
|
|
finally:
|
|
# Clean up override
|
|
app.dependency_overrides.pop(get_user_service, None)
|
|
|
|
def test_returns_401_for_invalid_token(self, client: TestClient):
|
|
"""Test that refresh endpoint returns 401 for invalid refresh token."""
|
|
response = client.post(
|
|
"/api/auth/refresh",
|
|
json={"refresh_token": "invalid.token.here"},
|
|
)
|
|
|
|
assert response.status_code == status.HTTP_401_UNAUTHORIZED
|
|
|
|
def test_returns_401_for_revoked_token(
|
|
self, client: TestClient, refresh_token_data, mock_get_redis
|
|
):
|
|
"""Test that refresh endpoint returns 401 for revoked token.
|
|
|
|
A refresh token not in Redis (revoked/expired) should be rejected.
|
|
"""
|
|
# Don't store the token in Redis - simulating revocation
|
|
# The mock_get_redis is already patched via conftest's app fixture
|
|
|
|
response = client.post(
|
|
"/api/auth/refresh",
|
|
json={"refresh_token": refresh_token_data["token"]},
|
|
)
|
|
|
|
assert response.status_code == status.HTTP_401_UNAUTHORIZED
|
|
assert "revoked" in response.json()["detail"]
|
|
|
|
def test_returns_401_for_deleted_user(
|
|
self,
|
|
app,
|
|
client: TestClient,
|
|
refresh_token_data,
|
|
mock_get_redis,
|
|
mock_user_service_instance,
|
|
):
|
|
"""Test that refresh endpoint returns 401 if user no longer exists."""
|
|
# Store the token
|
|
import asyncio
|
|
|
|
async def setup_token():
|
|
async with mock_get_redis() as redis:
|
|
key = f"refresh_token:{refresh_token_data['user_id']}:{refresh_token_data['jti']}"
|
|
await redis.setex(key, 86400, "1")
|
|
|
|
asyncio.get_event_loop().run_until_complete(setup_token())
|
|
|
|
# Configure mock to return None (user deleted)
|
|
mock_user_service_instance.get_by_id.return_value = None
|
|
|
|
# Override the dependency on the test app
|
|
app.dependency_overrides[get_user_service] = lambda: mock_user_service_instance
|
|
|
|
try:
|
|
response = client.post(
|
|
"/api/auth/refresh",
|
|
json={"refresh_token": refresh_token_data["token"]},
|
|
)
|
|
|
|
assert response.status_code == status.HTTP_401_UNAUTHORIZED
|
|
assert "User not found" in response.json()["detail"]
|
|
finally:
|
|
# Clean up override
|
|
app.dependency_overrides.pop(get_user_service, None)
|
|
|
|
|
|
class TestLogout:
|
|
"""Tests for POST /api/auth/logout endpoint."""
|
|
|
|
def test_revokes_token(self, client: TestClient, refresh_token_data, mock_get_redis):
|
|
"""Test that logout revokes the refresh token.
|
|
|
|
After logout, the token should no longer be in Redis.
|
|
"""
|
|
# Store the token first
|
|
import asyncio
|
|
|
|
async def setup_and_check():
|
|
async with mock_get_redis() as redis:
|
|
key = f"refresh_token:{refresh_token_data['user_id']}:{refresh_token_data['jti']}"
|
|
await redis.setex(key, 86400, "1")
|
|
return key
|
|
|
|
key = asyncio.get_event_loop().run_until_complete(setup_and_check())
|
|
|
|
# Logout
|
|
response = client.post(
|
|
"/api/auth/logout",
|
|
json={"refresh_token": refresh_token_data["token"]},
|
|
)
|
|
|
|
assert response.status_code == status.HTTP_204_NO_CONTENT
|
|
|
|
# Verify token is gone
|
|
async def verify_deleted():
|
|
async with mock_get_redis() as redis:
|
|
return await redis.exists(key)
|
|
|
|
exists = asyncio.get_event_loop().run_until_complete(verify_deleted())
|
|
assert exists == 0
|
|
|
|
def test_succeeds_for_invalid_token(self, client: TestClient):
|
|
"""Test that logout succeeds even for invalid tokens.
|
|
|
|
Invalid tokens are effectively "already logged out", so no error.
|
|
"""
|
|
response = client.post(
|
|
"/api/auth/logout",
|
|
json={"refresh_token": "invalid.token.here"},
|
|
)
|
|
|
|
assert response.status_code == status.HTTP_204_NO_CONTENT
|
|
|
|
|
|
class TestLogoutAll:
|
|
"""Tests for POST /api/auth/logout-all endpoint."""
|
|
|
|
def test_requires_authentication(self, client: TestClient):
|
|
"""Test that logout-all requires a valid access token.
|
|
|
|
Without authentication, endpoint should return 401.
|
|
"""
|
|
response = client.post("/api/auth/logout-all")
|
|
assert response.status_code == status.HTTP_401_UNAUTHORIZED
|
|
|
|
def test_revokes_all_tokens(
|
|
self, app, client: TestClient, test_user, access_token, mock_get_redis, mock_db_session
|
|
):
|
|
"""Test that logout-all revokes all refresh tokens for user.
|
|
|
|
Should delete all tokens matching the user's ID pattern.
|
|
"""
|
|
|
|
user_id = UUID(test_user.id) if isinstance(test_user.id, str) else test_user.id
|
|
|
|
# Store multiple tokens
|
|
import asyncio
|
|
|
|
async def setup_tokens():
|
|
async with mock_get_redis() as redis:
|
|
await redis.setex(f"refresh_token:{user_id}:jti-1", 86400, "1")
|
|
await redis.setex(f"refresh_token:{user_id}:jti-2", 86400, "1")
|
|
await redis.setex(f"refresh_token:{user_id}:jti-3", 86400, "1")
|
|
|
|
asyncio.get_event_loop().run_until_complete(setup_tokens())
|
|
|
|
# Set up mock db session to return test user when queried
|
|
# The get_current_user dependency now does a direct DB query
|
|
mock_result = MagicMock()
|
|
mock_result.scalar_one_or_none.return_value = test_user
|
|
mock_db_session.execute = AsyncMock(return_value=mock_result)
|
|
|
|
response = client.post(
|
|
"/api/auth/logout-all",
|
|
headers={"Authorization": f"Bearer {access_token}"},
|
|
)
|
|
|
|
assert response.status_code == status.HTTP_204_NO_CONTENT
|
|
|
|
# Verify all tokens are gone
|
|
async def count_remaining():
|
|
async with mock_get_redis() as redis:
|
|
count = 0
|
|
async for _ in redis.scan_iter(match=f"refresh_token:{user_id}:*"):
|
|
count += 1
|
|
return count
|
|
|
|
remaining = asyncio.get_event_loop().run_until_complete(count_remaining())
|
|
assert remaining == 0
|