Coverage for src/kwai/modules/identity/invite_user.py: 100%

1"""Module for the invite 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.name import Name 

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

9from kwai.core.domain.value_objects.unique_id import UniqueId 

10from kwai.core.events.publisher import Publisher 

11from kwai.modules.identity.user_invitations.user_invitation import UserInvitationEntity 

12from kwai.modules.identity.user_invitations.user_invitation_events import ( 

13 UserInvitationCreatedEvent, 

14) 

15from kwai.modules.identity.user_invitations.user_invitation_repository import ( 

16 UserInvitationRepository, 

17) 

18from kwai.modules.identity.users.user import UserEntity 

19from kwai.modules.identity.users.user_repository import ( 

20 UserNotFoundException, 

21 UserRepository, 

22) 

23 

24 

25@dataclass(kw_only=True, frozen=True, slots=True) 

26class InviteUserCommand: 

27 """Input for the use case. 

28 

29 [InviteUser][kwai.modules.identity.invite_user.InviteUser] 

30 

31 Attributes: 

32 first_name: Firstname of the invited 

33 last_name: Lastname of the invited 

34 email: Email address of the invited 

35 expiration_in_days: After how many days will the invitation expire? 

36 remark: A remark about this invitation 

37 """ 

38 

39 first_name: str 

40 last_name: str 

41 email: str 

42 expiration_in_days: int = 7 

43 remark: str = "" 

44 

45 

46class InviteUser: 

47 """Invite user use case. 

48 

49 This use case will try to create an invitation for a user. 

50 """ 

51 

52 def __init__( 

53 self, 

54 user: UserEntity, 

55 user_repo: UserRepository, 

56 user_invitation_repo: UserInvitationRepository, 

57 publisher: Publisher, 

58 ): 

59 """Initialize the use case. 

60 

61 Args: 

62 user: The inviter. 

63 user_repo: The repository to check if the email address is still free. 

64 user_invitation_repo: The repository for creating an invitation. 

65 publisher: A publisher for publishing the user invitation created event. 

66 """ 

67 self._user = user 

68 self._user_repo = user_repo 

69 self._user_invitation_repo = user_invitation_repo 

70 self._publisher = publisher 

71 

72 async def execute(self, command: InviteUserCommand) -> UserInvitationEntity: 

73 """Execute the use case. 

74 

75 Args: 

76 command: The input for this use case. 

77 

78 Returns: 

79 A user invitation 

80 

81 Raises: 

82 UnprocessableException: raised when the email address is already in use or 

83 when there are still pending invitations for the email address. 

84 """ 

85 email_address = EmailAddress(command.email) 

86 try: 

87 await self._user_repo.get_user_by_email(email_address) 

88 raise UnprocessableException(f"{command.email} is already used.") 

89 except UserNotFoundException: 

90 pass 

91 

92 query = ( 

93 self._user_invitation_repo.create_query() 

94 .filter_by_email(email_address) 

95 .filter_active(Timestamp.create_now()) 

96 ) 

97 if await query.count() > 0: 

98 raise UnprocessableException( 

99 f"There are still pending invitations for {command.email}" 

100 ) 

101 

102 invitation = await self._user_invitation_repo.create( 

103 UserInvitationEntity( 

104 email=email_address, 

105 name=Name(first_name=command.first_name, last_name=command.last_name), 

106 uuid=UniqueId.generate(), 

107 expired_at=Timestamp.create_with_delta(days=command.expiration_in_days), 

108 user=self._user, 

109 ) 

110 ) 

111 

112 await self._publisher.publish( 

113 UserInvitationCreatedEvent(uuid=str(invitation.uuid)) 

114 ) 

115 

116 return invitation