cardosofelipe/fast-next-template
Template
1
0
Fork 1
Code Issues 1 Pull Requests Packages Projects Releases Wiki Activity

Add extensive CRUD tests for session and user management; enhance cleanup logic

Browse Source 
- Introduced new unit tests for session CRUD operations, including `update_refresh_token`, `cleanup_expired`, and multi-user session handling.
- Added comprehensive tests for `CRUDBase` methods, covering edge cases, error handling, and UUID validation.
- Reduced default test session creation from 5 to 2 for performance optimization.
- Enhanced pagination, filtering, and sorting validations in `get_multi_with_total`.
- Improved error handling with descriptive assertions for database exceptions.
- Introduced tests for eager-loaded relationships in user sessions for comprehensive coverage.
This commit is contained in:
Felipe Cardoso
2025-11-01 12:18:29 +01:00
parent 293fbcb27e
commit 976fd1d4ad
6 changed files with 1502 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

324
backend/tests/api/test_auth.py Normal file
View File

@@ -0,0 +1,324 @@
# tests/api/test_auth.py
"""
Tests for authentication endpoints.
"""
import pytest
import pytest_asyncio
from fastapi import status
class TestRegisterEndpoint:
"""Tests for POST /auth/register endpoint."""
@pytest.mark.asyncio
async def test_register_success(self, client):
"""Test successful user registration."""
response = await client.post(
"/api/v1/auth/register",
json={
"email": "newuser@example.com",
"password": "NewPassword123!",
"first_name": "New",
"last_name": "User"
}
)
assert response.status_code == status.HTTP_201_CREATED
data = response.json()
assert data["email"] == "newuser@example.com"
@pytest.mark.asyncio
async def test_register_duplicate_email(self, client, async_test_user):
"""Test registration with duplicate email."""
response = await client.post(
"/api/v1/auth/register",
json={
"email": async_test_user.email,
"password": "TestPassword123!",
"first_name": "Test",
"last_name": "User"
}
)
assert response.status_code == status.HTTP_400_BAD_REQUEST
@pytest.mark.asyncio
async def test_register_weak_password(self, client):
"""Test registration with weak password."""
response = await client.post(
"/api/v1/auth/register",
json={
"email": "test@example.com",
"password": "weak",
"first_name": "Test",
"last_name": "User"
}
)
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
class TestLoginEndpoint:
"""Tests for POST /auth/login endpoint."""
@pytest.mark.asyncio
async def test_login_success(self, client, async_test_user):
"""Test successful login."""
response = await client.post(
"/api/v1/auth/login",
json={
"email": "testuser@example.com",
"password": "TestPassword123!"
}
)
assert response.status_code == status.HTTP_200_OK
data = response.json()
assert "access_token" in data
assert "refresh_token" in data
@pytest.mark.asyncio
async def test_login_invalid_credentials(self, client, async_test_user):
"""Test login with invalid password."""
response = await client.post(
"/api/v1/auth/login",
json={
"email": "testuser@example.com",
"password": "WrongPassword123!"
}
)
assert response.status_code == status.HTTP_401_UNAUTHORIZED
@pytest.mark.asyncio
async def test_login_nonexistent_user(self, client):
"""Test login with non-existent user."""
response = await client.post(
"/api/v1/auth/login",
json={
"email": "nonexistent@example.com",
"password": "TestPassword123!"
}
)
assert response.status_code == status.HTTP_401_UNAUTHORIZED
@pytest.mark.asyncio
async def test_login_inactive_user(self, client, async_test_db):
"""Test login with inactive user."""
test_engine, SessionLocal = async_test_db
async with SessionLocal() as session:
from app.models.user import User
from app.core.auth import get_password_hash
inactive_user = User(
email="inactive@example.com",
password_hash=get_password_hash("TestPassword123!"),
first_name="Inactive",
last_name="User",
is_active=False
)
session.add(inactive_user)
await session.commit()
response = await client.post(
"/api/v1/auth/login",
json={
"email": "inactive@example.com",
"password": "TestPassword123!"
}
)
assert response.status_code == status.HTTP_401_UNAUTHORIZED
class TestRefreshTokenEndpoint:
"""Tests for POST /auth/refresh endpoint."""
@pytest_asyncio.fixture
async def refresh_token(self, client, async_test_user):
"""Get a refresh token for testing."""
response = await client.post(
"/api/v1/auth/login",
json={
"email": "testuser@example.com",
"password": "TestPassword123!"
}
)
return response.json()["refresh_token"]
@pytest.mark.asyncio
async def test_refresh_token_success(self, client, refresh_token):
"""Test successful token refresh."""
response = await client.post(
"/api/v1/auth/refresh",
json={"refresh_token": refresh_token}
)
assert response.status_code == status.HTTP_200_OK
data = response.json()
assert "access_token" in data
assert "refresh_token" in data
@pytest.mark.asyncio
async def test_refresh_token_invalid(self, client):
"""Test refresh with invalid token."""
response = await client.post(
"/api/v1/auth/refresh",
json={"refresh_token": "invalid.token.here"}
)
assert response.status_code == status.HTTP_401_UNAUTHORIZED
class TestLogoutEndpoint:
"""Tests for POST /auth/logout endpoint."""
@pytest_asyncio.fixture
async def tokens(self, client, async_test_user):
"""Get tokens for testing."""
response = await client.post(
"/api/v1/auth/login",
json={
"email": "testuser@example.com",
"password": "TestPassword123!"
}
)
data = response.json()
return {"access_token": data["access_token"], "refresh_token": data["refresh_token"]}
@pytest.mark.asyncio
async def test_logout_success(self, client, tokens):
"""Test successful logout."""
response = await client.post(
"/api/v1/auth/logout",
headers={"Authorization": f"Bearer {tokens['access_token']}"},
json={"refresh_token": tokens["refresh_token"]}
)
assert response.status_code == status.HTTP_200_OK
@pytest.mark.asyncio
async def test_logout_without_auth(self, client):
"""Test logout without authentication."""
response = await client.post(
"/api/v1/auth/logout",
json={"refresh_token": "some.token"}
)
assert response.status_code == status.HTTP_401_UNAUTHORIZED
class TestPasswordResetRequest:
"""Tests for POST /auth/password-reset/request endpoint."""
@pytest.mark.asyncio
async def test_password_reset_request_success(self, client, async_test_user):
"""Test password reset request with existing user."""
response = await client.post(
"/api/v1/auth/password-reset/request",
json={"email": async_test_user.email}
)
assert response.status_code == status.HTTP_200_OK
data = response.json()
assert data["success"] is True
@pytest.mark.asyncio
async def test_password_reset_request_nonexistent_email(self, client):
"""Test password reset request with non-existent email."""
response = await client.post(
"/api/v1/auth/password-reset/request",
json={"email": "nonexistent@example.com"}
)
assert response.status_code == status.HTTP_200_OK
data = response.json()
assert data["success"] is True
class TestPasswordResetConfirm:
"""Tests for POST /auth/password-reset/confirm endpoint."""
@pytest.mark.asyncio
async def test_password_reset_confirm_invalid_token(self, client):
"""Test password reset with invalid token."""
response = await client.post(
"/api/v1/auth/password-reset/confirm",
json={
"token": "invalid.token.here",
"new_password": "NewPassword123!"
}
)
assert response.status_code == status.HTTP_400_BAD_REQUEST
class TestLogoutAll:
"""Tests for POST /auth/logout-all endpoint."""
@pytest_asyncio.fixture
async def tokens(self, client, async_test_user):
"""Get tokens for testing."""
response = await client.post(
"/api/v1/auth/login",
json={
"email": "testuser@example.com",
"password": "TestPassword123!"
}
)
data = response.json()
return {"access_token": data["access_token"], "refresh_token": data["refresh_token"]}
@pytest.mark.asyncio
async def test_logout_all_success(self, client, tokens):
"""Test logout from all devices."""
response = await client.post(
"/api/v1/auth/logout-all",
headers={"Authorization": f"Bearer {tokens['access_token']}"}
)
assert response.status_code == status.HTTP_200_OK
data = response.json()
assert data["success"] is True
assert "sessions terminated" in data["message"].lower()
@pytest.mark.asyncio
async def test_logout_all_unauthorized(self, client):
"""Test logout-all without authentication."""
response = await client.post("/api/v1/auth/logout-all")
assert response.status_code == status.HTTP_401_UNAUTHORIZED
class TestOAuthLogin:
"""Tests for POST /auth/login/oauth endpoint."""
@pytest.mark.asyncio
async def test_oauth_login_success(self, client, async_test_user):
"""Test successful OAuth login."""
response = await client.post(
"/api/v1/auth/login/oauth",
data={
"username": "testuser@example.com",
"password": "TestPassword123!"
}
)
assert response.status_code == status.HTTP_200_OK
data = response.json()
assert "access_token" in data
assert "refresh_token" in data
assert data["token_type"] == "bearer"
@pytest.mark.asyncio
async def test_oauth_login_invalid_credentials(self, client, async_test_user):
"""Test OAuth login with invalid credentials."""
response = await client.post(
"/api/v1/auth/login/oauth",
data={
"username": "testuser@example.com",
"password": "WrongPassword"
}
)
assert response.status_code == status.HTTP_401_UNAUTHORIZED

44
backend/tests/api/test_security_headers.py
View File

@@ -6,9 +6,9 @@ from unittest.mock import patch
from app.main import app
@pytest.fixture
@pytest.fixture(scope="module")
def client():
"""Create a FastAPI test client for the main app."""
"""Create a FastAPI test client for the main app (module-scoped for speed)."""
# Mock get_db to avoid database connection issues
with patch("app.core.database.get_db") as mock_get_db:
async def mock_session_generator():
@@ -25,46 +25,38 @@ def client():
class TestSecurityHeaders:
"""Tests for security headers middleware"""
def test_x_frame_options_header(self, client):
"""Test that X-Frame-Options header is set to DENY"""
def test_all_security_headers(self, client):
"""Test all security headers in a single request for speed"""
response = client.get("/health")
# Test X-Frame-Options
assert "X-Frame-Options" in response.headers
assert response.headers["X-Frame-Options"] == "DENY"
def test_x_content_type_options_header(self, client):
"""Test that X-Content-Type-Options header is set to nosniff"""
response = client.get("/health")
# Test X-Content-Type-Options
assert "X-Content-Type-Options" in response.headers
assert response.headers["X-Content-Type-Options"] == "nosniff"
def test_x_xss_protection_header(self, client):
"""Test that X-XSS-Protection header is set"""
response = client.get("/health")
# Test X-XSS-Protection
assert "X-XSS-Protection" in response.headers
assert response.headers["X-XSS-Protection"] == "1; mode=block"
def test_content_security_policy_header(self, client):
"""Test that Content-Security-Policy header is set"""
response = client.get("/health")
# Test Content-Security-Policy
assert "Content-Security-Policy" in response.headers
assert "default-src 'self'" in response.headers["Content-Security-Policy"]
assert "frame-ancestors 'none'" in response.headers["Content-Security-Policy"]
def test_permissions_policy_header(self, client):
"""Test that Permissions-Policy header is set"""
response = client.get("/health")
# Test Permissions-Policy
assert "Permissions-Policy" in response.headers
assert "geolocation=()" in response.headers["Permissions-Policy"]
assert "microphone=()" in response.headers["Permissions-Policy"]
assert "camera=()" in response.headers["Permissions-Policy"]
def test_referrer_policy_header(self, client):
"""Test that Referrer-Policy header is set"""
response = client.get("/health")
# Test Referrer-Policy
assert "Referrer-Policy" in response.headers
assert response.headers["Referrer-Policy"] == "strict-origin-when-cross-origin"
def test_strict_transport_security_not_in_development(self, client):
def test_hsts_not_in_development(self, client):
"""Test that Strict-Transport-Security header is not set in development"""
from app.core.config import settings
@@ -73,18 +65,6 @@ class TestSecurityHeaders:
response = client.get("/health")
assert "Strict-Transport-Security" not in response.headers
def test_security_headers_on_all_endpoints(self, client):
"""Test that security headers are present on all endpoints"""
# Test health endpoint
response = client.get("/health")
assert "X-Frame-Options" in response.headers
assert "X-Content-Type-Options" in response.headers
# Test root endpoint
response = client.get("/")
assert "X-Frame-Options" in response.headers
assert "X-Content-Type-Options" in response.headers
def test_security_headers_on_404(self, client):
"""Test that security headers are present even on 404 responses"""
response = client.get("/nonexistent-endpoint")

96
backend/tests/api/test_sessions.py
View File

@@ -365,3 +365,99 @@ class TestCleanupExpiredSessions:
response = await client.delete("/api/v1/sessions/me/expired")
assert response.status_code == status.HTTP_401_UNAUTHORIZED
# Additional tests for better coverage
class TestSessionsAdditionalCases:
"""Additional tests to improve sessions endpoint coverage."""
@pytest.mark.asyncio
async def test_list_sessions_pagination(self, client, async_test_user, async_test_db, user_token):
"""Test listing sessions with pagination."""
test_engine, SessionLocal = async_test_db
# Create multiple sessions
async with SessionLocal() as session:
from app.crud.session import session as session_crud
from app.schemas.sessions import SessionCreate
for i in range(5):
session_data = SessionCreate(
user_id=async_test_user.id,
refresh_token_jti=str(uuid4()),
device_name=f"Device {i}",
ip_address=f"192.168.1.{i}",
user_agent="Mozilla/5.0",
expires_at=datetime.now(timezone.utc) + timedelta(days=7),
last_used_at=datetime.now(timezone.utc)
)
await session_crud.create_session(session, obj_in=session_data)
await session.commit()
response = await client.get(
"/api/v1/sessions/me?page=1&limit=3",
headers={"Authorization": f"Bearer {user_token}"}
)
assert response.status_code == status.HTTP_200_OK
data = response.json()
assert "sessions" in data
assert "total" in data
@pytest.mark.asyncio
async def test_revoke_session_invalid_uuid(self, client, user_token):
"""Test revoking session with invalid UUID."""
response = await client.delete(
"/api/v1/sessions/not-a-uuid",
headers={"Authorization": f"Bearer {user_token}"}
)
# Should return 422 for invalid UUID format
assert response.status_code in [status.HTTP_422_UNPROCESSABLE_ENTITY, status.HTTP_404_NOT_FOUND]
@pytest.mark.asyncio
async def test_cleanup_expired_sessions_with_mixed_states(self, client, async_test_user, async_test_db, user_token):
"""Test cleanup with mix of active/inactive and expired/not-expired sessions."""
test_engine, SessionLocal = async_test_db
from app.crud.session import session as session_crud
from app.schemas.sessions import SessionCreate
async with SessionLocal() as db:
# Expired + inactive (should be cleaned)
e1_data = SessionCreate(
user_id=async_test_user.id,
refresh_token_jti=str(uuid4()),
device_name="Expired Inactive",
ip_address="192.168.1.100",
user_agent="Mozilla/5.0",
expires_at=datetime.now(timezone.utc) - timedelta(days=1),
last_used_at=datetime.now(timezone.utc) - timedelta(days=2)
)
e1 = await session_crud.create_session(db, obj_in=e1_data)
e1.is_active = False
db.add(e1)
# Expired but still active (should NOT be cleaned - only inactive+expired)
e2_data = SessionCreate(
user_id=async_test_user.id,
refresh_token_jti=str(uuid4()),
device_name="Expired Active",
ip_address="192.168.1.101",
user_agent="Mozilla/5.0",
expires_at=datetime.now(timezone.utc) - timedelta(hours=1),
last_used_at=datetime.now(timezone.utc) - timedelta(hours=2)
)
await session_crud.create_session(db, obj_in=e2_data)
await db.commit()
response = await client.delete(
"/api/v1/sessions/me/expired",
headers={"Authorization": f"Bearer {user_token}"}
)
assert response.status_code == status.HTTP_200_OK
data = response.json()
assert data["success"] is True