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

7from kwai_core.domain.value_objects.timestamp import Timestamp 

8from kwai_core.domain.value_objects.unique_id import UniqueId 

9from kwai_core.events.publisher import Publisher 

10 

11from kwai_bc_identity.user_invitations.user_invitation import UserInvitationEntity 

12from kwai_bc_identity.user_invitations.user_invitation_events import ( 

13 UserInvitationCreatedEvent, 

14) 

15from kwai_bc_identity.user_invitations.user_invitation_repository import ( 

16 UserInvitationRepository, 

17) 

18from kwai_bc_identity.users.user import UserEntity 

19from kwai_bc_identity.users.user_repository import ( 

20 UserNotFoundException, 

21 UserRepository, 

22) 

23 

24 

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

26class RecreateUserInvitationCommand: 

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

28 

29 [RecreateUserInvitation][kwai_bc_identity.invite_user.RecreateUserInvitation] 

30 

31 Attributes: 

32 uuid: The unique id of the original invitation. 

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

34 remark: A remark for this user invitation 

35 """ 

36 

37 uuid: str 

38 expiration_in_days: int = 7 

39 remark: str = "" 

40 

41 

42class RecreateUserInvitation: 

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

44 

45 def __init__( 

46 self, 

47 user: UserEntity, 

48 user_repo: UserRepository, 

49 user_invitation_repo: UserInvitationRepository, 

50 presenter: Presenter[UserInvitationEntity], 

51 publisher: Publisher, 

52 ): 

53 """Initialize the use case.""" 

54 self._user = user 

55 self._user_repo = user_repo 

56 self._user_invitation_repo = user_invitation_repo 

57 self._presenter = presenter 

58 self._publisher = publisher 

59 

60 async def execute(self, command: RecreateUserInvitationCommand): 

61 """Execute the use case. 

62 

63 Args: 

64 command: The input for this use case. 

65 

66 Raises: 

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

68 """ 

69 uuid = UniqueId.create_from_string(command.uuid) 

70 original_invitation = await self._user_invitation_repo.get_invitation_by_uuid( 

71 uuid 

72 ) 

73 original_invitation = original_invitation.revoke() 

74 await self._user_invitation_repo.update(original_invitation) 

75 

76 try: 

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

78 raise UnprocessableException( 

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

80 ) 

81 except UserNotFoundException: 

82 pass 

83 

84 query = ( 

85 self._user_invitation_repo.create_query() 

86 .filter_by_email(original_invitation.email) 

87 .filter_active() 

88 .filter_not_expired(Timestamp.create_now()) 

89 ) 

90 if await query.count() > 0: 

91 raise UnprocessableException( 

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

93 ) 

94 

95 invitation = await self._user_invitation_repo.create( 

96 UserInvitationEntity( 

97 email=original_invitation.email, 

98 name=original_invitation.name, 

99 uuid=UniqueId.generate(), 

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

101 remark=command.remark, 

102 user=self._user, 

103 ) 

104 ) 

105 

106 await self._publisher.publish( 

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

108 ) 

109 

110 self._presenter.present(invitation)