Coverage for src/kwai/modules/identity/recreate_user_invitation.py: 95%

1"""Module for implementing the use case of recreating a user invitation.""" 

2 

3from dataclasses import dataclass 

4 

5from kwai.core.domain.exceptions import UnprocessableException 

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

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

8from kwai.core.events.publisher import Publisher 

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

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

11 UserInvitationCreatedEvent, 

12) 

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

14 UserInvitationRepository, 

15) 

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

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

18 UserNotFoundException, 

19 UserRepository, 

20) 

21 

22 

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

24class RecreateUserInvitationCommand: 

25 """Input for the use case 'Recreate User Invitation'. 

26 

27 [RecreateUserInvitation][kwai.modules.identity.invite_user.RecreateUserInvitation] 

28 

29 Attributes: 

30 uuid: The unique id of the original invitation. 

31 expiration_in_days: When does the invitation expires (in days) 

32 remark: A remark for this user invitation 

33 """ 

34 

35 uuid: str 

36 expiration_in_days: int = 7 

37 remark: str = "" 

38 

39 

40class RecreateUserInvitation: 

41 """Use case for recreating a user invitation.""" 

42 

43 def __init__( 

44 self, 

45 user: UserEntity, 

46 user_repo: UserRepository, 

47 user_invitation_repo: UserInvitationRepository, 

48 publisher: Publisher, 

49 ): 

50 """Initialize the use case.""" 

51 self._user = user 

52 self._user_repo = user_repo 

53 self._user_invitation_repo = user_invitation_repo 

54 self._publisher = publisher 

55 

56 async def execute( 

57 self, command: RecreateUserInvitationCommand 

58 ) -> UserInvitationEntity: 

59 """Execute the use case. 

60 

61 Args: 

62 command: The input for this use case. 

63 

64 Returns: 

65 A user invitation. 

66 

67 Raises: 

68 UnprocessableException: raised when the email address is already in use. 

69 """ 

70 uuid = UniqueId.create_from_string(command.uuid) 

71 original_invitation = await self._user_invitation_repo.get_invitation_by_uuid( 

72 uuid 

73 ) 

74 original_invitation = original_invitation.revoke() 

75 await self._user_invitation_repo.update(original_invitation) 

76 

77 try: 

78 await self._user_repo.get_user_by_email(original_invitation.email) 

79 raise UnprocessableException( 

80 f"{original_invitation.email} is already in use" 

81 ) 

82 except UserNotFoundException: 

83 pass 

84 

85 query = ( 

86 self._user_invitation_repo.create_query() 

87 .filter_by_email(original_invitation.email) 

88 .filter_active() 

89 .filter_not_expired(Timestamp.create_now()) 

90 ) 

91 if await query.count() > 0: 

92 raise UnprocessableException( 

93 f"There are still pending invitations for {original_invitation.email}" 

94 ) 

95 

96 invitation = await self._user_invitation_repo.create( 

97 UserInvitationEntity( 

98 email=original_invitation.email, 

99 name=original_invitation.name, 

100 uuid=UniqueId.generate(), 

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

102 remark=command.remark, 

103 user=self._user, 

104 ) 

105 ) 

106 

107 await self._publisher.publish( 

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

109 ) 

110 

111 return invitation