456 lines
17 KiB
Python
456 lines
17 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"]
|