89 lines
3.4 KiB
Python
89 lines
3.4 KiB
Python
"""Unit tests for UserService using in-memory ports."""
|
|
|
|
import pytest
|
|
from app.application.auth_service import AuthService
|
|
from app.application.user_service import UserService
|
|
from app.core.config import Settings
|
|
from app.core.security import Argon2PasswordHasher, JwtTokenService
|
|
from app.domain.errors import AlreadyExistsError, AuthenticationError, NotFoundError
|
|
|
|
from tests.fakes import InMemoryRefreshTokenRepository, InMemoryUserRepository
|
|
|
|
|
|
@pytest.fixture
|
|
def env() -> tuple[UserService, AuthService]:
|
|
users = InMemoryUserRepository()
|
|
refresh = InMemoryRefreshTokenRepository()
|
|
hasher = Argon2PasswordHasher()
|
|
tokens = JwtTokenService(Settings(jwt_secret="u-test-secret"))
|
|
user_svc = UserService(users=users, refresh_tokens=refresh, hasher=hasher)
|
|
auth = AuthService(users=users, refresh_tokens=refresh, hasher=hasher, tokens=tokens)
|
|
return user_svc, auth
|
|
|
|
|
|
async def test_create_user_duplicate_username(env: tuple[UserService, AuthService]) -> None:
|
|
user_svc, _ = env
|
|
await user_svc.create_user(username="bob", password="password123")
|
|
with pytest.raises(AlreadyExistsError):
|
|
await user_svc.create_user(username="bob", password="another-one")
|
|
|
|
|
|
async def test_get_unknown_user_raises(env: tuple[UserService, AuthService]) -> None:
|
|
import uuid
|
|
|
|
user_svc, _ = env
|
|
with pytest.raises(NotFoundError):
|
|
await user_svc.get_user(uuid.uuid4())
|
|
|
|
|
|
async def test_change_password_requires_current(env: tuple[UserService, AuthService]) -> None:
|
|
user_svc, auth = env
|
|
user = await user_svc.create_user(username="bob", password="password123")
|
|
|
|
with pytest.raises(AuthenticationError):
|
|
await user_svc.change_password(
|
|
user.id, current_password="wrong", new_password="newpassword1"
|
|
)
|
|
|
|
await user_svc.change_password(
|
|
user.id, current_password="password123", new_password="newpassword1"
|
|
)
|
|
# New password works, old one no longer.
|
|
await auth.login("bob", "newpassword1")
|
|
with pytest.raises(AuthenticationError):
|
|
await auth.login("bob", "password123")
|
|
|
|
|
|
async def test_change_password_revokes_sessions(env: tuple[UserService, AuthService]) -> None:
|
|
user_svc, auth = env
|
|
user = await user_svc.create_user(username="bob", password="password123")
|
|
pair = await auth.login("bob", "password123")
|
|
|
|
await user_svc.change_password(
|
|
user.id, current_password="password123", new_password="newpassword1"
|
|
)
|
|
with pytest.raises(AuthenticationError):
|
|
await auth.refresh(pair.refresh.encoded)
|
|
|
|
|
|
async def test_reset_password_revokes_sessions(env: tuple[UserService, AuthService]) -> None:
|
|
user_svc, auth = env
|
|
user = await user_svc.create_user(username="bob", password="password123")
|
|
pair = await auth.login("bob", "password123")
|
|
|
|
await user_svc.reset_password(user.id, new_password="adminset12")
|
|
with pytest.raises(AuthenticationError):
|
|
await auth.refresh(pair.refresh.encoded)
|
|
await auth.login("bob", "adminset12")
|
|
|
|
|
|
async def test_deactivate_revokes_sessions(env: tuple[UserService, AuthService]) -> None:
|
|
user_svc, auth = env
|
|
user = await user_svc.create_user(username="bob", password="password123")
|
|
pair = await auth.login("bob", "password123")
|
|
|
|
deactivated = await user_svc.deactivate(user.id)
|
|
assert deactivated.is_active is False
|
|
with pytest.raises(AuthenticationError):
|
|
await auth.refresh(pair.refresh.encoded)
|