Coverage for kwai/modules/identity/logout.py: 100%

1"""Module that implements the logout use case.""" 

2from dataclasses import dataclass 

3 

4from kwai.modules.identity.tokens.access_token_repository import AccessTokenRepository 

5from kwai.modules.identity.tokens.refresh_token_repository import RefreshTokenRepository 

6from kwai.modules.identity.tokens.token_identifier import TokenIdentifier 

7 

8 

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

10class LogoutCommand: 

11 """Command for the logout use case. 

12 

13 Attributes: 

14 identifier(str): The refresh token to revoke 

15 """ 

16 

17 identifier: str 

18 

19 

20class Logout: 

21 """Use case: logout a user. 

22 

23 A user is logged out by revoking the refresh token. The access token that is 

24 related to this refresh token will also be revoked. 

25 

26 Attributes: 

27 _refresh_token_repository (RefreshTokenRepository): The repository to 

28 get and update the refresh token. 

29 _access_token_repository (AccessTokenRepository): The repository to 

30 get and update the access token. 

31 """ 

32 

33 def __init__( 

34 self, 

35 refresh_token_repository: RefreshTokenRepository, 

36 access_token_repository: AccessTokenRepository, 

37 ): 

38 self._refresh_token_repository = refresh_token_repository 

39 self._access_token_repository = access_token_repository 

40 

41 async def execute(self, command: LogoutCommand): 

42 """Execute the use case. 

43 

44 Args: 

45 command: The input for this use case. 

46 

47 Raises: 

48 RefreshTokenNotFoundException: The refresh token with the identifier 

49 could not be found. 

50 """ 

51 refresh_token = await self._refresh_token_repository.get_by_token_identifier( 

52 TokenIdentifier(hex_string=command.identifier) 

53 ) 

54 refresh_token.revoke() 

55 

56 await self._refresh_token_repository.update(refresh_token) 

57 await self._access_token_repository.update(refresh_token.access_token)