Add comprehensive test suite and utilities for user functionality

This commit introduces a suite of tests for user models, schemas, CRUD operations, and authentication services. It also adds utilities for in-memory database setup to support these tests and updates environment settings for consistency.
This commit is contained in:
2025-03-04 19:10:54 +01:00
parent 481b6d618e
commit 162e586e13
40 changed files with 2948 additions and 11 deletions

View File

View File

@@ -0,0 +1,369 @@
# tests/api/routes/test_auth.py
import json
import uuid
from datetime import datetime, timezone
from unittest.mock import patch, MagicMock, Mock
import pytest
from fastapi import FastAPI, Depends
from fastapi.testclient import TestClient
from sqlalchemy.orm import Session
from app.api.routes.auth import router as auth_router
from app.core.auth import get_password_hash
from app.core.database import get_db
from app.models.user import User
from app.services.auth_service import AuthService, AuthenticationError
from app.core.auth import TokenExpiredError, TokenInvalidError
# Mock the get_db dependency
@pytest.fixture
def override_get_db(db_session):
"""Override get_db dependency for testing."""
return db_session
@pytest.fixture
def app(override_get_db):
"""Create a FastAPI test application with overridden dependencies."""
app = FastAPI()
app.include_router(auth_router, prefix="/auth", tags=["auth"])
# Override the get_db dependency
app.dependency_overrides[get_db] = lambda: override_get_db
return app
@pytest.fixture
def client(app):
"""Create a FastAPI test client."""
return TestClient(app)
class TestRegisterUser:
"""Tests for the register_user endpoint."""
def test_register_user_success(self, client, monkeypatch, db_session):
"""Test successful user registration."""
# Mock the service method with a valid complete User object
mock_user = User(
id=uuid.uuid4(),
email="newuser@example.com",
password_hash="hashed_password",
first_name="New",
last_name="User",
is_active=True,
is_superuser=False,
created_at=datetime.now(timezone.utc),
updated_at=datetime.now(timezone.utc)
)
# Use patch for mocking
with patch.object(AuthService, 'create_user', return_value=mock_user):
# Test request
response = client.post(
"/auth/register",
json={
"email": "newuser@example.com",
"password": "Password123",
"first_name": "New",
"last_name": "User"
}
)
# Assertions
assert response.status_code == 201
data = response.json()
assert data["email"] == "newuser@example.com"
assert data["first_name"] == "New"
assert data["last_name"] == "User"
assert "password" not in data
def test_register_user_duplicate_email(self, client, db_session):
"""Test registration with duplicate email."""
# Use patch for mocking with a side effect
with patch.object(AuthService, 'create_user',
side_effect=AuthenticationError("User with this email already exists")):
# Test request
response = client.post(
"/auth/register",
json={
"email": "existing@example.com",
"password": "Password123",
"first_name": "Existing",
"last_name": "User"
}
)
# Assertions
assert response.status_code == 409
assert "already exists" in response.json()["detail"]
class TestLogin:
"""Tests for the login endpoint."""
def test_login_success(self, client, mock_user, db_session):
"""Test successful login."""
# Ensure mock_user has required attributes
if not hasattr(mock_user, 'created_at') or mock_user.created_at is None:
mock_user.created_at = datetime.now(timezone.utc)
if not hasattr(mock_user, 'updated_at') or mock_user.updated_at is None:
mock_user.updated_at = datetime.now(timezone.utc)
# Create mock tokens
mock_tokens = MagicMock(
access_token="mock_access_token",
refresh_token="mock_refresh_token",
token_type="bearer"
)
# Use context managers for patching
with patch.object(AuthService, 'authenticate_user', return_value=mock_user), \
patch.object(AuthService, 'create_tokens', return_value=mock_tokens):
# Test request
response = client.post(
"/auth/login",
json={
"email": "user@example.com",
"password": "Password123"
}
)
# Assertions
assert response.status_code == 200
data = response.json()
assert data["access_token"] == "mock_access_token"
assert data["refresh_token"] == "mock_refresh_token"
assert data["token_type"] == "bearer"
def test_login_invalid_credentials_debug(self, client, app):
"""Improved test for login with invalid credentials."""
# Print response for debugging
from app.services.auth_service import AuthService
# Create a complete mock for AuthService
class MockAuthService:
@staticmethod
def authenticate_user(db, email, password):
print(f"Mock called with: {email}, {password}")
return None
# Replace the entire class with our mock
original_service = AuthService
try:
# Replace with our mock
import sys
sys.modules['app.services.auth_service'].AuthService = MockAuthService
# Make the request
response = client.post(
"/auth/login",
json={
"email": "user@example.com",
"password": "WrongPassword"
}
)
# Print response details for debugging
print(f"Response status: {response.status_code}")
print(f"Response body: {response.text}")
# Assertions
assert response.status_code == 401
assert "Invalid email or password" in response.json()["detail"]
finally:
# Restore original service
sys.modules['app.services.auth_service'].AuthService = original_service
def test_login_inactive_user(self, client, db_session):
"""Test login with inactive user."""
# Mock authentication to raise an error
with patch.object(AuthService, 'authenticate_user',
side_effect=AuthenticationError("User account is inactive")):
# Test request
response = client.post(
"/auth/login",
json={
"email": "inactive@example.com",
"password": "Password123"
}
)
# Assertions
assert response.status_code == 401
assert "inactive" in response.json()["detail"]
class TestRefreshToken:
"""Tests for the refresh_token endpoint."""
def test_refresh_token_success(self, client, db_session):
"""Test successful token refresh."""
# Mock refresh to return tokens
mock_tokens = MagicMock(
access_token="new_access_token",
refresh_token="new_refresh_token",
token_type="bearer"
)
with patch.object(AuthService, 'refresh_tokens', return_value=mock_tokens):
# Test request
response = client.post(
"/auth/refresh",
json={
"refresh_token": "valid_refresh_token"
}
)
# Assertions
assert response.status_code == 200
data = response.json()
assert data["access_token"] == "new_access_token"
assert data["refresh_token"] == "new_refresh_token"
assert data["token_type"] == "bearer"
def test_refresh_token_expired(self, client, db_session):
"""Test refresh with expired token."""
# Mock refresh to raise expired token error
with patch.object(AuthService, 'refresh_tokens',
side_effect=TokenExpiredError("Token expired")):
# Test request
response = client.post(
"/auth/refresh",
json={
"refresh_token": "expired_refresh_token"
}
)
# Assertions
assert response.status_code == 401
assert "expired" in response.json()["detail"]
def test_refresh_token_invalid(self, client, db_session):
"""Test refresh with invalid token."""
# Mock refresh to raise invalid token error
with patch.object(AuthService, 'refresh_tokens',
side_effect=TokenInvalidError("Invalid token")):
# Test request
response = client.post(
"/auth/refresh",
json={
"refresh_token": "invalid_refresh_token"
}
)
# Assertions
assert response.status_code == 401
assert "Invalid" in response.json()["detail"]
class TestChangePassword:
"""Tests for the change_password endpoint."""
def test_change_password_success(self, client, mock_user, db_session, app):
"""Test successful password change."""
# Ensure mock_user has required attributes
if not hasattr(mock_user, 'created_at') or mock_user.created_at is None:
mock_user.created_at = datetime.now(timezone.utc)
if not hasattr(mock_user, 'updated_at') or mock_user.updated_at is None:
mock_user.updated_at = datetime.now(timezone.utc)
# Override get_current_user dependency
from app.api.dependencies.auth import get_current_user
app.dependency_overrides[get_current_user] = lambda: mock_user
# Mock password change to return success
with patch.object(AuthService, 'change_password', return_value=True):
# Test request
response = client.post(
"/auth/change-password",
json={
"current_password": "OldPassword123",
"new_password": "NewPassword123"
}
)
# Assertions
assert response.status_code == 200
assert "success" in response.json()["message"].lower()
# Clean up override
if get_current_user in app.dependency_overrides:
del app.dependency_overrides[get_current_user]
def test_change_password_incorrect_current_password(self, client, mock_user, db_session, app):
"""Test change password with incorrect current password."""
# Ensure mock_user has required attributes
if not hasattr(mock_user, 'created_at') or mock_user.created_at is None:
mock_user.created_at = datetime.now(timezone.utc)
if not hasattr(mock_user, 'updated_at') or mock_user.updated_at is None:
mock_user.updated_at = datetime.now(timezone.utc)
# Override get_current_user dependency
from app.api.dependencies.auth import get_current_user
app.dependency_overrides[get_current_user] = lambda: mock_user
# Mock password change to raise error
with patch.object(AuthService, 'change_password',
side_effect=AuthenticationError("Current password is incorrect")):
# Test request
response = client.post(
"/auth/change-password",
json={
"current_password": "WrongPassword",
"new_password": "NewPassword123"
}
)
# Assertions
assert response.status_code == 400
assert "incorrect" in response.json()["detail"].lower()
# Clean up override
if get_current_user in app.dependency_overrides:
del app.dependency_overrides[get_current_user]
class TestGetCurrentUserInfo:
"""Tests for the get_current_user_info endpoint."""
def test_get_current_user_info(self, client, mock_user, app):
"""Test getting current user info."""
# Ensure mock_user has required attributes
if not hasattr(mock_user, 'created_at') or mock_user.created_at is None:
mock_user.created_at = datetime.now(timezone.utc)
if not hasattr(mock_user, 'updated_at') or mock_user.updated_at is None:
mock_user.updated_at = datetime.now(timezone.utc)
# Override get_current_user dependency
from app.api.dependencies.auth import get_current_user
app.dependency_overrides[get_current_user] = lambda: mock_user
# Test request
response = client.get("/auth/me")
# Assertions
assert response.status_code == 200
data = response.json()
assert data["email"] == mock_user.email
assert data["first_name"] == mock_user.first_name
assert data["last_name"] == mock_user.last_name
assert "password" not in data
# Clean up override
if get_current_user in app.dependency_overrides:
del app.dependency_overrides[get_current_user]
def test_get_current_user_info_unauthorized(self, client):
"""Test getting user info without authentication."""
# Test request without authentication
response = client.get("/auth/me")
# Assertions
assert response.status_code == 401

View File

@@ -0,0 +1,211 @@
# tests/api/dependencies/test_auth_dependencies.py
import pytest
from unittest.mock import patch, MagicMock
from fastapi import HTTPException
from app.api.dependencies.auth import (
get_current_user,
get_current_active_user,
get_current_superuser,
get_optional_current_user
)
from app.core.auth import TokenExpiredError, TokenInvalidError
@pytest.fixture
def mock_token():
return "mock.jwt.token"
class TestGetCurrentUser:
"""Tests for get_current_user dependency"""
def test_get_current_user_success(self, db_session, mock_user, mock_token):
"""Test successfully getting the current user"""
# Mock get_token_data to return user_id that matches our mock_user
with patch('app.api.dependencies.auth.get_token_data') as mock_get_data:
mock_get_data.return_value.user_id = mock_user.id
# Call the dependency
user = get_current_user(db=db_session, token=mock_token)
# Verify the correct user was returned
assert user.id == mock_user.id
assert user.email == mock_user.email
def test_get_current_user_nonexistent(self, db_session, mock_token):
"""Test when the token contains a user ID that doesn't exist"""
# Mock get_token_data to return a non-existent user ID
# Use a real UUID object instead of a string
import uuid
nonexistent_id = uuid.UUID("11111111-1111-1111-1111-111111111111")
with patch('app.api.dependencies.auth.get_token_data') as mock_get_data:
mock_get_data.return_value.user_id = nonexistent_id # Using UUID object, not string
# Should raise HTTPException with 404 status
with pytest.raises(HTTPException) as exc_info:
get_current_user(db=db_session, token=mock_token)
assert exc_info.value.status_code == 404
def test_get_current_user_inactive(self, db_session, mock_user, mock_token):
"""Test when the user is inactive"""
# Make the user inactive
mock_user.is_active = False
db_session.commit()
# Mock get_token_data
with patch('app.api.dependencies.auth.get_token_data') as mock_get_data:
mock_get_data.return_value.user_id = mock_user.id
# Should raise HTTPException with 403 status
with pytest.raises(HTTPException) as exc_info:
get_current_user(db=db_session, token=mock_token)
assert exc_info.value.status_code == 403
def test_get_current_user_expired_token(self, db_session, mock_token):
"""Test with an expired token"""
# Mock get_token_data to raise TokenExpiredError
with patch('app.api.dependencies.auth.get_token_data') as mock_get_data:
mock_get_data.side_effect = TokenExpiredError("Token expired")
# Should raise HTTPException with 401 status
with pytest.raises(HTTPException) as exc_info:
get_current_user(db=db_session, token=mock_token)
assert exc_info.value.status_code == 401
assert "Token expired" in exc_info.value.detail
def test_get_current_user_invalid_token(self, db_session, mock_token):
"""Test with an invalid token"""
# Mock get_token_data to raise TokenInvalidError
with patch('app.api.dependencies.auth.get_token_data') as mock_get_data:
mock_get_data.side_effect = TokenInvalidError("Invalid token")
# Should raise HTTPException with 401 status
with pytest.raises(HTTPException) as exc_info:
get_current_user(db=db_session, token=mock_token)
assert exc_info.value.status_code == 401
assert "Could not validate credentials" in exc_info.value.detail
class TestGetCurrentActiveUser:
"""Tests for get_current_active_user dependency"""
def test_get_current_active_user(self, mock_user):
"""Test getting an active user"""
# Ensure user is active
mock_user.is_active = True
# Call the dependency with mocked current_user
user = get_current_active_user(current_user=mock_user)
# Should return the same user
assert user == mock_user
def test_get_current_inactive_user(self, mock_user):
"""Test getting an inactive user"""
# Make user inactive
mock_user.is_active = False
# Should raise HTTPException with 403 status
with pytest.raises(HTTPException) as exc_info:
get_current_active_user(current_user=mock_user)
assert exc_info.value.status_code == 403
assert "Inactive user" in exc_info.value.detail
class TestGetCurrentSuperuser:
"""Tests for get_current_superuser dependency"""
def test_get_current_superuser(self, mock_user):
"""Test getting a superuser"""
# Make user a superuser
mock_user.is_superuser = True
# Call the dependency with mocked current_user
user = get_current_superuser(current_user=mock_user)
# Should return the same user
assert user == mock_user
def test_get_current_non_superuser(self, mock_user):
"""Test getting a non-superuser"""
# Ensure user is not a superuser
mock_user.is_superuser = False
# Should raise HTTPException with 403 status
with pytest.raises(HTTPException) as exc_info:
get_current_superuser(current_user=mock_user)
assert exc_info.value.status_code == 403
assert "Not enough permissions" in exc_info.value.detail
class TestGetOptionalCurrentUser:
"""Tests for get_optional_current_user dependency"""
def test_get_optional_current_user_with_token(self, db_session, mock_user, mock_token):
"""Test getting optional user with a valid token"""
# Mock get_token_data
with patch('app.api.dependencies.auth.get_token_data') as mock_get_data:
mock_get_data.return_value.user_id = mock_user.id
# Call the dependency
user = get_optional_current_user(db=db_session, token=mock_token)
# Should return the correct user
assert user is not None
assert user.id == mock_user.id
def test_get_optional_current_user_no_token(self, db_session):
"""Test getting optional user with no token"""
# Call the dependency with no token
user = get_optional_current_user(db=db_session, token=None)
# Should return None
assert user is None
def test_get_optional_current_user_invalid_token(self, db_session, mock_token):
"""Test getting optional user with an invalid token"""
# Mock get_token_data to raise TokenInvalidError
with patch('app.api.dependencies.auth.get_token_data') as mock_get_data:
mock_get_data.side_effect = TokenInvalidError("Invalid token")
# Call the dependency
user = get_optional_current_user(db=db_session, token=mock_token)
# Should return None, not raise an exception
assert user is None
def test_get_optional_current_user_expired_token(self, db_session, mock_token):
"""Test getting optional user with an expired token"""
# Mock get_token_data to raise TokenExpiredError
with patch('app.api.dependencies.auth.get_token_data') as mock_get_data:
mock_get_data.side_effect = TokenExpiredError("Token expired")
# Call the dependency
user = get_optional_current_user(db=db_session, token=mock_token)
# Should return None, not raise an exception
assert user is None
def test_get_optional_current_user_inactive(self, db_session, mock_user, mock_token):
"""Test getting optional user when user is inactive"""
# Make the user inactive
mock_user.is_active = False
db_session.commit()
# Mock get_token_data
with patch('app.api.dependencies.auth.get_token_data') as mock_get_data:
mock_get_data.return_value.user_id = mock_user.id
# Call the dependency
user = get_optional_current_user(db=db_session, token=mock_token)
# Should return None for inactive users
assert user is None

66
backend/tests/conftest.py Normal file
View File

@@ -0,0 +1,66 @@
# tests/conftest.py
import uuid
from datetime import datetime, timezone
import pytest
from app.models.user import User
from app.utils.test_utils import setup_test_db, teardown_test_db, setup_async_test_db, teardown_async_test_db
@pytest.fixture(scope="function")
def db_session():
"""
Creates a fresh SQLite in-memory database for each test function.
Yields a SQLAlchemy session that can be used for testing.
"""
# Set up the database
test_engine, TestingSessionLocal = setup_test_db()
# Create a session
with TestingSessionLocal() as session:
yield session
# Clean up
teardown_test_db(test_engine)
@pytest.fixture(scope="function") # Define a fixture
async def async_test_db():
"""Fixture provides new testing engine and session for each test run to improve isolation."""
test_engine, AsyncTestingSessionLocal = await setup_async_test_db()
yield test_engine, AsyncTestingSessionLocal
await teardown_async_test_db(test_engine)
@pytest.fixture
def user_create_data():
return {
"email": "newtest@example.com", # Changed to avoid conflict with mock_user
"password": "TestPassword123!",
"first_name": "Test",
"last_name": "User",
"phone_number": "+1234567890",
"is_superuser": False,
"preferences": None
}
@pytest.fixture
def mock_user(db_session):
"""Fixture to create and return a mock User instance."""
mock_user = User(
id=uuid.uuid4(),
email="mockuser@example.com",
password_hash="mockhashedpassword",
first_name="Mock",
last_name="User",
phone_number="1234567890",
is_active=True,
is_superuser=False,
preferences=None,
)
db_session.add(mock_user)
db_session.commit()
return mock_user

View File

View File

@@ -0,0 +1,260 @@
# tests/core/test_auth.py
import uuid
import pytest
from datetime import datetime, timedelta, timezone
from jose import jwt
from pydantic import ValidationError
from app.core.auth import (
verify_password,
get_password_hash,
create_access_token,
create_refresh_token,
decode_token,
get_token_data,
TokenExpiredError,
TokenInvalidError,
TokenMissingClaimError
)
from app.core.config import settings
class TestPasswordHandling:
"""Tests for password hashing and verification functions"""
def test_password_hash_different_from_password(self):
"""Test that a password hash is different from the original password"""
password = "TestPassword123"
hashed = get_password_hash(password)
assert hashed != password
def test_verify_correct_password(self):
"""Test that verify_password returns True for the correct password"""
password = "TestPassword123"
hashed = get_password_hash(password)
assert verify_password(password, hashed) is True
def test_verify_incorrect_password(self):
"""Test that verify_password returns False for an incorrect password"""
password = "TestPassword123"
wrong_password = "WrongPassword123"
hashed = get_password_hash(password)
assert verify_password(wrong_password, hashed) is False
def test_same_password_different_hash(self):
"""Test that the same password gets a different hash each time"""
password = "TestPassword123"
hash1 = get_password_hash(password)
hash2 = get_password_hash(password)
assert hash1 != hash2
class TestTokenCreation:
"""Tests for token creation functions"""
def test_create_access_token(self):
"""Test that an access token is created with the correct claims"""
user_id = str(uuid.uuid4())
custom_claims = {
"email": "test@example.com",
"first_name": "Test",
"is_superuser": True
}
token = create_access_token(subject=user_id, claims=custom_claims)
# Decode token to verify claims
payload = jwt.decode(
token,
settings.SECRET_KEY,
algorithms=[settings.ALGORITHM]
)
# Check standard claims
assert payload["sub"] == user_id
assert "jti" in payload
assert "exp" in payload
assert "iat" in payload
assert payload["type"] == "access"
# Check custom claims
for key, value in custom_claims.items():
assert payload[key] == value
def test_create_refresh_token(self):
"""Test that a refresh token is created with the correct claims"""
user_id = str(uuid.uuid4())
token = create_refresh_token(subject=user_id)
# Decode token to verify claims
payload = jwt.decode(
token,
settings.SECRET_KEY,
algorithms=[settings.ALGORITHM]
)
# Check standard claims
assert payload["sub"] == user_id
assert "jti" in payload
assert "exp" in payload
assert "iat" in payload
assert payload["type"] == "refresh"
def test_token_expiration(self):
"""Test that tokens have the correct expiration time"""
user_id = str(uuid.uuid4())
expires = timedelta(minutes=5)
# Create token with specific expiration
token = create_access_token(
subject=user_id,
expires_delta=expires
)
# Decode token
payload = jwt.decode(
token,
settings.SECRET_KEY,
algorithms=[settings.ALGORITHM]
)
# Get actual expiration time from token
expiration = datetime.fromtimestamp(payload["exp"], tz=timezone.utc)
# Calculate expected expiration (approximately)
now = datetime.now(timezone.utc)
expected_expiration = now + expires
# Difference should be small (less than 1 second)
difference = abs((expiration - expected_expiration).total_seconds())
assert difference < 1
class TestTokenDecoding:
"""Tests for token decoding and validation functions"""
def test_decode_valid_token(self):
"""Test that a valid token can be decoded"""
user_id = str(uuid.uuid4())
token = create_access_token(subject=user_id)
# Decode token
payload = decode_token(token)
# Check that the subject matches
assert payload.sub == user_id
def test_decode_expired_token(self):
"""Test that an expired token raises TokenExpiredError"""
user_id = str(uuid.uuid4())
# Create a token that's already expired by directly manipulating the payload
now = datetime.now(timezone.utc)
expired_time = now - timedelta(hours=1) # 1 hour in the past
# Create the expired token manually
payload = {
"sub": user_id,
"exp": int(expired_time.timestamp()), # Set expiration in the past
"iat": int(now.timestamp()),
"jti": str(uuid.uuid4()),
"type": "access"
}
expired_token = jwt.encode(
payload,
settings.SECRET_KEY,
algorithm=settings.ALGORITHM
)
# Attempting to decode should raise TokenExpiredError
with pytest.raises(TokenExpiredError):
decode_token(expired_token)
def test_decode_invalid_token(self):
"""Test that an invalid token raises TokenInvalidError"""
invalid_token = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJpbnZhbGlkIn0.invalid-signature"
with pytest.raises(TokenInvalidError):
decode_token(invalid_token)
def test_decode_token_with_missing_sub(self):
"""Test that a token without 'sub' claim raises TokenMissingClaimError"""
# Create a token without a subject
now = datetime.now(timezone.utc)
payload = {
"exp": int((now + timedelta(minutes=30)).timestamp()),
"iat": int(now.timestamp()),
"jti": str(uuid.uuid4()),
"type": "access"
# No 'sub' claim
}
token = jwt.encode(
payload,
settings.SECRET_KEY,
algorithm=settings.ALGORITHM
)
with pytest.raises(TokenMissingClaimError):
decode_token(token)
def test_decode_token_with_wrong_type(self):
"""Test that verifying a token with wrong type raises TokenInvalidError"""
user_id = str(uuid.uuid4())
token = create_access_token(subject=user_id)
# Try to verify it as a refresh token
with pytest.raises(TokenInvalidError):
decode_token(token, verify_type="refresh")
def test_decode_with_invalid_payload(self):
"""Test that a token with invalid payload structure raises TokenInvalidError"""
# Create a token with an invalid payload structure - missing 'sub' which is required
# but including 'exp' to avoid the expiration check
now = datetime.now(timezone.utc)
payload = {
# Missing "sub" field which is required
"exp": int((now + timedelta(minutes=30)).timestamp()),
"iat": int(now.timestamp()),
"jti": str(uuid.uuid4()),
"invalid_field": "test"
}
token = jwt.encode(
payload,
settings.SECRET_KEY,
algorithm=settings.ALGORITHM
)
# Should raise TokenMissingClaimError due to missing 'sub'
with pytest.raises(TokenMissingClaimError):
decode_token(token)
# Create another token with invalid type for required field
payload = {
"sub": 123, # sub should be a string, not an integer
"exp": int((now + timedelta(minutes=30)).timestamp()),
}
token = jwt.encode(
payload,
settings.SECRET_KEY,
algorithm=settings.ALGORITHM
)
# Should raise TokenInvalidError due to ValidationError
with pytest.raises(TokenInvalidError):
decode_token(token)
def test_get_token_data(self):
"""Test extracting TokenData from a token"""
user_id = uuid.uuid4()
token = create_access_token(
subject=str(user_id),
claims={"is_superuser": True}
)
token_data = get_token_data(token)
assert token_data.user_id == user_id
assert token_data.is_superuser is True

View File

View File

@@ -0,0 +1,125 @@
import pytest
from app.crud.user import user as user_crud
from app.models.user import User
from app.schemas.users import UserCreate, UserUpdate
def test_create_user(db_session, user_create_data):
user_in = UserCreate(**user_create_data)
user_obj = user_crud.create(db_session, obj_in=user_in)
assert user_obj.email == user_create_data["email"]
assert user_obj.first_name == user_create_data["first_name"]
assert user_obj.last_name == user_create_data["last_name"]
assert user_obj.phone_number == user_create_data["phone_number"]
assert user_obj.is_superuser == user_create_data["is_superuser"]
assert user_obj.password_hash is not None
assert user_obj.id is not None
def test_get_user(db_session, mock_user):
# Using mock_user fixture instead of creating new user
stored_user = user_crud.get(db_session, id=mock_user.id)
assert stored_user
assert stored_user.id == mock_user.id
assert stored_user.email == mock_user.email
def test_get_user_by_email(db_session, mock_user):
stored_user = user_crud.get_by_email(db_session, email=mock_user.email)
assert stored_user
assert stored_user.id == mock_user.id
assert stored_user.email == mock_user.email
def test_update_user(db_session, mock_user):
update_data = UserUpdate(
first_name="Updated",
last_name="Name",
phone_number="+9876543210"
)
updated_user = user_crud.update(db_session, db_obj=mock_user, obj_in=update_data)
assert updated_user.first_name == "Updated"
assert updated_user.last_name == "Name"
assert updated_user.phone_number == "+9876543210"
assert updated_user.email == mock_user.email
def test_delete_user(db_session, mock_user):
user_crud.remove(db_session, id=mock_user.id)
deleted_user = user_crud.get(db_session, id=mock_user.id)
assert deleted_user is None
def test_get_multi_users(db_session, mock_user, user_create_data):
# Create additional users (mock_user is already in db)
users_data = [
{**user_create_data, "email": f"test{i}@example.com"}
for i in range(2) # Creating 2 more users + mock_user = 3 total
]
for user_data in users_data:
user_in = UserCreate(**user_data)
user_crud.create(db_session, obj_in=user_in)
users = user_crud.get_multi(db_session, skip=0, limit=10)
assert len(users) == 3
assert all(isinstance(user, User) for user in users)
def test_is_active(db_session, mock_user):
assert user_crud.is_active(mock_user) is True
# Test deactivating user
update_data = UserUpdate(is_active=False)
deactivated_user = user_crud.update(db_session, db_obj=mock_user, obj_in=update_data)
assert user_crud.is_active(deactivated_user) is False
def test_is_superuser(db_session, mock_user, user_create_data):
# mock_user is regular user
assert user_crud.is_superuser(mock_user) is False
# Create superuser
super_user_data = {**user_create_data, "email": "super@example.com", "is_superuser": True}
super_user_in = UserCreate(**super_user_data)
super_user = user_crud.create(db_session, obj_in=super_user_in)
assert user_crud.is_superuser(super_user) is True
# Additional test cases
def test_create_duplicate_email(db_session, mock_user):
user_data = UserCreate(
email=mock_user.email, # Try to create user with existing email
password="TestPassword123!",
first_name="Test",
last_name="User"
)
with pytest.raises(Exception): # Should raise an integrity error
user_crud.create(db_session, obj_in=user_data)
def test_update_user_preferences(db_session, mock_user):
preferences = {"theme": "dark", "notifications": True}
update_data = UserUpdate(preferences=preferences)
updated_user = user_crud.update(db_session, db_obj=mock_user, obj_in=update_data)
assert updated_user.preferences == preferences
def test_get_multi_users_pagination(db_session, user_create_data):
# Create 5 users
for i in range(5):
user_in = UserCreate(**{**user_create_data, "email": f"test{i}@example.com"})
user_crud.create(db_session, obj_in=user_in)
# Test pagination
first_page = user_crud.get_multi(db_session, skip=0, limit=2)
second_page = user_crud.get_multi(db_session, skip=2, limit=2)
assert len(first_page) == 2
assert len(second_page) == 2
assert first_page[0].id != second_page[0].id

View File

View File

@@ -0,0 +1,249 @@
# tests/models/test_user.py
import uuid
import pytest
from datetime import datetime
from sqlalchemy.exc import IntegrityError
from app.models.user import User
def test_create_user(db_session):
"""Test creating a basic user."""
# Arrange
user_id = uuid.uuid4()
new_user = User(
id=user_id,
email="test@example.com",
password_hash="hashedpassword",
first_name="Test",
last_name="User",
phone_number="1234567890",
is_active=True,
is_superuser=False,
preferences={"theme": "dark"},
)
db_session.add(new_user)
# Act
db_session.commit()
created_user = db_session.query(User).filter_by(email="test@example.com").first()
# Assert
assert created_user is not None
assert created_user.email == "test@example.com"
assert created_user.first_name == "Test"
assert created_user.last_name == "User"
assert created_user.phone_number == "1234567890"
assert created_user.is_active is True
assert created_user.is_superuser is False
assert created_user.preferences == {"theme": "dark"}
# UUID should be preserved
assert created_user.id == user_id
# Timestamps should be set
assert isinstance(created_user.created_at, datetime)
assert isinstance(created_user.updated_at, datetime)
def test_update_user(db_session):
"""Test updating an existing user."""
# Arrange - Create a user
user_id = uuid.uuid4()
user = User(
id=user_id,
email="update@example.com",
password_hash="hashedpassword",
first_name="Before",
last_name="Update",
)
db_session.add(user)
db_session.commit()
# Record the original creation timestamp
original_created_at = user.created_at
# Act - Update the user
user.first_name = "After"
user.last_name = "Updated"
user.phone_number = "9876543210"
user.preferences = {"theme": "light", "notifications": True}
db_session.commit()
# Fetch the updated user to verify changes were persisted
updated_user = db_session.query(User).filter_by(id=user_id).first()
# Assert
assert updated_user.first_name == "After"
assert updated_user.last_name == "Updated"
assert updated_user.phone_number == "9876543210"
assert updated_user.preferences == {"theme": "light", "notifications": True}
# created_at shouldn't change on update
assert updated_user.created_at == original_created_at
# updated_at should be newer than created_at
assert updated_user.updated_at > original_created_at
def test_delete_user(db_session):
"""Test deleting a user."""
# Arrange - Create a user
user_id = uuid.uuid4()
user = User(
id=user_id,
email="delete@example.com",
password_hash="hashedpassword",
first_name="Delete",
last_name="Me",
)
db_session.add(user)
db_session.commit()
# Act - Delete the user
db_session.delete(user)
db_session.commit()
# Assert
deleted_user = db_session.query(User).filter_by(id=user_id).first()
assert deleted_user is None
def test_user_unique_email_constraint(db_session):
"""Test that users cannot have duplicate emails."""
# Arrange - Create a user
user1 = User(
id=uuid.uuid4(),
email="duplicate@example.com",
password_hash="hashedpassword",
first_name="First",
last_name="User",
)
db_session.add(user1)
db_session.commit()
# Act & Assert - Try to create another user with the same email
user2 = User(
id=uuid.uuid4(),
email="duplicate@example.com", # Same email
password_hash="differenthash",
first_name="Second",
last_name="User",
)
db_session.add(user2)
# Should raise IntegrityError due to unique constraint
with pytest.raises(IntegrityError):
db_session.commit()
# Rollback for cleanup
db_session.rollback()
def test_user_required_fields(db_session):
"""Test that required fields are enforced."""
# Test each required field by creating a user without it
# Missing email
user_no_email = User(
id=uuid.uuid4(),
# email is missing
password_hash="hashedpassword",
first_name="Test",
last_name="User",
)
db_session.add(user_no_email)
with pytest.raises(IntegrityError):
db_session.commit()
db_session.rollback()
# Missing password_hash
user_no_password = User(
id=uuid.uuid4(),
email="nopassword@example.com",
# password_hash is missing
first_name="Test",
last_name="User",
)
db_session.add(user_no_password)
with pytest.raises(IntegrityError):
db_session.commit()
db_session.rollback()
def test_user_defaults(db_session):
"""Test that default values are correctly set."""
# Arrange - Create a minimal user with only required fields
minimal_user = User(
id=uuid.uuid4(),
email="minimal@example.com",
password_hash="hashedpassword",
first_name="Minimal",
last_name="User",
)
db_session.add(minimal_user)
db_session.commit()
# Act - Retrieve the user
created_user = db_session.query(User).filter_by(email="minimal@example.com").first()
# Assert - Check default values
assert created_user.is_active is True # Default should be True
assert created_user.is_superuser is False # Default should be False
assert created_user.phone_number is None # Optional field
assert created_user.preferences is None # Optional field
def test_user_string_representation(db_session):
"""Test the string representation of a user."""
# Arrange
user = User(
id=uuid.uuid4(),
email="repr@example.com",
password_hash="hashedpassword",
first_name="String",
last_name="Repr",
)
# Act & Assert
assert str(user) == "<User repr@example.com>"
assert repr(user) == "<User repr@example.com>"
def test_user_with_complex_json_preferences(db_session):
"""Test storing and retrieving complex JSON preferences."""
# Arrange - Create a user with nested JSON preferences
complex_preferences = {
"theme": {
"mode": "dark",
"colors": {
"primary": "#333",
"secondary": "#666"
}
},
"notifications": {
"email": True,
"sms": False,
"push": {
"enabled": True,
"quiet_hours": [22, 7]
}
},
"tags": ["important", "family", "events"]
}
user = User(
id=uuid.uuid4(),
email="complex@example.com",
password_hash="hashedpassword",
first_name="Complex",
last_name="JSON",
preferences=complex_preferences
)
db_session.add(user)
db_session.commit()
# Act - Retrieve the user
retrieved_user = db_session.query(User).filter_by(email="complex@example.com").first()
# Assert - The complex JSON should be preserved
assert retrieved_user.preferences == complex_preferences
assert retrieved_user.preferences["theme"]["colors"]["primary"] == "#333"
assert retrieved_user.preferences["notifications"]["push"]["quiet_hours"] == [22, 7]
assert "important" in retrieved_user.preferences["tags"]

View File

View File

@@ -0,0 +1,127 @@
# tests/schemas/test_user_schemas.py
import pytest
import re
from pydantic import ValidationError
from app.schemas.users import UserBase, UserCreate
class TestPhoneNumberValidation:
"""Tests for phone number validation in user schemas"""
def test_valid_swiss_numbers(self):
"""Test valid Swiss phone numbers are accepted"""
# International format
user = UserBase(email="test@example.com", first_name="Test", last_name="User", phone_number="+41791234567")
assert user.phone_number == "+41791234567"
# Local format
user = UserBase(email="test@example.com", first_name="Test", last_name="User", phone_number="0791234567")
assert user.phone_number == "0791234567"
# With formatting characters
user = UserBase(email="test@example.com", first_name="Test", last_name="User", phone_number="+41 79 123 45 67")
assert re.sub(r'[\s\-\(\)]', '', user.phone_number) == "+41791234567"
user = UserBase(email="test@example.com", first_name="Test", last_name="User", phone_number="079 123 45 67")
assert re.sub(r'[\s\-\(\)]', '', user.phone_number) == "0791234567"
user = UserBase(email="test@example.com", first_name="Test", last_name="User", phone_number="+41-79-123-45-67")
assert re.sub(r'[\s\-\(\)]', '', user.phone_number) == "+41791234567"
user = UserBase(email="test@example.com", first_name="Test", last_name="User", phone_number="079-123-45-67")
assert re.sub(r'[\s\-\(\)]', '', user.phone_number) == "0791234567"
user = UserBase(email="test@example.com", first_name="Test", last_name="User", phone_number="+41 (79) 123 45 67")
assert re.sub(r'[\s\-\(\)]', '', user.phone_number) == "+41791234567"
user = UserBase(email="test@example.com", first_name="Test", last_name="User", phone_number="079 (123) 45 67")
assert re.sub(r'[\s\-\(\)]', '', user.phone_number) == "0791234567"
def test_valid_italian_numbers(self):
"""Test valid Italian phone numbers are accepted"""
# International format
user = UserBase(email="test@example.com", first_name="Test", last_name="User", phone_number="+393451234567")
assert user.phone_number == "+393451234567"
user = UserBase(email="test@example.com", first_name="Test", last_name="User", phone_number="+39345123456")
assert user.phone_number == "+39345123456"
# Local format
user = UserBase(email="test@example.com", first_name="Test", last_name="User", phone_number="03451234567")
assert user.phone_number == "03451234567"
user = UserBase(email="test@example.com", first_name="Test", last_name="User", phone_number="0345123456789")
assert user.phone_number == "0345123456789"
# With formatting characters
user = UserBase(email="test@example.com", first_name="Test", last_name="User", phone_number="+39 345 123 4567")
assert re.sub(r'[\s\-\(\)]', '', user.phone_number) == "+393451234567"
user = UserBase(email="test@example.com", first_name="Test", last_name="User", phone_number="0345 123 4567")
assert re.sub(r'[\s\-\(\)]', '', user.phone_number) == "03451234567"
user = UserBase(email="test@example.com", first_name="Test", last_name="User", phone_number="+39-345-123-4567")
assert re.sub(r'[\s\-\(\)]', '', user.phone_number) == "+393451234567"
user = UserBase(email="test@example.com", first_name="Test", last_name="User", phone_number="0345-123-4567")
assert re.sub(r'[\s\-\(\)]', '', user.phone_number) == "03451234567"
user = UserBase(email="test@example.com", first_name="Test", last_name="User", phone_number="+39 (345) 123 4567")
assert re.sub(r'[\s\-\(\)]', '', user.phone_number) == "+393451234567"
user = UserBase(email="test@example.com", first_name="Test", last_name="User", phone_number="0345 (123) 4567")
assert re.sub(r'[\s\-\(\)]', '', user.phone_number) == "03451234567"
def test_none_phone_number(self):
"""Test that None is accepted as a valid value (optional phone number)"""
user = UserBase(email="test@example.com", first_name="Test", last_name="User", phone_number=None)
assert user.phone_number is None
def test_invalid_phone_numbers(self):
"""Test that invalid phone numbers are rejected"""
invalid_numbers = [
# Too short
"+12",
"012",
# Invalid characters
"+41xyz123456",
"079abc4567",
"123-abc-7890",
"+1(800)CALL-NOW",
# Completely invalid formats
"++4412345678", # Double plus
"()+41123456", # Misplaced parentheses
# Empty string
"",
# Spaces only
" ",
]
for number in invalid_numbers:
with pytest.raises(ValidationError):
UserBase(email="test@example.com", first_name="Test", last_name="User", phone_number=number)
def test_phone_validation_in_user_create(self):
"""Test that phone validation also works in UserCreate schema"""
# Valid phone number
user = UserCreate(
email="test@example.com",
first_name="Test",
last_name="User",
password="Password123",
phone_number="+41791234567"
)
assert user.phone_number == "+41791234567"
# Invalid phone number should raise ValidationError
with pytest.raises(ValidationError):
UserCreate(
email="test@example.com",
first_name="Test",
last_name="User",
password="Password123",
phone_number="invalid-number"
)

View File

View File

@@ -0,0 +1,252 @@
# tests/services/test_auth_service.py
import uuid
import pytest
from unittest.mock import patch
from app.core.auth import get_password_hash, verify_password, TokenExpiredError, TokenInvalidError
from app.models.user import User
from app.schemas.users import UserCreate, Token
from app.services.auth_service import AuthService, AuthenticationError
class TestAuthServiceAuthentication:
"""Tests for AuthService.authenticate_user method"""
def test_authenticate_valid_user(self, db_session, mock_user):
"""Test authenticating a user with valid credentials"""
# Set a known password for the mock user
password = "TestPassword123"
mock_user.password_hash = get_password_hash(password)
db_session.commit()
# Authenticate with correct credentials
user = AuthService.authenticate_user(
db=db_session,
email=mock_user.email,
password=password
)
assert user is not None
assert user.id == mock_user.id
assert user.email == mock_user.email
def test_authenticate_nonexistent_user(self, db_session):
"""Test authenticating with an email that doesn't exist"""
user = AuthService.authenticate_user(
db=db_session,
email="nonexistent@example.com",
password="password"
)
assert user is None
def test_authenticate_with_wrong_password(self, db_session, mock_user):
"""Test authenticating with the wrong password"""
# Set a known password for the mock user
password = "TestPassword123"
mock_user.password_hash = get_password_hash(password)
db_session.commit()
# Authenticate with wrong password
user = AuthService.authenticate_user(
db=db_session,
email=mock_user.email,
password="WrongPassword123"
)
assert user is None
def test_authenticate_inactive_user(self, db_session, mock_user):
"""Test authenticating an inactive user"""
# Set a known password and make user inactive
password = "TestPassword123"
mock_user.password_hash = get_password_hash(password)
mock_user.is_active = False
db_session.commit()
# Should raise AuthenticationError
with pytest.raises(AuthenticationError):
AuthService.authenticate_user(
db=db_session,
email=mock_user.email,
password=password
)
class TestAuthServiceUserCreation:
"""Tests for AuthService.create_user method"""
def test_create_new_user(self, db_session):
"""Test creating a new user"""
user_data = UserCreate(
email="newuser@example.com",
password="TestPassword123",
first_name="New",
last_name="User",
phone_number="1234567890"
)
user = AuthService.create_user(db=db_session, user_data=user_data)
# Verify user was created with correct data
assert user is not None
assert user.email == user_data.email
assert user.first_name == user_data.first_name
assert user.last_name == user_data.last_name
assert user.phone_number == user_data.phone_number
# Verify password was hashed
assert user.password_hash != user_data.password
assert verify_password(user_data.password, user.password_hash)
# Verify default values
assert user.is_active is True
assert user.is_superuser is False
def test_create_user_with_existing_email(self, db_session, mock_user):
"""Test creating a user with an email that already exists"""
user_data = UserCreate(
email=mock_user.email, # Use existing email
password="TestPassword123",
first_name="Duplicate",
last_name="User"
)
# Should raise AuthenticationError
with pytest.raises(AuthenticationError):
AuthService.create_user(db=db_session, user_data=user_data)
class TestAuthServiceTokens:
"""Tests for AuthService token-related methods"""
def test_create_tokens(self, mock_user):
"""Test creating access and refresh tokens for a user"""
tokens = AuthService.create_tokens(mock_user)
# Verify token structure
assert isinstance(tokens, Token)
assert tokens.access_token is not None
assert tokens.refresh_token is not None
assert tokens.token_type == "bearer"
# This is a more in-depth test that would decode the tokens to verify claims
# but we'll rely on the auth module tests for token verification
def test_refresh_tokens(self, db_session, mock_user):
"""Test refreshing tokens with a valid refresh token"""
# Create initial tokens
initial_tokens = AuthService.create_tokens(mock_user)
# Refresh tokens
new_tokens = AuthService.refresh_tokens(
db=db_session,
refresh_token=initial_tokens.refresh_token
)
# Verify new tokens are different from old ones
assert new_tokens.access_token != initial_tokens.access_token
assert new_tokens.refresh_token != initial_tokens.refresh_token
def test_refresh_tokens_with_invalid_token(self, db_session):
"""Test refreshing tokens with an invalid token"""
# Create an invalid token
invalid_token = "invalid.token.string"
# Should raise TokenInvalidError
with pytest.raises(TokenInvalidError):
AuthService.refresh_tokens(
db=db_session,
refresh_token=invalid_token
)
def test_refresh_tokens_with_access_token(self, db_session, mock_user):
"""Test refreshing tokens with an access token instead of refresh token"""
# Create tokens
tokens = AuthService.create_tokens(mock_user)
# Try to refresh with access token
with pytest.raises(TokenInvalidError):
AuthService.refresh_tokens(
db=db_session,
refresh_token=tokens.access_token
)
def test_refresh_tokens_with_nonexistent_user(self, db_session):
"""Test refreshing tokens for a user that doesn't exist in the database"""
# Create a token for a non-existent user
non_existent_id = str(uuid.uuid4())
with patch('app.core.auth.decode_token'), patch('app.core.auth.get_token_data') as mock_get_data:
# Mock the token data to return a non-existent user ID
mock_get_data.return_value.user_id = uuid.UUID(non_existent_id)
# Should raise TokenInvalidError
with pytest.raises(TokenInvalidError):
AuthService.refresh_tokens(
db=db_session,
refresh_token="some.refresh.token"
)
class TestAuthServicePasswordChange:
"""Tests for AuthService.change_password method"""
def test_change_password(self, db_session, mock_user):
"""Test changing a user's password"""
# Set a known password for the mock user
current_password = "CurrentPassword123"
mock_user.password_hash = get_password_hash(current_password)
db_session.commit()
# Change password
new_password = "NewPassword456"
result = AuthService.change_password(
db=db_session,
user_id=mock_user.id,
current_password=current_password,
new_password=new_password
)
# Verify operation was successful
assert result is True
# Refresh user from DB
db_session.refresh(mock_user)
# Verify old password no longer works
assert not verify_password(current_password, mock_user.password_hash)
# Verify new password works
assert verify_password(new_password, mock_user.password_hash)
def test_change_password_wrong_current_password(self, db_session, mock_user):
"""Test changing password with incorrect current password"""
# Set a known password for the mock user
current_password = "CurrentPassword123"
mock_user.password_hash = get_password_hash(current_password)
db_session.commit()
# Try to change password with wrong current password
wrong_password = "WrongPassword123"
with pytest.raises(AuthenticationError):
AuthService.change_password(
db=db_session,
user_id=mock_user.id,
current_password=wrong_password,
new_password="NewPassword456"
)
# Verify password was not changed
assert verify_password(current_password, mock_user.password_hash)
def test_change_password_nonexistent_user(self, db_session):
"""Test changing password for a user that doesn't exist"""
non_existent_id = uuid.uuid4()
with pytest.raises(AuthenticationError):
AuthService.change_password(
db=db_session,
user_id=non_existent_id,
current_password="CurrentPassword123",
new_password="NewPassword456"
)