Coverage for src/kwai/modules/identity/recover_user.py: 96%

1"""Module that implements the recover user use case.""" 

2 

3from dataclasses import dataclass 

4 

5from kwai.core.domain.exceptions import UnprocessableException 

6from kwai.core.domain.value_objects.email_address import EmailAddress 

7from kwai.core.domain.value_objects.timestamp import Timestamp 

8from kwai.core.events.publisher import Publisher 

9from kwai.modules.identity.user_recoveries.user_recovery import UserRecoveryEntity 

10from kwai.modules.identity.user_recoveries.user_recovery_events import ( 

11 UserRecoveryCreatedEvent, 

12) 

13from kwai.modules.identity.user_recoveries.user_recovery_repository import ( 

14 UserRecoveryRepository, 

15) 

16from kwai.modules.identity.users.user_account_repository import UserAccountRepository 

17 

18 

19@dataclass(frozen=True, kw_only=True) 

20class RecoverUserCommand: 

21 """Command for the recover user use case.""" 

22 

23 email: str 

24 

25 

26class RecoverUser: 

27 """Use case: recover user. 

28 

29 Attributes: 

30 _user_account_repo (UserAccountRepository): The repository for getting the 

31 user account. 

32 _user_recovery_repo (UserRecoveryRepository): The repository for creating a 

33 user recovery. 

34 _publisher (Bus): An event bus for dispatching the UserRecoveryCreatedEvent 

35 event. 

36 """ 

37 

38 def __init__( 

39 self, 

40 user_repo: UserAccountRepository, 

41 user_recovery_repo: UserRecoveryRepository, 

42 publisher: Publisher, 

43 ): 

44 self._user_account_repo = user_repo 

45 self._user_recovery_repo = user_recovery_repo 

46 self._publisher = publisher 

47 

48 async def execute(self, command: RecoverUserCommand) -> UserRecoveryEntity: 

49 """Execute the use case. 

50 

51 Args: 

52 command: The input for this use case. 

53 

54 Raises: 

55 UserAccountNotFoundException: Raised when the user with the given email 

56 address does not exist. 

57 UnprocessableException: Raised when the user is revoked 

58 """ 

59 user_account = await self._user_account_repo.get_user_by_email( 

60 EmailAddress(command.email) 

61 ) 

62 if user_account.revoked: 

63 raise UnprocessableException("User account is revoked") 

64 

65 user_recovery = await self._user_recovery_repo.create( 

66 UserRecoveryEntity( 

67 user=user_account.user, 

68 expiration=Timestamp.create_with_delta(hours=2), 

69 ) 

70 ) 

71 

72 await self._publisher.publish( 

73 UserRecoveryCreatedEvent(uuid=str(user_recovery.uuid)) 

74 ) 

75 

76 return user_recovery