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

1"""Module for a refresh token entity.""" 

2 

3from dataclasses import dataclass, field, replace 

4from typing import ClassVar, Self, Type 

5 

6from kwai.core.domain.entity import DataclassEntity 

7from kwai.core.domain.value_objects.identifier import IntIdentifier 

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

9from kwai.modules.identity.tokens.access_token import AccessTokenEntity 

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

11 

12 

13class RefreshTokenIdentifier(IntIdentifier): 

14 """Identifier for a refresh token.""" 

15 

16 

17@dataclass(kw_only=True, eq=False, slots=True, frozen=True) 

18class RefreshTokenEntity(DataclassEntity): 

19 """A refresh token entity. 

20 

21 Attributes: 

22 access_token: The access token associated with this refresh token. 

23 identifier: The actual token. 

24 expiration: The expiration timestamp of the token. 

25 revoked: Whether the token has been revoked. 

26 """ 

27 

28 ID: ClassVar[Type] = RefreshTokenIdentifier 

29 

30 access_token: AccessTokenEntity 

31 identifier: TokenIdentifier = field(default_factory=TokenIdentifier.generate) 

32 expiration: Timestamp = field(default_factory=Timestamp.create_now) 

33 revoked: bool = field(default=False) 

34 

35 @property 

36 def expired(self) -> bool: 

37 """Return True when the token is expired.""" 

38 return self.expiration.is_past 

39 

40 def revoke(self) -> Self: 

41 """Revoke the refresh token. 

42 

43 Returns: 

44 A revoked refresh token. 

45 """ 

46 return replace( 

47 self, revoked=True, traceable_time=self.traceable_time.mark_for_update() 

48 ) 

49 

50 def renew(self, expiry_minutes: int, access_token_expiry_minutes: int) -> Self: 

51 """Renew the refresh token. 

52 

53 The access token will also be renewed. 

54 

55 Args: 

56 expiry_minutes: The number of minutes until the refresh token expires. 

57 access_token_expiry_minutes: 

58 The number of minutes until the access token expires. 

59 """ 

60 return replace( 

61 self, 

62 identifier=TokenIdentifier.generate(), 

63 expiration=Timestamp.create_with_delta(minutes=expiry_minutes), 

64 access_token=self.access_token.renew(access_token_expiry_minutes), 

65 traceable_time=self.traceable_time.mark_for_update(), 

66 )