Files
sdb2-backend/tests/api/v1/test_auth_endpoints.py
JSC d3d7edb287
Some checks failed
Backend CI / lint (push) Has been cancelled
Backend CI / test (push) Has been cancelled
feat: Add tests for dashboard service including statistics and date filters
2025-08-12 23:34:02 +02:00

705 lines
25 KiB
Python

"""Tests for authentication endpoints."""
from typing import Any
from unittest.mock import patch
import pytest
import pytest_asyncio
from httpx import AsyncClient
from app.models.plan import Plan
from app.models.user import User
from app.services.auth import OAuthUserInfo
from app.utils.auth import JWTUtils
@pytest_asyncio.fixture
async def auth_cookies(test_user: User) -> dict[str, str]:
"""Create authentication cookies with JWT token."""
token_data = {
"sub": str(test_user.id),
"email": test_user.email,
"role": test_user.role,
}
access_token = JWTUtils.create_access_token(token_data)
return {"access_token": access_token}
@pytest_asyncio.fixture
async def admin_cookies(admin_user: User) -> dict[str, str]:
"""Create admin authentication cookies with JWT token."""
token_data = {
"sub": str(admin_user.id),
"email": admin_user.email,
"role": admin_user.role,
}
access_token = JWTUtils.create_access_token(token_data)
return {"access_token": access_token}
class TestAuthEndpoints:
"""Test authentication API endpoints."""
@pytest.mark.asyncio
async def test_register_success(
self,
test_client: AsyncClient,
test_user_data: dict[str, Any],
ensure_plans: tuple[Plan, Plan],
) -> None:
"""Test successful user registration."""
response = await test_client.post("/api/v1/auth/register", json=test_user_data)
assert response.status_code == 201
data = response.json()
# Check user data in response (no token in response body with cookies)
assert data["email"] == test_user_data["email"]
assert data["name"] == test_user_data["name"]
assert data["role"] == "admin" # First user gets admin role
assert data["is_active"] is True
assert data["credits"] > 0
assert "plan" in data
# Check cookies are set - HTTPX AsyncClient preserves Set-Cookie headers
set_cookie_headers = response.headers.get_list("set-cookie")
cookie_names = [header.split("=")[0] for header in set_cookie_headers]
assert "access_token" in cookie_names
assert "refresh_token" in cookie_names
@pytest.mark.asyncio
async def test_register_duplicate_email(
self,
test_client: AsyncClient,
test_user: User,
) -> None:
"""Test registration with duplicate email."""
user_data = {
"email": test_user.email,
"password": "password123",
"name": "Another User",
}
response = await test_client.post("/api/v1/auth/register", json=user_data)
assert response.status_code == 400
data = response.json()
assert "Email address is already registered" in data["detail"]
@pytest.mark.asyncio
async def test_register_invalid_email(self, test_client: AsyncClient) -> None:
"""Test registration with invalid email."""
user_data = {
"email": "invalid-email",
"password": "password123",
"name": "Test User",
}
response = await test_client.post("/api/v1/auth/register", json=user_data)
assert response.status_code == 422 # Validation error
@pytest.mark.asyncio
async def test_register_short_password(self, test_client: AsyncClient) -> None:
"""Test registration with short password."""
user_data = {
"email": "test@example.com",
"password": "short",
"name": "Test User",
}
response = await test_client.post("/api/v1/auth/register", json=user_data)
assert response.status_code == 422 # Validation error
@pytest.mark.asyncio
async def test_register_missing_fields(self, test_client: AsyncClient) -> None:
"""Test registration with missing fields."""
user_data = {
"email": "test@example.com",
# Missing password and name
}
response = await test_client.post("/api/v1/auth/register", json=user_data)
assert response.status_code == 422 # Validation error
@pytest.mark.asyncio
async def test_login_success(
self,
test_client: AsyncClient,
test_user: User,
test_login_data: dict[str, str],
) -> None:
"""Test successful user login."""
response = await test_client.post("/api/v1/auth/login", json=test_login_data)
assert response.status_code == 200
data = response.json()
# Check user data in response (no token in response body with cookies)
assert data["email"] == test_login_data["email"]
assert "name" in data
assert "role" in data
assert data["is_active"] is True
# Check cookies are set - HTTPX AsyncClient preserves Set-Cookie headers
set_cookie_headers = response.headers.get_list("set-cookie")
cookie_names = [header.split("=")[0] for header in set_cookie_headers]
assert "access_token" in cookie_names
assert "refresh_token" in cookie_names
@pytest.mark.asyncio
async def test_login_invalid_email(self, test_client: AsyncClient) -> None:
"""Test login with invalid email."""
login_data = {"email": "nonexistent@example.com", "password": "password123"}
response = await test_client.post("/api/v1/auth/login", json=login_data)
assert response.status_code == 401
data = response.json()
assert "Invalid email or password" in data["detail"]
@pytest.mark.asyncio
async def test_login_invalid_password(
self,
test_client: AsyncClient,
test_user: User,
) -> None:
"""Test login with invalid password."""
login_data = {"email": test_user.email, "password": "wrongpassword"}
response = await test_client.post("/api/v1/auth/login", json=login_data)
assert response.status_code == 401
data = response.json()
assert "Invalid email or password" in data["detail"]
@pytest.mark.asyncio
async def test_login_malformed_request(self, test_client: AsyncClient) -> None:
"""Test login with malformed request."""
login_data = {"email": "invalid-email", "password": "password123"}
response = await test_client.post("/api/v1/auth/login", json=login_data)
assert response.status_code == 422 # Validation error
@pytest.mark.asyncio
async def test_get_current_user_success(
self,
test_client: AsyncClient,
test_user: User,
auth_cookies: dict[str, str],
) -> None:
"""Test getting current user info successfully."""
# Set cookies on client instance to avoid deprecation warning
test_client.cookies.update(auth_cookies)
response = await test_client.get("/api/v1/auth/me")
assert response.status_code == 200
data = response.json()
# Check user data structure
assert "id" in data
assert "email" in data
assert "name" in data
assert "role" in data
assert data["is_active"] is True
assert "plan" in data
@pytest.mark.asyncio
async def test_get_current_user_no_token(self, test_client: AsyncClient) -> None:
"""Test getting current user without authentication token."""
response = await test_client.get("/api/v1/auth/me")
assert response.status_code == 401 # Unauthorized (no cookie provided)
@pytest.mark.asyncio
async def test_get_current_user_invalid_token(
self,
test_client: AsyncClient,
) -> None:
"""Test getting current user with invalid token."""
# Set invalid cookies on client instance
test_client.cookies.update({"access_token": "invalid_token"})
response = await test_client.get("/api/v1/auth/me")
assert response.status_code == 401
data = response.json()
assert "Could not validate credentials" in data["detail"]
@pytest.mark.asyncio
async def test_get_current_user_expired_token(
self,
test_client: AsyncClient,
test_user: User,
) -> None:
"""Test getting current user with expired token."""
from datetime import timedelta
from app.utils.auth import JWTUtils
# Create an expired token (expires immediately)
token_data = {
"sub": "1", # Use a dummy user ID
"email": "test@example.com",
"role": "user",
}
expired_token = JWTUtils.create_access_token(
token_data,
expires_delta=timedelta(seconds=-1),
)
# Set expired cookies on client instance
test_client.cookies.update({"access_token": expired_token})
response = await test_client.get("/api/v1/auth/me")
assert response.status_code == 401
data = response.json()
# The actual error message comes from the JWT library for expired tokens
assert "Token has expired" in data["detail"]
@pytest.mark.asyncio
async def test_logout_success(self, test_client: AsyncClient) -> None:
"""Test logout endpoint."""
# Logout should work even without cookies (just clears them)
test_client.cookies.update({"access_token": "", "refresh_token": ""})
response = await test_client.post("/api/v1/auth/logout")
assert response.status_code == 200
data = response.json()
assert "Successfully logged out" in data["message"]
@pytest.mark.asyncio
async def test_admin_access_with_user_role(
self,
test_client: AsyncClient,
auth_cookies: dict[str, str],
) -> None:
"""Test that regular users cannot access admin endpoints."""
# This test would be for admin-only endpoints when they're created
# For now, we'll test the dependency behavior
import pytest
from fastapi import HTTPException
from app.core.dependencies import get_admin_user
from app.models.user import User
# Create a mock user with regular role
regular_user = User(
id=1,
email="user@example.com",
name="Regular User",
role="user",
is_active=True,
plan_id=1,
credits=100,
)
# Test that get_admin_user raises exception for regular user
with pytest.raises(HTTPException) as exc_info:
await get_admin_user(regular_user)
assert exc_info.value.status_code == 403
assert "Not enough permissions" in exc_info.value.detail
@pytest.mark.asyncio
async def test_admin_access_with_admin_role(
self,
test_client: AsyncClient,
admin_cookies: dict[str, str],
) -> None:
"""Test that admin users can access admin endpoints."""
from app.core.dependencies import get_admin_user
from app.models.user import User
# Create a mock admin user
admin_user = User(
id=1,
email="admin@example.com",
name="Admin User",
role="admin",
is_active=True,
plan_id=1,
credits=1000,
)
# Test that get_admin_user passes for admin user
result = await get_admin_user(admin_user)
assert result == admin_user
@pytest.mark.asyncio
async def test_get_oauth_providers(self, test_client: AsyncClient) -> None:
"""Test getting list of OAuth providers."""
response = await test_client.get("/api/v1/auth/providers")
assert response.status_code == 200
data = response.json()
assert "providers" in data
assert "google" in data["providers"]
assert "github" in data["providers"]
@pytest.mark.asyncio
async def test_oauth_authorize_google(self, test_client: AsyncClient) -> None:
"""Test OAuth authorization URL generation for Google."""
with patch("app.services.oauth.OAuthService.generate_state") as mock_state:
mock_state.return_value = "test_state_123"
response = await test_client.get("/api/v1/auth/google/authorize")
assert response.status_code == 200
data = response.json()
assert "authorization_url" in data
assert "state" in data
assert data["state"] == "test_state_123"
assert "accounts.google.com" in data["authorization_url"]
@pytest.mark.asyncio
async def test_oauth_authorize_github(self, test_client: AsyncClient) -> None:
"""Test OAuth authorization URL generation for GitHub."""
with patch("app.services.oauth.OAuthService.generate_state") as mock_state:
mock_state.return_value = "test_state_456"
response = await test_client.get("/api/v1/auth/github/authorize")
assert response.status_code == 200
data = response.json()
assert "authorization_url" in data
assert "state" in data
assert data["state"] == "test_state_456"
assert "github.com" in data["authorization_url"]
@pytest.mark.asyncio
async def test_oauth_authorize_invalid_provider(
self,
test_client: AsyncClient,
) -> None:
"""Test OAuth authorization with invalid provider."""
response = await test_client.get("/api/v1/auth/invalid/authorize")
assert response.status_code == 400
data = response.json()
assert "Unsupported OAuth provider" in data["detail"]
@pytest.mark.asyncio
async def test_oauth_callback_new_user(
self,
test_client: AsyncClient,
ensure_plans: tuple[Any, Any],
) -> None:
"""Test OAuth callback for new user creation."""
# Mock OAuth user info
mock_user_info = OAuthUserInfo(
provider="google",
provider_user_id="google_123",
email="newuser@gmail.com",
name="New User",
picture="https://example.com/avatar.jpg",
)
# Mock the entire handle_callback method to avoid actual OAuth API calls
with patch("app.services.oauth.OAuthService.handle_callback") as mock_callback:
mock_callback.return_value = mock_user_info
response = await test_client.get(
"/api/v1/auth/google/callback",
params={"code": "auth_code_123", "state": "test_state"},
follow_redirects=False,
)
# OAuth callback should successfully process and redirect to frontend with temp code
assert response.status_code == 302
location = response.headers["location"]
assert location.startswith("http://localhost:8001/auth/callback?code=")
# The fact that we get a 302 redirect means the OAuth login was successful
# Detailed cookie testing can be done in integration tests
@pytest.mark.asyncio
async def test_oauth_callback_existing_user_link(
self,
test_client: AsyncClient,
test_user: Any,
ensure_plans: tuple[Any, Any],
) -> None:
"""Test OAuth callback for linking to existing user."""
# Mock OAuth user info with same email as test user
mock_user_info = OAuthUserInfo(
provider="github",
provider_user_id="github_456",
email=test_user.email, # Same email as existing user
name="Test User",
picture="https://github.com/avatar.jpg",
)
# Mock the entire handle_callback method to avoid actual OAuth API calls
with patch("app.services.oauth.OAuthService.handle_callback") as mock_callback:
mock_callback.return_value = mock_user_info
response = await test_client.get(
"/api/v1/auth/github/callback",
params={"code": "auth_code_456", "state": "test_state"},
follow_redirects=False,
)
# OAuth callback should successfully process and redirect to frontend with temp code
assert response.status_code == 302
location = response.headers["location"]
assert location.startswith("http://localhost:8001/auth/callback?code=")
# The fact that we get a 302 redirect means the OAuth login was successful
# Detailed cookie testing can be done in integration tests
@pytest.mark.asyncio
async def test_oauth_callback_missing_code(self, test_client: AsyncClient) -> None:
"""Test OAuth callback with missing authorization code."""
response = await test_client.get(
"/api/v1/auth/google/callback",
params={"state": "test_state"}, # Missing code parameter
)
assert response.status_code == 422 # Validation error
@pytest.mark.asyncio
async def test_oauth_callback_invalid_provider(
self,
test_client: AsyncClient,
) -> None:
"""Test OAuth callback with invalid provider."""
response = await test_client.get(
"/api/v1/auth/invalid/callback",
params={"code": "auth_code_123", "state": "test_state"},
)
assert response.status_code == 400
data = response.json()
assert "Unsupported OAuth provider" in data["detail"]
@pytest.mark.asyncio
async def test_refresh_token_success(
self,
test_client: AsyncClient,
test_user: User,
test_plan: Plan,
) -> None:
"""Test refresh token success."""
with patch("app.services.auth.AuthService.refresh_access_token") as mock_refresh:
mock_refresh.return_value = type("TokenResponse", (), {
"access_token": "new_access_token",
"expires_in": 3600,
})()
response = await test_client.post(
"/api/v1/auth/refresh",
cookies={"refresh_token": "valid_refresh_token"}
)
assert response.status_code == 200
data = response.json()
assert data["message"] == "Token refreshed successfully"
mock_refresh.assert_called_once_with("valid_refresh_token")
@pytest.mark.asyncio
async def test_refresh_token_no_token(self, test_client: AsyncClient) -> None:
"""Test refresh token without providing refresh token."""
response = await test_client.post("/api/v1/auth/refresh")
assert response.status_code == 401
data = response.json()
assert "No refresh token provided" in data["detail"]
@pytest.mark.asyncio
async def test_refresh_token_service_error(self, test_client: AsyncClient) -> None:
"""Test refresh token with service error."""
with patch("app.services.auth.AuthService.refresh_access_token") as mock_refresh:
mock_refresh.side_effect = Exception("Database error")
response = await test_client.post(
"/api/v1/auth/refresh",
cookies={"refresh_token": "valid_refresh_token"}
)
assert response.status_code == 500
data = response.json()
assert "Token refresh failed" in data["detail"]
@pytest.mark.asyncio
async def test_exchange_oauth_token_invalid_code(
self,
test_client: AsyncClient,
) -> None:
"""Test OAuth token exchange with invalid code."""
response = await test_client.post(
"/api/v1/auth/exchange-oauth-token",
json={"code": "invalid_code"}
)
assert response.status_code == 400
data = response.json()
assert "Invalid or expired OAuth code" in data["detail"]
@pytest.mark.asyncio
async def test_update_profile_success(
self,
test_client: AsyncClient,
test_user: User,
auth_cookies: dict[str, str],
) -> None:
"""Test update profile success."""
with (
patch("app.services.auth.AuthService.update_user_profile") as mock_update,
patch("app.services.auth.AuthService.user_to_response") as mock_user_to_response,
):
updated_user = User(
id=test_user.id,
email=test_user.email,
name="Updated Name",
role=test_user.role,
credits=test_user.credits,
plan_id=test_user.plan_id,
is_active=test_user.is_active,
)
mock_update.return_value = updated_user
# Mock the user_to_response to return UserResponse format
from app.schemas.auth import UserResponse
mock_user_to_response.return_value = UserResponse(
id=test_user.id,
email=test_user.email,
name="Updated Name",
picture=None,
role=test_user.role,
credits=test_user.credits,
is_active=test_user.is_active,
plan={
"id": test_user.plan_id,
"name": "Test Plan",
"max_credits": 100,
"features": [],
},
created_at=test_user.created_at,
updated_at=test_user.updated_at,
)
response = await test_client.patch(
"/api/v1/auth/me",
json={"name": "Updated Name"},
cookies=auth_cookies
)
assert response.status_code == 200
data = response.json()
assert data["name"] == "Updated Name"
@pytest.mark.asyncio
async def test_update_profile_unauthenticated(self, test_client: AsyncClient) -> None:
"""Test update profile without authentication."""
response = await test_client.patch(
"/api/v1/auth/me",
json={"name": "Updated Name"}
)
assert response.status_code == 401
@pytest.mark.asyncio
async def test_change_password_success(
self,
test_client: AsyncClient,
test_user: User,
auth_cookies: dict[str, str],
) -> None:
"""Test change password success."""
with patch("app.services.auth.AuthService.change_user_password") as mock_change:
mock_change.return_value = None
response = await test_client.post(
"/api/v1/auth/change-password",
json={
"current_password": "old_password",
"new_password": "new_password"
},
cookies=auth_cookies
)
assert response.status_code == 200
data = response.json()
assert data["message"] == "Password changed successfully"
@pytest.mark.asyncio
async def test_change_password_unauthenticated(self, test_client: AsyncClient) -> None:
"""Test change password without authentication."""
response = await test_client.post(
"/api/v1/auth/change-password",
json={
"current_password": "old_password",
"new_password": "new_password"
}
)
assert response.status_code == 401
@pytest.mark.asyncio
async def test_get_user_providers_success(
self,
test_client: AsyncClient,
test_user: User,
auth_cookies: dict[str, str],
) -> None:
"""Test get user OAuth providers success."""
with patch("app.services.auth.AuthService.get_user_oauth_providers") as mock_providers:
from app.models.user_oauth import UserOauth
from datetime import datetime, timezone
mock_oauth_google = UserOauth(
id=1,
user_id=test_user.id,
provider="google",
provider_user_id="google123",
email="test@example.com",
name="Test User",
created_at=datetime.now(timezone.utc),
updated_at=datetime.now(timezone.utc),
)
mock_oauth_github = UserOauth(
id=2,
user_id=test_user.id,
provider="github",
provider_user_id="github456",
email="test@example.com",
name="Test User",
created_at=datetime.now(timezone.utc),
updated_at=datetime.now(timezone.utc),
)
mock_providers.return_value = [mock_oauth_google, mock_oauth_github]
response = await test_client.get(
"/api/v1/auth/user-providers",
cookies=auth_cookies
)
assert response.status_code == 200
data = response.json()
assert len(data) == 3 # password + 2 OAuth providers
# Check password provider (first)
assert data[0]["provider"] == "password"
assert data[0]["display_name"] == "Password"
# Check OAuth providers
assert data[1]["provider"] == "google"
assert data[1]["display_name"] == "Google"
assert data[2]["provider"] == "github"
assert data[2]["display_name"] == "GitHub"
@pytest.mark.asyncio
async def test_get_user_providers_unauthenticated(self, test_client: AsyncClient) -> None:
"""Test get user OAuth providers without authentication."""
response = await test_client.get("/api/v1/auth/user-providers")
assert response.status_code == 401