Coverage for bc/kwai-bc-identity/src/kwai_bc_identity/revoke_user.py: 100%

1"""Module that defines the use case for revoking a user.""" 

2 

3from dataclasses import dataclass 

4 

5from kwai_core.domain.presenter import Presenter 

6from kwai_core.domain.value_objects.unique_id import UniqueId 

7 

8from kwai_bc_identity.tokens.user_token_repository import UserTokenRepository 

9from kwai_bc_identity.users.user_account import UserAccountEntity 

10from kwai_bc_identity.users.user_account_repository import UserAccountRepository 

11 

12 

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

14class RevokeUserCommand: 

15 """Input for the RevokeUser use case. 

16 

17 Attributes: 

18 uuid: The UUID of the user to revoke. 

19 """ 

20 

21 uuid: str 

22 

23 

24class RevokeUser: 

25 """Use case for revoking a user.""" 

26 

27 def __init__( 

28 self, 

29 repo: UserAccountRepository, 

30 user_token_repo: UserTokenRepository, 

31 presenter: Presenter[UserAccountEntity], 

32 ): 

33 """Initialize the use case. 

34 

35 Args: 

36 repo: The user account repository to use. 

37 user_token_repo: The user token repository to use. 

38 presenter: The presenter that will be used to return the user. 

39 """ 

40 self._repo = repo 

41 self._user_token_repo = user_token_repo 

42 self._presenter = presenter 

43 

44 async def execute(self, command: RevokeUserCommand) -> None: 

45 """Execute the use case. 

46 

47 Raises: 

48 UserAccountNotFoundException: If the user does not exist. 

49 """ 

50 user_account = await self._repo.get_user_by_uuid( 

51 UniqueId.create_from_string(command.uuid) 

52 ) 

53 user_account = user_account.revoke() 

54 await self._repo.update(user_account) 

55 await self._user_token_repo.revoke(user_account) 

56 self._presenter.present(user_account)