Unit testing AuthService

This commit is contained in:
2025-10-18 12:26:55 +02:00
commit 79a31ecf40
26 changed files with 3467 additions and 0 deletions

0
tests/__init__.py Normal file
View File

0
tests/core/__init__.py Normal file
View File

106
tests/core/conftest.py Normal file
View File

@@ -0,0 +1,106 @@
# tests/core/conftest.py
import shutil
from pathlib import Path
from unittest.mock import MagicMock
import pytest
from my_auth.core.password import PasswordManager
from my_auth.core.token import TokenManager
from src.my_auth.core.auth import AuthService
from src.my_auth.models.user import UserCreate
from src.my_auth.persistence.sqlite import SQLiteUserRepository, SQLiteTokenRepository
@pytest.fixture
def test_user_data_create():
"""Provides valid data for creating a test user."""
return UserCreate(
email="test.service@example.com",
username="TestServiceUser",
password="ValidPassword123!",
roles=["member"],
user_settings={"theme": "dark"}
)
@pytest.fixture
def test_user_hashed_password():
"""Provides a dummy hashed password (only used internally by service)"""
# Note: In service tests, we rely on the service to do the hashing/verification,
# but this is kept for completeness if needed elsewhere.
return "$2b$12$R.S/XfI2tQYt3Kk.iF1XwOQz0Qe.L0T0mD/O1H8E2V5D4Q6F7G8H9I0"
@pytest.fixture()
def sqlite_db_path(tmp_path_factory):
"""
Creates a temporary directory and an SQLite file path for the test session.
The directory is deleted after the session.
"""
temp_dir = tmp_path_factory.mktemp("sqlite_auth_service_test")
db_file: Path = temp_dir / "auth_service_test.db"
yield str(db_file)
# Cleanup phase
try:
if temp_dir.exists():
shutil.rmtree(temp_dir)
except OSError as e:
print(f"Error during cleanup of temporary DB directory: {e}")
@pytest.fixture
def user_repository(sqlite_db_path: str) -> SQLiteUserRepository:
"""Provides a real SQLiteUserRepository instance."""
return SQLiteUserRepository(db_path=sqlite_db_path)
@pytest.fixture
def token_repository(sqlite_db_path: str) -> SQLiteTokenRepository:
"""Provides a real SQLiteTokenRepository instance."""
return SQLiteTokenRepository(db_path=sqlite_db_path)
@pytest.fixture
def mock_password_manager() -> PasswordManager:
"""Provides a PasswordManager instance for injection (low rounds for speed)."""
mock = MagicMock(spec=PasswordManager)
mock.hash_password.return_value = "PREDICTABLE_HASHED_PASSWORD_FOR_TESTING"
mock.verify_password.return_value = True
return mock
@pytest.fixture
def mock_token_manager() -> TokenManager:
"""Provides a TokenManager instance for injection (fast expiration settings)."""
mock = MagicMock(spec=TokenManager)
mock.create_access_token.return_value = "MOCKED_ACCESS_TOKEN"
mock.create_refresh_token.return_value = "MOCKED_REFRESH_TOKEN"
return mock
# --- Service Fixture ---
@pytest.fixture
def auth_service(user_repository: SQLiteUserRepository,
token_repository: SQLiteTokenRepository,
mock_password_manager: PasswordManager,
mock_token_manager: TokenManager
) -> AuthService:
"""
Provides an AuthService instance initialized with real repositories.
"""
# NOTE: To test hashing/verification, we must ensure password_hash_rounds is low.
# NOTE: To simplify JWT testing, we mock the internal components (hashing/JWT)
# as the AuthService shouldn't be responsible for these core algorithms,
# only for orchestrating them. If your service integrates them directly,
# we'll need to patch them below.
return AuthService(
user_repository=user_repository,
token_repository=token_repository,
password_manager=mock_password_manager,
token_manager=mock_token_manager,
)

View File

@@ -0,0 +1,270 @@
# tests/core/test_auth_service.py
from datetime import datetime, timedelta
from unittest.mock import MagicMock, patch
import pytest
from src.my_auth.core.auth import AuthService
from src.my_auth.exceptions import UserAlreadyExistsError, InvalidCredentialsError, InvalidTokenError, ExpiredTokenError
from src.my_auth.models.token import TokenData
from src.my_auth.models.user import UserCreate, UserUpdate
class TestAuthServiceRegisterLogin(object):
"""Tests for the user registration and login processes."""
# The mocks are injected via the auth_service fixture from conftest.py
def test_register_success(self, auth_service: AuthService,
test_user_data_create: UserCreate):
"""Success: Registration works and stores the predictable hash."""
user = auth_service.register(test_user_data_create)
assert user is not None
assert user.hashed_password == "PREDICTABLE_HASHED_PASSWORD_FOR_TESTING"
auth_service.password_manager.hash_password.assert_called_once_with(test_user_data_create.password)
def test_register_failure_if_email_exists(self, auth_service: AuthService,
test_user_data_create: UserCreate):
"""Failure: Cannot register if the email already exists."""
auth_service.register(test_user_data_create)
with pytest.raises(UserAlreadyExistsError):
auth_service.register(test_user_data_create)
def test_login_success(self, auth_service: AuthService, test_user_data_create: UserCreate):
"""Success: Logging in with correct credentials generates and saves tokens."""
# Execute register to insert the user
auth_service.register(test_user_data_create)
# Execute login
user, tokens = auth_service.login(test_user_data_create.email, test_user_data_create.password)
assert user.email == test_user_data_create.email
auth_service.password_manager.verify_password.assert_called_once()
auth_service.token_manager.create_access_token.assert_called_once()
assert tokens.access_token == "MOCKED_ACCESS_TOKEN"
assert tokens.refresh_token == "MOCKED_REFRESH_TOKEN"
def test_login_failure_invalid_password(self, auth_service: AuthService, # Removed mock_password_manager injection
test_user_data_create: UserCreate):
"""Failure: Login fails with InvalidCredentialsError if the password is wrong."""
# Access the manager via the service instance
pm = auth_service.password_manager
# Setup: Mock hash for registration
pm.hash_password.return_value = "HASH"
auth_service.register(test_user_data_create)
# Mock verify for failure
pm.verify_password.return_value = False
with pytest.raises(InvalidCredentialsError):
auth_service.login(test_user_data_create.email, "WrongPassword!")
# Restore the default mock behavior defined in conftest for subsequent tests
pm.hash_password.return_value = "PREDICTABLE_HASHED_PASSWORD_FOR_TESTING"
pm.verify_password.return_value = True
def test_login_failure_user_not_found(self, auth_service: AuthService):
"""Failure: Login fails if the user does not exist."""
with pytest.raises(InvalidCredentialsError):
auth_service.login("non.existent@example.com", "AnyPassword")
class TestAuthServiceTokenManagement(object):
"""Tests for token-related flows (Refresh, Logout, GetCurrentUser)."""
@pytest.fixture(autouse=True)
def setup_user_and_token(self, auth_service: AuthService, test_user_data_create: UserCreate):
"""
Sets up a registered user and an initial set of tokens for management tests.
Temporarily overrides manager behavior for setup.
"""
# Temporarily set up predictable values for the registration/login flow within the fixture setup
pm = auth_service.password_manager
tm = auth_service.token_manager
original_hash = pm.hash_password.return_value
original_verify = pm.verify_password.return_value
original_access = tm.create_access_token.return_value
original_refresh = tm.create_refresh_token.return_value
pm.hash_password.return_value = "HASHED_PASS"
pm.verify_password.return_value = True
tm.create_access_token.return_value = "SETUP_ACCESS_TOKEN"
tm.create_refresh_token.return_value = "SETUP_REFRESH_TOKEN"
user = auth_service.register(test_user_data_create)
_, tokens = auth_service.login(test_user_data_create.email, test_user_data_create.password)
self.user = user
self.refresh_token = tokens.refresh_token
self.access_token = tokens.access_token
# Restore mock values to default conftest behavior for the actual tests
pm.hash_password.return_value = original_hash
pm.verify_password.return_value = original_verify
tm.create_access_token.return_value = original_access
tm.create_refresh_token.return_value = original_refresh
def test_refresh_access_token_success(self, auth_service: AuthService):
"""Success: Refreshing an access token works with a valid refresh token."""
# Access the manager via the service instance
tm = auth_service.token_manager
# Use patch.object on the *instance* for granular control within the test
with patch.object(tm, 'create_access_token',
return_value="NEW_MOCKED_ACCESS_TOKEN") as mock_create_access:
tokens = auth_service.refresh_access_token(self.refresh_token)
assert tokens.access_token == "NEW_MOCKED_ACCESS_TOKEN"
mock_create_access.assert_called_once()
def test_refresh_access_token_failure_invalid_token(self, auth_service: AuthService):
"""Failure: Refreshing fails if the token is invalid (revoked, expired, etc.)."""
auth_service.logout(self.refresh_token)
with pytest.raises(InvalidTokenError):
auth_service.refresh_access_token("invalid_token")
def test_logout_success(self, auth_service: AuthService):
"""Success: Logout revokes the specified refresh token."""
result = auth_service.logout(self.refresh_token)
assert result is True
with pytest.raises(InvalidTokenError):
auth_service.refresh_access_token(self.refresh_token)
def test_get_current_user_success(self, auth_service: AuthService):
"""Success: Getting the current user works by successfully decoding the JWT."""
# Mock the decoder to simulate a decoded payload
with patch.object(auth_service.token_manager, 'decode_access_token',
return_value={"sub": self.user.id}) as mock_decode:
user = auth_service.get_current_user("dummy_jwt")
assert user.id == self.user.id
mock_decode.assert_called_once()
def test_get_current_user_failure_invalid_token(self, auth_service: AuthService):
"""Failure: Getting the current user fails if the access token is invalid/expired."""
with patch.object(auth_service.token_manager, 'decode_access_token',
side_effect=InvalidTokenError("Invalid signature")):
with pytest.raises(InvalidTokenError):
auth_service.get_current_user("invalid_access_jwt")
with patch.object(auth_service.token_manager, 'decode_access_token',
side_effect=ExpiredTokenError("Token expired")):
with pytest.raises(ExpiredTokenError):
auth_service.get_current_user("expired_access_jwt")
class TestAuthServiceResetVerification(object):
"""Tests for password reset and email verification flows."""
@pytest.fixture(autouse=True)
def setup_user(self, auth_service: AuthService, test_user_data_create: UserCreate):
"""Sets up a registered user using a mock hash for speed."""
pm = auth_service.password_manager
original_hash = pm.hash_password.return_value
# Temporarily set hash for setup
pm.hash_password.return_value = "HASHED_PASS"
user = auth_service.register(test_user_data_create)
self.user = user
# Restore hash mock
pm.hash_password.return_value = original_hash
@patch('src.my_auth.core.email.send_email')
def test_request_password_reset_success(self, mock_send_email: MagicMock, auth_service: AuthService):
"""Success: Requesting a password reset generates a token and sends an email."""
tm = auth_service.token_manager
with patch.object(tm, 'create_password_reset_token',
return_value="MOCKED_RESET_TOKEN") as mock_create_token:
token_string = auth_service.request_password_reset(self.user.email)
assert token_string == "MOCKED_RESET_TOKEN"
mock_create_token.assert_called_once()
mock_send_email.assert_called_once()
def test_reset_password_success(self, auth_service: AuthService):
"""Success: Resetting the password works with a valid token."""
# Setup: Manually create a valid reset token
auth_service.token_repository.save_token(
TokenData(token="valid_reset_token", token_type="password_reset", user_id=self.user.id,
expires_at=datetime.now() + timedelta(minutes=10))
)
# Patch the PasswordManager instance to control the hash output
pm = auth_service.password_manager
with patch.object(pm, 'hash_password',
return_value="NEW_HASHED_PASSWORD_FOR_RESET") as mock_hash:
new_password = "NewPassword123!"
result = auth_service.reset_password("valid_reset_token", new_password)
assert result is True
mock_hash.assert_called_once_with(new_password)
# Verification: Check that user data was updated
updated_user = auth_service.user_repository.get_user_by_id(self.user.id)
assert updated_user.hashed_password == "NEW_HASHED_PASSWORD_FOR_RESET"
@patch('src.my_auth.core.email.send_email')
def test_request_email_verification_success(self, mock_send_email: MagicMock, auth_service: AuthService):
"""Success: Requesting verification generates a token and sends an email."""
tm = auth_service.token_manager
with patch.object(tm, 'create_email_verification_token',
return_value="MOCKED_JWT_VERIFY_TOKEN") as mock_create_token:
token_string = auth_service.request_email_verification(self.user.email)
assert token_string == "MOCKED_JWT_VERIFY_TOKEN"
mock_create_token.assert_called_once_with(self.user.email)
mock_send_email.assert_called_once()
def test_verify_email_success(self, auth_service: AuthService):
"""Success: Verification updates the user's status."""
# The token_manager is mocked in conftest, so we must access its real create method
# or rely on the mock's return value to get a token string to use in the call.
# Since we need a real token for the decode logic to pass, we need to bypass the mock here.
# We will temporarily use the real TokenManager to create a valid, decodable token.
# This requires an *unmocked* token manager instance, which is tricky in this setup.
# Alternative: Temporarily inject a real TokenManager for this test (or rely on a non-mocked method)
# Assuming TokenManager.create_email_verification_token can be mocked to return a static string
# and TokenManager.decode_email_verification_token can be patched to simulate success.
# Since the method calls decode_email_verification_token internally, we mock the output of the decode step.
# Setup: Ensure user is unverified
auth_service.user_repository.update_user(self.user.id, UserUpdate(is_verified=False))
tm = auth_service.token_manager
# Mock the decode step to ensure it returns the email used for verification
with patch.object(tm, 'decode_email_verification_token', return_value=self.user.email) as mock_decode:
# Test (we use a dummy token string as the decode step is mocked)
result = auth_service.verify_email("dummy_verification_token")
assert result is True
mock_decode.assert_called_once()
# Verification: User is verified
updated_user = auth_service.user_repository.get_user_by_id(self.user.id)
assert updated_user.is_verified is True

View File

View File

@@ -0,0 +1,75 @@
from uuid import uuid4
from datetime import datetime, timedelta
from pathlib import Path
import pytest
from my_auth.models.token import TokenData
from my_auth.models.user import UserCreate
from my_auth.persistence.sqlite import SQLiteUserRepository, SQLiteTokenRepository
@pytest.fixture
def test_user_data_create():
"""Provides valid data for creating a test user."""
return UserCreate(
email="test.user@example.com",
username="TestUser",
password="ValidPassword123!", # Meets all strength criteria
roles=["member"],
user_settings={"theme": "dark"}
)
@pytest.fixture
def test_user_hashed_password():
"""Provides a dummy hashed password for persistence (hashing is tested elsewhere)."""
return "$2b$12$R.S/XfI2tQYt3Kk.iF1XwOQz0Qe.L0T0mD/O1H8E2V5D4Q6F7G8H9I0"
@pytest.fixture
def test_token_data():
"""Provides valid data for a refresh token."""
now = datetime.now()
return TokenData(
token=f"opaque_refresh_token_{uuid4()}",
token_type="refresh",
user_id="user_id_for_token_test",
expires_at=now + timedelta(days=7),
is_revoked=False,
created_at=now
)
@pytest.fixture()
def sqlite_db_path(tmp_path_factory):
"""
Creates a temporary directory and an SQLite file path for the test session.
The directory and its contents are deleted after the session.
"""
temp_dir = tmp_path_factory.mktemp("sqlite_auth_test")
db_file: Path = temp_dir / "auth_test.db"
yield str(db_file)
try:
if temp_dir.exists():
import shutil
shutil.rmtree(temp_dir)
except OSError as e:
# Handle case where directory might be locked, though rare in tests
print(f"Error during cleanup of temporary DB directory: {e}")
@pytest.fixture
def user_repository(sqlite_db_path: str) -> SQLiteUserRepository:
"""Provides an instance of SQLiteUserRepository initialized with the in-memory DB."""
# Assuming the repository takes the connection object or path (using connection here)
return SQLiteUserRepository(sqlite_db_path)
@pytest.fixture
def token_repository(sqlite_db_path: str) -> SQLiteTokenRepository:
"""Provides an instance of SQLiteTokenRepository initialized with the in-memory DB."""
# Assuming the repository takes the connection object or path (using connection here)
return SQLiteTokenRepository(sqlite_db_path)

View File

@@ -0,0 +1,87 @@
# tests/persistence/test_sqlite_token.py
from datetime import datetime, timedelta
from my_auth.models.token import TokenData
from my_auth.persistence.sqlite import SQLiteTokenRepository
def test_i_can_save_and_retrieve_token(token_repository: SQLiteTokenRepository,
test_token_data: TokenData):
"""Verifies token saving and successful retrieval by token string and type."""
# 1. Save Token
token_repository.save_token(test_token_data)
# 2. Retrieve Token
retrieved_token = token_repository.get_token(test_token_data.token, test_token_data.token_type)
# Assertions
assert retrieved_token is not None
assert retrieved_token.token == test_token_data.token
assert retrieved_token.user_id == test_token_data.user_id
assert retrieved_token.is_revoked is False
assert retrieved_token.token_type == test_token_data.token_type
def test_i_can_revoke_token(token_repository: SQLiteTokenRepository,
test_token_data: TokenData):
"""Verifies a token can be revoked and its revoked status is updated."""
# Setup: Save the token
token_repository.save_token(test_token_data)
# 1. Revoke the token
was_revoked = token_repository.revoke_token(test_token_data.token)
assert was_revoked is True
# 2. Retrieve and check status
revoked_token = token_repository.get_token(test_token_data.token, test_token_data.token_type)
assert revoked_token is not None
assert revoked_token.is_revoked is True
# 3. Attempt to revoke a non-existent token
was_revoked_again = token_repository.revoke_token("non_existent_token")
assert was_revoked_again is False
def test_i_can_use_is_token_valid_for_valid_token(token_repository: SQLiteTokenRepository,
test_token_data: TokenData):
"""Verifies the convenience method returns True for a fresh, unexpired token."""
token_repository.save_token(test_token_data)
is_valid = token_repository.is_token_valid(test_token_data.token, test_token_data.token_type)
assert is_valid is True
is_valid = token_repository.is_token_valid("non_existent_token", test_token_data.token_type)
assert is_valid is False
def test_i_can_use_is_token_valid_for_revoked_token(token_repository: SQLiteTokenRepository,
test_token_data: TokenData):
"""Verifies is_token_valid returns False for a token marked as revoked."""
token_repository.save_token(test_token_data)
token_repository.revoke_token(test_token_data.token)
is_valid = token_repository.is_token_valid(test_token_data.token, test_token_data.token_type)
assert is_valid is False
def test_i_can_use_is_token_valid_for_expired_token(token_repository: SQLiteTokenRepository):
"""Verifies is_token_valid returns False for a token whose expiration is in the past."""
expired_token_data = TokenData(
token="expired_token_test",
token_type="password_reset",
user_id="user_id_expired",
expires_at=datetime.now() - timedelta(hours=1), # Set expiration to 1 hour ago
is_revoked=False,
created_at=datetime.now() - timedelta(hours=2)
)
token_repository.save_token(expired_token_data)
is_valid = token_repository.is_token_valid(expired_token_data.token, expired_token_data.token_type)
assert is_valid is False

View File

@@ -0,0 +1,155 @@
# tests/persistence/test_sqlite_user.py
import pytest
import json
from datetime import datetime
from my_auth.persistence.sqlite import SQLiteUserRepository
from my_auth.models.user import UserCreate, UserUpdate
from my_auth.exceptions import UserAlreadyExistsError, UserNotFoundError
def test_i_can_create_and_retrieve_user_by_email(user_repository: SQLiteUserRepository,
test_user_data_create: UserCreate,
test_user_hashed_password: str):
"""Verifies user creation and successful retrieval by email."""
# 1. Create User
created_user = user_repository.create_user(
user_data=test_user_data_create,
hashed_password=test_user_hashed_password
)
# Assertions on creation
assert created_user is not None
assert created_user.email == test_user_data_create.email
assert created_user.hashed_password == test_user_hashed_password
assert created_user.is_active is True
assert created_user.is_verified is False
assert created_user.id is not None
assert isinstance(created_user.created_at, datetime)
# 2. Retrieve User
retrieved_user = user_repository.get_user_by_email(test_user_data_create.email)
# Assertions on retrieval
assert retrieved_user is not None
assert retrieved_user.id == created_user.id
assert retrieved_user.username == "TestUser"
def test_i_can_retrieve_user_by_id(user_repository: SQLiteUserRepository,
test_user_data_create: UserCreate,
test_user_hashed_password: str):
"""Verifies user retrieval by unique ID."""
created_user = user_repository.create_user(test_user_data_create, test_user_hashed_password)
retrieved_user = user_repository.get_user_by_id(created_user.id)
assert retrieved_user is not None
assert retrieved_user.email == created_user.email
def test_i_cannot_create_user_if_email_exists(user_repository: SQLiteUserRepository,
test_user_data_create: UserCreate,
test_user_hashed_password: str):
"""Ensures that creating a user with an existing email raises UserAlreadyExistsError."""
# First creation should succeed
user_repository.create_user(test_user_data_create, test_user_hashed_password)
# Second creation with same email should fail
with pytest.raises(UserAlreadyExistsError):
user_repository.create_user(test_user_data_create, test_user_hashed_password)
def test_i_can_check_if_email_exists(user_repository: SQLiteUserRepository,
test_user_data_create: UserCreate,
test_user_hashed_password: str):
"""Verifies the email_exists method returns correct boolean results."""
email = test_user_data_create.email
non_existent_email = "non.existent@example.com"
# 1. Check before creation (Should be False)
assert user_repository.email_exists(email) is False
user_repository.create_user(test_user_data_create, test_user_hashed_password)
# 2. Check after creation (Should be True)
assert user_repository.email_exists(email) is True
# 3. Check for another non-existent email
assert user_repository.email_exists(non_existent_email) is False
def test_i_can_update_username_and_roles(user_repository: SQLiteUserRepository,
test_user_data_create: UserCreate,
test_user_hashed_password: str):
"""Tests partial update of user fields (username and roles)."""
created_user = user_repository.create_user(test_user_data_create, test_user_hashed_password)
updates = UserUpdate(username="NewUsername", roles=["admin", "staff"])
updated_user = user_repository.update_user(created_user.id, updates)
assert updated_user.username == "NewUsername"
assert updated_user.roles == ["admin", "staff"]
# Check that unrelated fields remain the same
assert updated_user.email == created_user.email
# Check that update timestamp changed
assert updated_user.updated_at > created_user.updated_at
assert updated_user.is_verified == created_user.is_verified
def test_i_can_update_is_active_status(user_repository: SQLiteUserRepository,
test_user_data_create: UserCreate,
test_user_hashed_password: str):
"""Tests the specific update of the 'is_active' status."""
created_user = user_repository.create_user(test_user_data_create, test_user_hashed_password)
# Deactivate
updates_deactivate = UserUpdate(is_active=False)
deactivated_user = user_repository.update_user(created_user.id, updates_deactivate)
assert deactivated_user.is_active is False
# Reactivate
updates_activate = UserUpdate(is_active=True)
reactivated_user = user_repository.update_user(created_user.id, updates_activate)
assert reactivated_user.is_active is True
def test_i_cannot_update_non_existent_user(user_repository: SQLiteUserRepository):
"""Ensures that updating a user with an unknown ID raises UserNotFoundError."""
non_existent_id = "unknown_id_123"
updates = UserUpdate(username="Phantom")
with pytest.raises(UserNotFoundError):
user_repository.update_user(non_existent_id, updates)
def test_i_can_delete_user(user_repository: SQLiteUserRepository,
test_user_data_create: UserCreate,
test_user_hashed_password: str):
"""Verifies user deletion and subsequent failure to retrieve."""
created_user = user_repository.create_user(test_user_data_create, test_user_hashed_password)
# 1. Delete the user
was_deleted = user_repository.delete_user(created_user.id)
assert was_deleted is True
# 2. Verify deletion by attempting retrieval
retrieved_user = user_repository.get_user_by_id(created_user.id)
assert retrieved_user is None
# 3. Verify attempting to delete again returns False
was_deleted_again = user_repository.delete_user(created_user.id)
assert was_deleted_again is False
def test_i_cannot_retrieve_non_existent_user_by_email(user_repository: SQLiteUserRepository):
"""Ensures retrieval by email returns None for non-existent email."""
retrieved_user = user_repository.get_user_by_email("ghost@example.com")
assert retrieved_user is None