190 lines
7.2 KiB
Python
190 lines
7.2 KiB
Python
"""Tests for API token authentication dependencies."""
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
from app.core.dependencies import get_current_user_api_token, get_current_user_flexible
|
|
from app.models.user import User
|
|
from app.services.auth import AuthService
|
|
|
|
|
|
class TestApiTokenDependencies:
|
|
"""Test API token authentication dependencies."""
|
|
|
|
@pytest.fixture
|
|
def mock_auth_service(self):
|
|
"""Create a mock auth service."""
|
|
return AsyncMock(spec=AuthService)
|
|
|
|
@pytest.fixture
|
|
def test_user(self):
|
|
"""Create a test user."""
|
|
return User(
|
|
id=1,
|
|
email="test@example.com",
|
|
name="Test User",
|
|
role="user",
|
|
is_active=True,
|
|
plan_id=1,
|
|
credits=100,
|
|
api_token="test_api_token_123",
|
|
api_token_expires_at=datetime.now(UTC) + timedelta(days=30),
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_current_user_api_token_success(
|
|
self, mock_auth_service, test_user,
|
|
):
|
|
"""Test successful API token authentication."""
|
|
mock_auth_service.get_user_by_api_token.return_value = test_user
|
|
|
|
api_token_header = "test_api_token_123"
|
|
|
|
result = await get_current_user_api_token(mock_auth_service, api_token_header)
|
|
|
|
assert result == test_user
|
|
mock_auth_service.get_user_by_api_token.assert_called_once_with("test_api_token_123")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_current_user_api_token_no_header(self, mock_auth_service):
|
|
"""Test API token authentication without API-TOKEN header."""
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await get_current_user_api_token(mock_auth_service, None)
|
|
|
|
assert exc_info.value.status_code == 401
|
|
assert "API-TOKEN header required" in exc_info.value.detail
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_current_user_api_token_empty_token(self, mock_auth_service):
|
|
"""Test API token authentication with empty token."""
|
|
api_token_header = " "
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await get_current_user_api_token(mock_auth_service, api_token_header)
|
|
|
|
assert exc_info.value.status_code == 401
|
|
assert "API token required" in exc_info.value.detail
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_current_user_api_token_whitespace_token(self, mock_auth_service):
|
|
"""Test API token authentication with whitespace-only token."""
|
|
api_token_header = " "
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await get_current_user_api_token(mock_auth_service, api_token_header)
|
|
|
|
assert exc_info.value.status_code == 401
|
|
assert "API token required" in exc_info.value.detail
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_current_user_api_token_invalid_token(self, mock_auth_service):
|
|
"""Test API token authentication with invalid token."""
|
|
mock_auth_service.get_user_by_api_token.return_value = None
|
|
|
|
api_token_header = "invalid_token"
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await get_current_user_api_token(mock_auth_service, api_token_header)
|
|
|
|
assert exc_info.value.status_code == 401
|
|
assert "Invalid API token" in exc_info.value.detail
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_current_user_api_token_expired_token(
|
|
self, mock_auth_service, test_user,
|
|
):
|
|
"""Test API token authentication with expired token."""
|
|
# Set expired token
|
|
test_user.api_token_expires_at = datetime.now(UTC) - timedelta(days=1)
|
|
mock_auth_service.get_user_by_api_token.return_value = test_user
|
|
|
|
api_token_header = "expired_token"
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await get_current_user_api_token(mock_auth_service, api_token_header)
|
|
|
|
assert exc_info.value.status_code == 401
|
|
assert "API token has expired" in exc_info.value.detail
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_current_user_api_token_inactive_user(
|
|
self, mock_auth_service, test_user,
|
|
):
|
|
"""Test API token authentication with inactive user."""
|
|
test_user.is_active = False
|
|
mock_auth_service.get_user_by_api_token.return_value = test_user
|
|
|
|
api_token_header = "test_token"
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await get_current_user_api_token(mock_auth_service, api_token_header)
|
|
|
|
assert exc_info.value.status_code == 401
|
|
assert "Account is deactivated" in exc_info.value.detail
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_current_user_api_token_service_exception(self, mock_auth_service):
|
|
"""Test API token authentication with service exception."""
|
|
mock_auth_service.get_user_by_api_token.side_effect = Exception("Database error")
|
|
|
|
api_token_header = "test_token"
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await get_current_user_api_token(mock_auth_service, api_token_header)
|
|
|
|
assert exc_info.value.status_code == 401
|
|
assert "Could not validate API token" in exc_info.value.detail
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_current_user_flexible_uses_api_token(
|
|
self, mock_auth_service, test_user,
|
|
):
|
|
"""Test flexible authentication uses API token when available."""
|
|
mock_auth_service.get_user_by_api_token.return_value = test_user
|
|
|
|
api_token_header = "test_api_token_123"
|
|
access_token = "jwt_token"
|
|
|
|
result = await get_current_user_flexible(
|
|
mock_auth_service, access_token, api_token_header,
|
|
)
|
|
|
|
assert result == test_user
|
|
mock_auth_service.get_user_by_api_token.assert_called_once_with("test_api_token_123")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_current_user_flexible_falls_back_to_jwt(self, mock_auth_service):
|
|
"""Test flexible authentication falls back to JWT when no API token."""
|
|
# Mock the get_current_user function (normally imported)
|
|
with pytest.raises(Exception):
|
|
# This will fail because we can't easily mock the get_current_user import
|
|
# In a real test, you'd mock the import or use dependency injection
|
|
await get_current_user_flexible(mock_auth_service, "jwt_token", None)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_api_token_no_expiry_never_expires(self, mock_auth_service, test_user):
|
|
"""Test API token with no expiry date never expires."""
|
|
test_user.api_token_expires_at = None
|
|
mock_auth_service.get_user_by_api_token.return_value = test_user
|
|
|
|
api_token_header = "test_token"
|
|
|
|
result = await get_current_user_api_token(mock_auth_service, api_token_header)
|
|
|
|
assert result == test_user
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_api_token_with_whitespace(self, mock_auth_service, test_user):
|
|
"""Test API token with leading/trailing whitespace is handled correctly."""
|
|
mock_auth_service.get_user_by_api_token.return_value = test_user
|
|
|
|
api_token_header = " test_token "
|
|
|
|
result = await get_current_user_api_token(mock_auth_service, api_token_header)
|
|
|
|
assert result == test_user
|
|
mock_auth_service.get_user_by_api_token.assert_called_once_with("test_token")
|