Coverage for bc/kwai-bc-identity/src/kwai_bc_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 

9 

10from kwai_bc_identity.user_recoveries.user_recovery import UserRecoveryEntity 

11from kwai_bc_identity.user_recoveries.user_recovery_events import ( 

12 UserRecoveryCreatedEvent, 

13) 

14from kwai_bc_identity.user_recoveries.user_recovery_repository import ( 

15 UserRecoveryRepository, 

16) 

17from kwai_bc_identity.users.user_account_repository import UserAccountRepository 

18 

19 

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

21class RecoverUserCommand: 

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

23 

24 email: str 

25 

26 

27class RecoverUser: 

28 """Use case: recover user. 

29 

30 Attributes: 

31 _user_account_repo (UserAccountRepository): The repository for getting the 

32 user account. 

33 _user_recovery_repo (UserRecoveryRepository): The repository for creating a 

34 user recovery. 

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

36 event. 

37 """ 

38 

39 def __init__( 

40 self, 

41 user_repo: UserAccountRepository, 

42 user_recovery_repo: UserRecoveryRepository, 

43 publisher: Publisher, 

44 ): 

45 self._user_account_repo = user_repo 

46 self._user_recovery_repo = user_recovery_repo 

47 self._publisher = publisher 

48 

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

50 """Execute the use case. 

51 

52 Args: 

53 command: The input for this use case. 

54 

55 Raises: 

56 UserAccountNotFoundException: Raised when the user with the given email 

57 address does not exist. 

58 UnprocessableException: Raised when the user is revoked 

59 """ 

60 user_account = await self._user_account_repo.get_user_by_email( 

61 EmailAddress(command.email) 

62 ) 

63 if user_account.revoked: 

64 raise UnprocessableException("User account is revoked") 

65 

66 user_recovery = await self._user_recovery_repo.create( 

67 UserRecoveryEntity( 

68 user=user_account.user, 

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

70 ) 

71 ) 

72 

73 await self._publisher.publish( 

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

75 ) 

76 

77 return user_recovery