diff --git a/tests/unit/test_auth_register_password_reset.py b/tests/unit/test_auth_register_password_reset.py new file mode 100755 index 000000000..95dcc2cd9 --- /dev/null +++ b/tests/unit/test_auth_register_password_reset.py @@ -0,0 +1,705 @@ +"""Auth register + password reset use cases unit tests. + +Covers RegisterUserUseCase, VerifyEmailUseCase, +RequestPasswordResetUseCase, ResetPasswordUseCase. +""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +from unittest.mock import MagicMock + +import pytest + +from packages.application.auth.password_hasher import password_hasher +from packages.application.auth.password_reset_use_case import ( + RequestPasswordResetRequest, + RequestPasswordResetUseCase, + ResetPasswordRequest, + ResetPasswordUseCase, +) +from packages.application.auth.register_user_use_case import ( + RegisterUserRequest, + RegisterUserResponse, + RegisterUserUseCase, + VerifyEmailRequest, + VerifyEmailUseCase, +) + +# ── Test helpers ───────────────────────────────────────── + + +class FakeUser: + def __init__(self, **kwargs): + self.id = kwargs.get("id", "user-123") + self.email = kwargs.get("email", "test@example.com") + self.display_name = kwargs.get("display_name", "Test User") + self.username = kwargs.get("username", "testuser") + self.password_hash = kwargs.get("password_hash", "") + self.email_verified = kwargs.get("email_verified", False) + self.email_verification_token = kwargs.get("email_verification_token", None) + self.password_reset_token = kwargs.get("password_reset_token", None) + self.password_reset_expires_at = kwargs.get("password_reset_expires_at", None) + self.created_at = kwargs.get("created_at", datetime.now(timezone.utc)) + + +class FakeUserRepository: + def __init__(self, user=None): + self._user = user + self.saved_user = None + self.save_called = 0 + + def find_by_email(self, email): + if self._user and self._user.email == email: + return self._user + return None + + def find_by_username(self, username): + if self._user and self._user.username == username: + return self._user + return None + + def find_by_verification_token(self, token): + if self._user and self._user.email_verification_token == token: + return self._user + return None + + def find_by_password_reset_token(self, token): + if self._user and self._user.password_reset_token == token: + return self._user + return None + + def save(self, user): + self.saved_user = user + self.save_called += 1 + self._user = user + return user + + +class FakeEmailService: + def __init__(self, send_success=True, send_error=None): + self._send_success = send_success + self._send_error = send_error + self.sent_emails = [] + self.verification_emails = [] + self.password_reset_emails = [] + + def send_verification_email(self, to_email, username, verification_url): + self.verification_emails.append( + { + "to": to_email, + "username": username, + "url": verification_url, + } + ) + self.sent_emails.append(("verification", to_email)) + return self._send_success, self._send_error + + def send_password_reset_email(self, to_email, username, reset_url): + self.password_reset_emails.append( + { + "to": to_email, + "username": username, + "url": reset_url, + } + ) + self.sent_emails.append(("password_reset", to_email)) + return self._send_success, self._send_error + + +class FailingEmailService: + """Email service that raises an exception.""" + + def send_verification_email(self, **kwargs): + raise RuntimeError("SMTP connection failed") + + def send_password_reset_email(self, **kwargs): + raise RuntimeError("SMTP connection failed") + + +# ── RegisterUserUseCase tests ─────────────────────────── + + +class TestRegisterUserUseCase: + def _make_use_case(self, repo=None, email_service=None): + return RegisterUserUseCase( + user_repository=repo or FakeUserRepository(), + base_url="https://app.example.com", + email_service=email_service or FakeEmailService(), + ) + + def test_successful_registration(self): + repo = FakeUserRepository() + email_svc = FakeEmailService() + use_case = self._make_use_case(repo=repo, email_service=email_svc) + + req = RegisterUserRequest( + email="newuser@example.com", + password="StrongPass123!", + username="newuser", + display_name="New User", + ) + response, error = use_case.execute(req) + + assert error is None + assert response is not None + assert isinstance(response, RegisterUserResponse) + assert response.email == "newuser@example.com" + assert response.username == "newuser" + assert response.display_name == "New User" + assert response.user_id + assert response.email_verification_sent is True + + # user was saved + assert repo.saved_user is not None + assert repo.saved_user.email == "newuser@example.com" + assert repo.saved_user.email_verified is False + assert repo.saved_user.email_verification_token is not None + # password was hashed + assert repo.saved_user.password_hash != "StrongPass123!" + assert password_hasher.verify_password("StrongPass123!", repo.saved_user.password_hash) + + # email was sent + assert len(email_svc.verification_emails) == 1 + sent = email_svc.verification_emails[0] + assert sent["to"] == "newuser@example.com" + assert sent["username"] == "newuser" + assert "verify-email?token=" in sent["url"] + assert "https://app.example.com" in sent["url"] + + def test_email_stripped_and_lowercased(self): + repo = FakeUserRepository() + use_case = self._make_use_case(repo=repo) + + req = RegisterUserRequest( + email=" NEWUSER@EXAMPLE.COM ", + password="StrongPass123!", + username="newuser", + display_name="New User", + ) + response, error = use_case.execute(req) + + assert error is None + assert response is not None + assert response.email == "newuser@example.com" + assert repo.saved_user.email == "newuser@example.com" + + def test_username_stripped(self): + repo = FakeUserRepository() + use_case = self._make_use_case(repo=repo) + + req = RegisterUserRequest( + email="a@b.com", + password="StrongPass123!", + username=" myuser ", + display_name="Display", + ) + response, error = use_case.execute(req) + + assert error is None + assert response.username == "myuser" + + def test_display_name_stripped(self): + repo = FakeUserRepository() + use_case = self._make_use_case(repo=repo) + + req = RegisterUserRequest( + email="a@b.com", + password="StrongPass123!", + username="user", + display_name=" My Name ", + ) + response, error = use_case.execute(req) + + assert error is None + assert response.display_name == "My Name" + + def test_empty_email_returns_error(self): + use_case = self._make_use_case() + req = RegisterUserRequest(email="", password="StrongPass123!", username="u", display_name="D") + response, error = use_case.execute(req) + + assert response is None + assert "Email is required" in error + + def test_empty_username_returns_error(self): + use_case = self._make_use_case() + req = RegisterUserRequest(email="a@b.com", password="StrongPass123!", username=" ", display_name="D") + response, error = use_case.execute(req) + + assert response is None + assert "Username is required" in error + + def test_empty_display_name_returns_error(self): + use_case = self._make_use_case() + req = RegisterUserRequest(email="a@b.com", password="StrongPass123!", username="u", display_name=" ") + response, error = use_case.execute(req) + + assert response is None + assert "Display name is required" in error + + def test_weak_password_returns_error(self): + use_case = self._make_use_case() + req = RegisterUserRequest(email="a@b.com", password="weak", username="u", display_name="D") + response, error = use_case.execute(req) + + assert response is None + assert error is not None + # password validation error message + assert len(error) > 0 + + def test_email_already_registered(self): + existing = FakeUser(email="existing@example.com", username="existinguser") + repo = FakeUserRepository(user=existing) + use_case = self._make_use_case(repo=repo) + + req = RegisterUserRequest( + email="existing@example.com", + password="StrongPass123!", + username="newuser", + display_name="New User", + ) + response, error = use_case.execute(req) + + assert response is None + assert "Email already registered" in error + # no new user saved + assert repo.save_called == 0 + + def test_username_already_taken(self): + existing = FakeUser(email="other@example.com", username="taken") + repo = FakeUserRepository(user=existing) + use_case = self._make_use_case(repo=repo) + + req = RegisterUserRequest( + email="new@example.com", + password="StrongPass123!", + username="taken", + display_name="New User", + ) + response, error = use_case.execute(req) + + assert response is None + assert "Username already taken" in error + + def test_email_service_failure_user_still_created(self): + repo = FakeUserRepository() + email_svc = FakeEmailService(send_success=False, send_error="SMTP error") + use_case = self._make_use_case(repo=repo, email_service=email_svc) + + req = RegisterUserRequest( + email="a@b.com", + password="StrongPass123!", + username="user", + display_name="User", + ) + response, error = use_case.execute(req) + + # user still created + assert error is None + assert response is not None + assert response.email_verification_sent is False + assert repo.saved_user is not None + + def test_email_service_exception_user_still_created(self): + repo = FakeUserRepository() + use_case = RegisterUserUseCase( + user_repository=repo, + base_url="https://app.example.com", + email_service=FailingEmailService(), + ) + + req = RegisterUserRequest( + email="a@b.com", + password="StrongPass123!", + username="user", + display_name="User", + ) + response, error = use_case.execute(req) + + assert error is None + assert response is not None + assert response.email_verification_sent is False + assert repo.saved_user is not None + + def test_exception_handling(self): + repo = MagicMock() + repo.find_by_email.side_effect = RuntimeError("DB down") + use_case = self._make_use_case(repo=repo) + + req = RegisterUserRequest( + email="a@b.com", + password="StrongPass123!", + username="user", + display_name="User", + ) + response, error = use_case.execute(req) + + assert response is None + assert "Registration failed" in error + + def test_user_id_is_generated(self): + repo = FakeUserRepository() + use_case = self._make_use_case(repo=repo) + + req = RegisterUserRequest(email="a@b.com", password="StrongPass123!", username="u", display_name="D") + response, _ = use_case.execute(req) + + assert response.user_id + assert len(response.user_id) == 32 # uuid4 hex + + +# ── VerifyEmailUseCase tests ──────────────────────────── + + +class TestVerifyEmailUseCase: + def test_successful_verification(self): + user = FakeUser(email_verified=False, email_verification_token="test-token-123") + repo = FakeUserRepository(user=user) + use_case = VerifyEmailUseCase(user_repository=repo) + + req = VerifyEmailRequest(token="test-token-123") + success, error = use_case.execute(req) + + assert success is True + assert error is None + assert repo.saved_user.email_verified is True + assert repo.saved_user.email_verification_token is None + + def test_empty_token(self): + repo = FakeUserRepository() + use_case = VerifyEmailUseCase(user_repository=repo) + + req = VerifyEmailRequest(token="") + success, error = use_case.execute(req) + + assert success is False + assert "Verification token is required" in error + + def test_invalid_token(self): + repo = FakeUserRepository() + use_case = VerifyEmailUseCase(user_repository=repo) + + req = VerifyEmailRequest(token="nonexistent-token") + success, error = use_case.execute(req) + + assert success is False + assert "Invalid or expired verification token" in error + + def test_already_verified_returns_success(self): + user = FakeUser(email_verified=True, email_verification_token="some-token") + repo = FakeUserRepository(user=user) + use_case = VerifyEmailUseCase(user_repository=repo) + + req = VerifyEmailRequest(token="some-token") + success, error = use_case.execute(req) + + assert success is True + assert error is None + # idempotent - no save needed + # (depends on implementation - current returns early without save) + + def test_token_cleared_after_verification(self): + user = FakeUser(email_verified=False, email_verification_token="tok123") + repo = FakeUserRepository(user=user) + use_case = VerifyEmailUseCase(user_repository=repo) + + req = VerifyEmailRequest(token="tok123") + use_case.execute(req) + + assert repo.saved_user.email_verification_token is None + + def test_exception_handling(self): + repo = MagicMock() + repo.find_by_verification_token.side_effect = RuntimeError("DB down") + use_case = VerifyEmailUseCase(user_repository=repo) + + req = VerifyEmailRequest(token="token") + success, error = use_case.execute(req) + + assert success is False + assert "Email verification failed" in error + + +# ── RequestPasswordResetUseCase tests ─────────────────── + + +class TestRequestPasswordResetUseCase: + def _make_use_case(self, repo=None, email_service=None, expire_hours=1): + return RequestPasswordResetUseCase( + user_repository=repo or FakeUserRepository(), + base_url="https://app.example.com", + token_expire_hours=expire_hours, + email_service=email_service or FakeEmailService(), + ) + + def test_successful_request(self): + user = FakeUser(email="user@example.com", username="testuser") + repo = FakeUserRepository(user=user) + email_svc = FakeEmailService() + use_case = self._make_use_case(repo=repo, email_service=email_svc) + + req = RequestPasswordResetRequest(email="user@example.com") + success, error = use_case.execute(req) + + assert success is True + assert error is None + + # token set on user + assert repo.saved_user.password_reset_token is not None + assert len(repo.saved_user.password_reset_token) > 10 + assert repo.saved_user.password_reset_expires_at is not None + + # email sent + assert len(email_svc.password_reset_emails) == 1 + sent = email_svc.password_reset_emails[0] + assert sent["to"] == "user@example.com" + assert "reset-password?token=" in sent["url"] + + def test_empty_email(self): + use_case = self._make_use_case() + req = RequestPasswordResetRequest(email="") + success, error = use_case.execute(req) + + assert success is False + assert "Email is required" in error + + def test_nonexistent_user_returns_true_security(self): + repo = FakeUserRepository() # no users + use_case = self._make_use_case(repo=repo) + + req = RequestPasswordResetRequest(email="nobody@example.com") + success, error = use_case.execute(req) + + # Always returns true to prevent user enumeration + assert success is True + assert error is None + assert repo.save_called == 0 # no save + + def test_token_expiry_set_correctly(self): + user = FakeUser(email="u@e.com") + repo = FakeUserRepository(user=user) + use_case = self._make_use_case(repo=repo, expire_hours=2) + + before = datetime.now(timezone.utc) + req = RequestPasswordResetRequest(email="u@e.com") + use_case.execute(req) + after = datetime.now(timezone.utc) + + expires_at = repo.saved_user.password_reset_expires_at + assert expires_at is not None + # should be ~2 hours from now + min_expected = before + timedelta(hours=2) + max_expected = after + timedelta(hours=2) + assert min_expected <= expires_at <= max_expected + timedelta(seconds=1) + + def test_default_expire_hours(self): + user = FakeUser(email="u@e.com") + repo = FakeUserRepository(user=user) + use_case = RequestPasswordResetUseCase( + user_repository=repo, + base_url="https://app.example.com", + email_service=FakeEmailService(), + ) + + before = datetime.now(timezone.utc) + req = RequestPasswordResetRequest(email="u@e.com") + use_case.execute(req) + + expires_at = repo.saved_user.password_reset_expires_at + assert expires_at is not None + # default is 1 hour + assert timedelta(minutes=55) < (expires_at - before) < timedelta(hours=1, minutes=1) + + def test_email_service_failure_still_returns_true(self): + user = FakeUser(email="u@e.com") + repo = FakeUserRepository(user=user) + email_svc = FakeEmailService(send_success=False, send_error="SMTP down") + use_case = self._make_use_case(repo=repo, email_service=email_svc) + + req = RequestPasswordResetRequest(email="u@e.com") + success, error = use_case.execute(req) + + # still returns true for security + assert success is True + assert error is None + # token still set + assert repo.saved_user.password_reset_token is not None + + def test_email_service_exception_still_returns_true(self): + user = FakeUser(email="u@e.com") + repo = FakeUserRepository(user=user) + use_case = RequestPasswordResetUseCase( + user_repository=repo, + base_url="https://app.example.com", + email_service=FailingEmailService(), + ) + + req = RequestPasswordResetRequest(email="u@e.com") + success, error = use_case.execute(req) + + assert success is True + assert error is None + assert repo.saved_user.password_reset_token is not None + + def test_email_stripped_lowercased(self): + user = FakeUser(email="u@e.com") + repo = FakeUserRepository(user=user) + use_case = self._make_use_case(repo=repo) + + req = RequestPasswordResetRequest(email=" U@E.COM ") + # email would be normalized but find_by_email should still find it + # since our fake repo compares exact strings + # Let's just check the normalization happens + assert req.email == "u@e.com" + + def test_exception_handling(self): + repo = MagicMock() + repo.find_by_email.side_effect = RuntimeError("DB down") + use_case = self._make_use_case(repo=repo) + + req = RequestPasswordResetRequest(email="a@b.com") + success, error = use_case.execute(req) + + assert success is False + assert "Password reset request failed" in error + + +# ── ResetPasswordUseCase tests ────────────────────────── + + +class TestResetPasswordUseCase: + def test_successful_reset(self): + token = "reset-token-123" + old_hash = password_hasher.hash_password("oldpassword") + user = FakeUser( + password_hash=old_hash, + password_reset_token=token, + password_reset_expires_at=datetime.now(timezone.utc) + timedelta(hours=1), + ) + repo = FakeUserRepository(user=user) + use_case = ResetPasswordUseCase(user_repository=repo) + + req = ResetPasswordRequest(token=token, new_password="NewStrongPass456!") + success, error = use_case.execute(req) + + assert success is True + assert error is None + + # password updated + assert repo.saved_user.password_hash != old_hash + assert password_hasher.verify_password("NewStrongPass456!", repo.saved_user.password_hash) + + # token cleared + assert repo.saved_user.password_reset_token is None + assert repo.saved_user.password_reset_expires_at is None + + def test_empty_token(self): + use_case = ResetPasswordUseCase(user_repository=FakeUserRepository()) + req = ResetPasswordRequest(token="", new_password="StrongPass123!") + success, error = use_case.execute(req) + + assert success is False + assert "Reset token is required" in error + + def test_empty_new_password(self): + use_case = ResetPasswordUseCase(user_repository=FakeUserRepository()) + req = ResetPasswordRequest(token="token", new_password="") + success, error = use_case.execute(req) + + assert success is False + assert "New password is required" in error + + def test_weak_new_password(self): + use_case = ResetPasswordUseCase(user_repository=FakeUserRepository()) + req = ResetPasswordRequest(token="token", new_password="weak") + success, error = use_case.execute(req) + + assert success is False + assert error is not None + # some validation error + assert len(error) > 0 + + def test_invalid_token(self): + repo = FakeUserRepository() # no user with this token + use_case = ResetPasswordUseCase(user_repository=repo) + + req = ResetPasswordRequest(token="bad-token", new_password="StrongPass123!") + success, error = use_case.execute(req) + + assert success is False + assert "Invalid or expired reset token" in error + + def test_expired_token(self): + token = "expired-token" + user = FakeUser( + password_reset_token=token, + password_reset_expires_at=datetime.now(timezone.utc) - timedelta(hours=1), + ) + repo = FakeUserRepository(user=user) + use_case = ResetPasswordUseCase(user_repository=repo) + + req = ResetPasswordRequest(token=token, new_password="StrongPass123!") + success, error = use_case.execute(req) + + assert success is False + assert "expired" in error.lower() + + def test_naive_datetime_expiry_treated_as_utc(self): + token = "naive-token" + # naive datetime representing UTC time 1 hour in the past + naive_expired = datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(hours=1) + user = FakeUser( + password_reset_token=token, + password_reset_expires_at=naive_expired, + ) + repo = FakeUserRepository(user=user) + use_case = ResetPasswordUseCase(user_repository=repo) + + req = ResetPasswordRequest(token=token, new_password="StrongPass123!") + success, error = use_case.execute(req) + + assert success is False + assert "expired" in error.lower() + + def test_no_expiry_set_does_not_expire(self): + token = "no-expiry-token" + user = FakeUser( + password_reset_token=token, + password_reset_expires_at=None, + ) + repo = FakeUserRepository(user=user) + use_case = ResetPasswordUseCase(user_repository=repo) + + req = ResetPasswordRequest(token=token, new_password="StrongPass123!") + success, error = use_case.execute(req) + + # no expiry = should work + assert success is True + assert error is None + + def test_token_cleared_after_reset(self): + token = "clear-me" + user = FakeUser( + password_reset_token=token, + password_reset_expires_at=datetime.now(timezone.utc) + timedelta(hours=1), + ) + repo = FakeUserRepository(user=user) + use_case = ResetPasswordUseCase(user_repository=repo) + + req = ResetPasswordRequest(token=token, new_password="StrongPass123!") + use_case.execute(req) + + assert repo.saved_user.password_reset_token is None + assert repo.saved_user.password_reset_expires_at is None + + def test_exception_handling(self): + repo = MagicMock() + repo.find_by_password_reset_token.side_effect = RuntimeError("DB down") + use_case = ResetPasswordUseCase(user_repository=repo) + + req = ResetPasswordRequest(token="token", new_password="StrongPass123!") + success, error = use_case.execute(req) + + assert success is False + assert "Password reset failed" in error