Refactor tests for improved consistency and readability
- Updated test cases in `test_auth_endpoints.py` to ensure consistent formatting and style. - Enhanced `test_socket_endpoints.py` with consistent parameter formatting and improved readability. - Cleaned up `conftest.py` by ensuring consistent parameter formatting in fixtures. - Added comprehensive tests for API token dependencies in `test_api_token_dependencies.py`. - Refactored `test_auth_service.py` to maintain consistent parameter formatting. - Cleaned up `test_oauth_service.py` by removing unnecessary imports. - Improved `test_socket_service.py` with consistent formatting and readability. - Enhanced `test_cookies.py` by ensuring consistent formatting and readability. - Introduced new tests for token utilities in `test_token_utils.py` to validate token generation and expiration logic.
This commit is contained in:
JSC
343
tests/api/v1/test_api_token_endpoints.py
Normal file
343
tests/api/v1/test_api_token_endpoints.py
Normal file
@@ -0,0 +1,343 @@
|
||||
"""Tests for API token endpoints."""
|
||||
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
|
||||
from app.models.user import User
|
||||
|
||||
|
||||
class TestApiTokenEndpoints:
|
||||
"""Test API token management endpoints."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_generate_api_token_success(
|
||||
self, authenticated_client: AsyncClient, authenticated_user: User,
|
||||
):
|
||||
"""Test successful API token generation."""
|
||||
request_data = {"expires_days": 30}
|
||||
|
||||
response = await authenticated_client.post(
|
||||
"/api/v1/auth/api-token",
|
||||
json=request_data,
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
|
||||
assert "api_token" in data
|
||||
assert "expires_at" in data
|
||||
assert len(data["api_token"]) > 0
|
||||
|
||||
# Verify token format (should be URL-safe base64)
|
||||
import base64
|
||||
try:
|
||||
base64.urlsafe_b64decode(data["api_token"] + "===") # Add padding
|
||||
except Exception:
|
||||
pytest.fail("API token should be valid URL-safe base64")
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_generate_api_token_default_expiry(
|
||||
self, authenticated_client: AsyncClient,
|
||||
):
|
||||
"""Test API token generation with default expiry."""
|
||||
response = await authenticated_client.post("/api/v1/auth/api-token", json={})
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
|
||||
expires_at_str = data["expires_at"]
|
||||
# Handle both ISO format with/without timezone info
|
||||
if expires_at_str.endswith("Z"):
|
||||
expires_at = datetime.fromisoformat(expires_at_str.replace("Z", "+00:00"))
|
||||
elif "+" in expires_at_str or expires_at_str.count("-") > 2:
|
||||
expires_at = datetime.fromisoformat(expires_at_str)
|
||||
else:
|
||||
# Naive datetime, assume UTC
|
||||
expires_at = datetime.fromisoformat(expires_at_str).replace(tzinfo=UTC)
|
||||
|
||||
expected_expiry = datetime.now(UTC) + timedelta(days=365)
|
||||
|
||||
# Allow 1 minute tolerance
|
||||
assert abs((expires_at - expected_expiry).total_seconds()) < 60
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_generate_api_token_custom_expiry(
|
||||
self, authenticated_client: AsyncClient,
|
||||
):
|
||||
"""Test API token generation with custom expiry."""
|
||||
expires_days = 90
|
||||
request_data = {"expires_days": expires_days}
|
||||
|
||||
response = await authenticated_client.post(
|
||||
"/api/v1/auth/api-token",
|
||||
json=request_data,
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
|
||||
expires_at_str = data["expires_at"]
|
||||
# Handle both ISO format with/without timezone info
|
||||
if expires_at_str.endswith("Z"):
|
||||
expires_at = datetime.fromisoformat(expires_at_str.replace("Z", "+00:00"))
|
||||
elif "+" in expires_at_str or expires_at_str.count("-") > 2:
|
||||
expires_at = datetime.fromisoformat(expires_at_str)
|
||||
else:
|
||||
# Naive datetime, assume UTC
|
||||
expires_at = datetime.fromisoformat(expires_at_str).replace(tzinfo=UTC)
|
||||
|
||||
expected_expiry = datetime.now(UTC) + timedelta(days=expires_days)
|
||||
|
||||
# Allow 1 minute tolerance
|
||||
assert abs((expires_at - expected_expiry).total_seconds()) < 60
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_generate_api_token_validation_errors(
|
||||
self, authenticated_client: AsyncClient,
|
||||
):
|
||||
"""Test API token generation with validation errors."""
|
||||
# Test minimum validation
|
||||
response = await authenticated_client.post(
|
||||
"/api/v1/auth/api-token",
|
||||
json={"expires_days": 0},
|
||||
)
|
||||
assert response.status_code == 422
|
||||
|
||||
# Test maximum validation
|
||||
response = await authenticated_client.post(
|
||||
"/api/v1/auth/api-token",
|
||||
json={"expires_days": 4000},
|
||||
)
|
||||
assert response.status_code == 422
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_generate_api_token_unauthenticated(self, client: AsyncClient):
|
||||
"""Test API token generation without authentication."""
|
||||
response = await client.post(
|
||||
"/api/v1/auth/api-token",
|
||||
json={"expires_days": 30},
|
||||
)
|
||||
assert response.status_code == 401
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_api_token_status_no_token(
|
||||
self, authenticated_client: AsyncClient,
|
||||
):
|
||||
"""Test getting API token status when user has no token."""
|
||||
response = await authenticated_client.get("/api/v1/auth/api-token/status")
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
|
||||
assert data["has_token"] is False
|
||||
assert data["expires_at"] is None
|
||||
assert data["is_expired"] is False
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_api_token_status_with_token(
|
||||
self, authenticated_client: AsyncClient,
|
||||
):
|
||||
"""Test getting API token status when user has a token."""
|
||||
# First generate a token
|
||||
await authenticated_client.post(
|
||||
"/api/v1/auth/api-token",
|
||||
json={"expires_days": 30},
|
||||
)
|
||||
|
||||
# Then check status
|
||||
response = await authenticated_client.get("/api/v1/auth/api-token/status")
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
|
||||
assert data["has_token"] is True
|
||||
assert data["expires_at"] is not None
|
||||
assert data["is_expired"] is False
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_api_token_status_expired_token(
|
||||
self, authenticated_client: AsyncClient, authenticated_user: User,
|
||||
):
|
||||
"""Test getting API token status with expired token."""
|
||||
# Mock expired token
|
||||
with patch("app.utils.auth.TokenUtils.is_token_expired", return_value=True):
|
||||
# Set a token on the user
|
||||
authenticated_user.api_token = "expired_token"
|
||||
authenticated_user.api_token_expires_at = datetime.now(UTC) - timedelta(days=1)
|
||||
|
||||
response = await authenticated_client.get("/api/v1/auth/api-token/status")
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
|
||||
assert data["has_token"] is True
|
||||
assert data["expires_at"] is not None
|
||||
assert data["is_expired"] is True
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_api_token_status_unauthenticated(self, client: AsyncClient):
|
||||
"""Test getting API token status without authentication."""
|
||||
response = await client.get("/api/v1/auth/api-token/status")
|
||||
assert response.status_code == 401
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_revoke_api_token_success(
|
||||
self, authenticated_client: AsyncClient,
|
||||
):
|
||||
"""Test successful API token revocation."""
|
||||
# First generate a token
|
||||
await authenticated_client.post(
|
||||
"/api/v1/auth/api-token",
|
||||
json={"expires_days": 30},
|
||||
)
|
||||
|
||||
# Verify token exists
|
||||
status_response = await authenticated_client.get("/api/v1/auth/api-token/status")
|
||||
assert status_response.json()["has_token"] is True
|
||||
|
||||
# Revoke the token
|
||||
response = await authenticated_client.delete("/api/v1/auth/api-token")
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["message"] == "API token revoked successfully"
|
||||
|
||||
# Verify token is gone
|
||||
status_response = await authenticated_client.get("/api/v1/auth/api-token/status")
|
||||
assert status_response.json()["has_token"] is False
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_revoke_api_token_no_token(
|
||||
self, authenticated_client: AsyncClient,
|
||||
):
|
||||
"""Test revoking API token when user has no token."""
|
||||
response = await authenticated_client.delete("/api/v1/auth/api-token")
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["message"] == "API token revoked successfully"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_revoke_api_token_unauthenticated(self, client: AsyncClient):
|
||||
"""Test revoking API token without authentication."""
|
||||
response = await client.delete("/api/v1/auth/api-token")
|
||||
assert response.status_code == 401
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_api_token_authentication_success(
|
||||
self, client: AsyncClient, authenticated_client: AsyncClient,
|
||||
):
|
||||
"""Test successful authentication using API token."""
|
||||
# Generate API token
|
||||
token_response = await authenticated_client.post(
|
||||
"/api/v1/auth/api-token",
|
||||
json={"expires_days": 30},
|
||||
)
|
||||
api_token = token_response.json()["api_token"]
|
||||
|
||||
# Use API token to authenticate
|
||||
headers = {"Authorization": f"Bearer {api_token}"}
|
||||
response = await client.get("/api/v1/auth/me", headers=headers)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert "id" in data
|
||||
assert "email" in data
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_api_token_authentication_invalid_token(self, client: AsyncClient):
|
||||
"""Test authentication with invalid API token."""
|
||||
headers = {"Authorization": "Bearer invalid_token"}
|
||||
response = await client.get("/api/v1/auth/me", headers=headers)
|
||||
|
||||
assert response.status_code == 401
|
||||
data = response.json()
|
||||
assert "Invalid API token" in data["detail"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_api_token_authentication_expired_token(
|
||||
self, client: AsyncClient, authenticated_client: AsyncClient,
|
||||
):
|
||||
"""Test authentication with expired API token."""
|
||||
# Generate API token
|
||||
token_response = await authenticated_client.post(
|
||||
"/api/v1/auth/api-token",
|
||||
json={"expires_days": 30},
|
||||
)
|
||||
api_token = token_response.json()["api_token"]
|
||||
|
||||
# Mock expired token
|
||||
with patch("app.utils.auth.TokenUtils.is_token_expired", return_value=True):
|
||||
headers = {"Authorization": f"Bearer {api_token}"}
|
||||
response = await client.get("/api/v1/auth/me", headers=headers)
|
||||
|
||||
assert response.status_code == 401
|
||||
data = response.json()
|
||||
assert "API token has expired" in data["detail"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_api_token_authentication_malformed_header(self, client: AsyncClient):
|
||||
"""Test authentication with malformed Authorization header."""
|
||||
# Missing Bearer prefix
|
||||
headers = {"Authorization": "invalid_format"}
|
||||
response = await client.get("/api/v1/auth/me", headers=headers)
|
||||
|
||||
assert response.status_code == 401
|
||||
data = response.json()
|
||||
assert "Invalid authorization header format" in data["detail"]
|
||||
|
||||
# Empty token
|
||||
headers = {"Authorization": "Bearer "}
|
||||
response = await client.get("/api/v1/auth/me", headers=headers)
|
||||
|
||||
assert response.status_code == 401
|
||||
data = response.json()
|
||||
assert "API token required" in data["detail"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_api_token_authentication_inactive_user(
|
||||
self, client: AsyncClient, authenticated_client: AsyncClient, authenticated_user: User,
|
||||
):
|
||||
"""Test authentication with API token for inactive user."""
|
||||
# Generate API token
|
||||
token_response = await authenticated_client.post(
|
||||
"/api/v1/auth/api-token",
|
||||
json={"expires_days": 30},
|
||||
)
|
||||
api_token = token_response.json()["api_token"]
|
||||
|
||||
# Deactivate user
|
||||
authenticated_user.is_active = False
|
||||
|
||||
# Try to authenticate with API token
|
||||
headers = {"Authorization": f"Bearer {api_token}"}
|
||||
response = await client.get("/api/v1/auth/me", headers=headers)
|
||||
|
||||
assert response.status_code == 401
|
||||
data = response.json()
|
||||
assert "Account is deactivated" in data["detail"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_flexible_authentication_prefers_api_token(
|
||||
self, client: AsyncClient, authenticated_client: AsyncClient, auth_cookies: dict[str, str],
|
||||
):
|
||||
"""Test that flexible authentication prefers API token over cookie."""
|
||||
# Generate API token
|
||||
token_response = await authenticated_client.post(
|
||||
"/api/v1/auth/api-token",
|
||||
json={"expires_days": 30},
|
||||
)
|
||||
api_token = token_response.json()["api_token"]
|
||||
|
||||
# Set both cookies and Authorization header
|
||||
client.cookies.update(auth_cookies)
|
||||
headers = {"Authorization": f"Bearer {api_token}"}
|
||||
|
||||
# This should use API token authentication
|
||||
response = await client.get("/api/v1/auth/me", headers=headers)
|
||||
|
||||
assert response.status_code == 200
|
||||
# If it used API token auth, it should work even if cookies are invalid
|
||||
Reference in New Issue
Block a user