"""Tests for AuthService: register, login, token round-trip.""" from datetime import datetime from uuid import uuid4 import pytest from app.entities.user import User from app.repositories_impl.sqlite.personal_data_repository import SQLitePersonalDataRepository from app.repositories_impl.sqlite.user_repository import SQLiteUserRepository from app.repositories_impl.sqlite.workspace_repository import SQLiteWorkspaceRepository from app.services.auth_service import AuthService from app.services.workspace_service import WorkspaceService from app.utils.errors import AuthError, ConflictError, InvitationError, UnauthorizedError class TestAuthServiceRegisterLogin: async def test_register_creates_user_and_personal_data(self, auth_service: AuthService) -> None: user, pd = await auth_service.register("alice@example.com", "securepassword", "Alice") assert user.email == "alice@example.com" assert pd.name == "Alice" assert user.personal_data_uuid == pd.uuid async def test_register_duplicate_email_raises_conflict( self, auth_service: AuthService ) -> None: await auth_service.register("bob@example.com", "password123", "Bob") with pytest.raises(ConflictError): await auth_service.register("bob@example.com", "otherpass", "Bob2") async def test_login_success(self, auth_service: AuthService) -> None: await auth_service.register("carol@example.com", "mypassword", "Carol") user = await auth_service.login("carol@example.com", "mypassword") assert user.email == "carol@example.com" async def test_login_wrong_password_raises(self, auth_service: AuthService) -> None: await auth_service.register("dave@example.com", "correctpass", "Dave") with pytest.raises(AuthError): await auth_service.login("dave@example.com", "wrongpass") async def test_login_unknown_email_raises(self, auth_service: AuthService) -> None: with pytest.raises(AuthError): await auth_service.login("nobody@example.com", "anything") async def test_register_conflict_message_is_generic(self, auth_service: AuthService) -> None: await auth_service.register("henry@example.com", "password123", "Henry") with pytest.raises(ConflictError) as exc_info: await auth_service.register("henry@example.com", "otherpass", "Henry2") assert "henry@example.com" not in str(exc_info.value) assert str(exc_info.value) == "Email already registered" async def test_login_wrong_password_and_unknown_email_share_message( self, auth_service: AuthService ) -> None: await auth_service.register("ivy@example.com", "correctpass", "Ivy") with pytest.raises(AuthError) as wrong_password: await auth_service.login("ivy@example.com", "wrongpass") with pytest.raises(AuthError) as unknown_email: await auth_service.login("nobody2@example.com", "anything") assert str(wrong_password.value) == str(unknown_email.value) async def test_login_inactive_account_raises_unauthorized( self, auth_service: AuthService, user_repo: SQLiteUserRepository, ) -> None: user, _ = await auth_service.register("jack@example.com", "password123", "Jack") user.is_active = False await user_repo.update(user.id, user) with pytest.raises(UnauthorizedError): await auth_service.login("jack@example.com", "password123") class TestAuthServiceTokens: async def test_access_token_round_trip(self, auth_service: AuthService) -> None: user, _ = await auth_service.register("eve@example.com", "pass1234", "Eve") token = auth_service.create_access_token(user.id) decoded_user = await auth_service.get_user_from_token(token) assert decoded_user.id == user.id async def test_refresh_token_issues_new_access_token(self, auth_service: AuthService) -> None: user, _ = await auth_service.register("frank@example.com", "pass5678", "Frank") refresh = auth_service.create_refresh_token(user.id) access = await auth_service.refresh_access_token(refresh) decoded = await auth_service.get_user_from_token(access) assert decoded.id == user.id async def test_using_refresh_as_access_raises_unauthorized( self, auth_service: AuthService ) -> None: user, _ = await auth_service.register("grace@example.com", "pass9012", "Grace") refresh = auth_service.create_refresh_token(user.id) with pytest.raises(UnauthorizedError): await auth_service.get_user_from_token(refresh) async def test_invalid_token_raises_unauthorized(self, auth_service: AuthService) -> None: with pytest.raises(UnauthorizedError): await auth_service.get_user_from_token("not.a.valid.jwt") async def test_token_for_inactive_user_raises_unauthorized( self, auth_service: AuthService, user_repo: SQLiteUserRepository, ) -> None: user, _ = await auth_service.register("heidi@example.com", "pass3456", "Heidi") token = auth_service.create_access_token(user.id) user.is_active = False await user_repo.update(user.id, user) with pytest.raises(UnauthorizedError): await auth_service.get_user_from_token(token) class TestAuthServiceRegisterWithInviteToken: """`register(..., invite_token=...)` lets a partner join straight from an invite link instead of first creating a throwaway solo workspace. Mirrors `WorkspaceService.redeem_invitation`'s claim logic -- see `test_workspace_service.py::TestInvitationScoping` for the service-level coverage of that logic; these tests cover the registration entry point. """ async def test_claims_existing_placeholder( self, auth_service: AuthService, workspace_service: WorkspaceService, user_repo: SQLiteUserRepository, ) -> None: ws = await workspace_service.create_workspace("Shared plan") owner_pd = await workspace_service.add_placeholder_parent(ws.id, "Owner") owner_user = await user_repo.create( User( id=uuid4(), personal_data_uuid=owner_pd.uuid, email="owner@example.com", hashed_password="not-a-real-hash", created_at=datetime.now(), ) ) placeholder = await workspace_service.add_placeholder_parent(ws.id, "Partner") token = await workspace_service.invite_partner(ws.id, owner_user.id) user, pd = await auth_service.register( "partner@example.com", "password123", "Partner Real Name", invite_token=token ) assert pd.uuid == placeholder.uuid assert pd.name == "Partner Real Name" assert user.personal_data_uuid == placeholder.uuid members = await workspace_service.list_members(ws.id) assert {m.name for m in members} == {"Owner", "Partner Real Name"} async def test_joins_free_slot_when_no_placeholder_exists( self, auth_service: AuthService, workspace_service: WorkspaceService, user_repo: SQLiteUserRepository, ) -> None: ws = await workspace_service.create_workspace("Shared plan") owner_pd = await workspace_service.add_placeholder_parent(ws.id, "Owner") owner_user = await user_repo.create( User( id=uuid4(), personal_data_uuid=owner_pd.uuid, email="owner2@example.com", hashed_password="not-a-real-hash", created_at=datetime.now(), ) ) token = await workspace_service.invite_partner(ws.id, owner_user.id) user, pd = await auth_service.register( "joiner@example.com", "password123", "Joiner", invite_token=token ) assert pd.workspace_id == ws.id assert user.personal_data_uuid == pd.uuid members = await workspace_service.list_members(ws.id) assert {m.name for m in members} == {"Owner", "Joiner"} async def test_invalid_token_raises_invitation_error(self, auth_service: AuthService) -> None: with pytest.raises(InvitationError, match="not found"): await auth_service.register( "nope@example.com", "password123", "Nope", invite_token="bogus-token" ) async def test_refuses_when_both_slots_have_real_logins( self, auth_service: AuthService, workspace_service: WorkspaceService, user_repo: SQLiteUserRepository, ) -> None: ws = await workspace_service.create_workspace("Shared plan") owner_pd = await workspace_service.add_placeholder_parent(ws.id, "Owner") owner_user = await user_repo.create( User( id=uuid4(), personal_data_uuid=owner_pd.uuid, email="owner3@example.com", hashed_password="not-a-real-hash", created_at=datetime.now(), ) ) second_pd = await workspace_service.add_placeholder_parent(ws.id, "Partner") await user_repo.create( User( id=uuid4(), personal_data_uuid=second_pd.uuid, email="partner3@example.com", hashed_password="not-a-real-hash", created_at=datetime.now(), ) ) token = await workspace_service.invite_partner(ws.id, owner_user.id) with pytest.raises(InvitationError, match="maximum number of members"): await auth_service.register( "late@example.com", "password123", "Late", invite_token=token ) async def test_raises_when_invitations_not_configured( self, user_repo: SQLiteUserRepository, personal_data_repo: SQLitePersonalDataRepository, workspace_repo: SQLiteWorkspaceRepository, ) -> None: """A direct-construction call site (e.g. a script) that never wires `inv_repo` gets a clear error rather than a crash if it's ever handed an invite token.""" bare_auth_service = AuthService(user_repo, personal_data_repo, workspace_repo) with pytest.raises(InvitationError, match="not available"): await bare_auth_service.register( "x@example.com", "password123", "X", invite_token="whatever" )