Coverage for src/kwai/modules/identity/user_recoveries/user_recovery.py: 100%

30 statements  

« prev     ^ index     » next       coverage.py v7.6.10, created at 2024-01-01 00:00 +0000

1"""Module that implements a user recovery 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.core.domain.value_objects.unique_id import UniqueId 

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

11 

12 

13class UserRecoveryIdentifier(IntIdentifier): 

14 """Identifier for a user recovery entity.""" 

15 

16 

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

18class UserRecoveryEntity(DataclassEntity): 

19 """A user recovery entity.""" 

20 

21 ID: ClassVar[Type] = UserRecoveryIdentifier 

22 

23 user: UserEntity 

24 expiration: Timestamp = field(default_factory=Timestamp) 

25 uuid: UniqueId = field(default_factory=UniqueId.generate) 

26 remark: str = "" 

27 confirmation: Timestamp = field(default_factory=Timestamp) 

28 mailed_at: Timestamp = field(default_factory=Timestamp) 

29 

30 def confirm(self) -> Self: 

31 """Confirm the user recovery.""" 

32 return replace( 

33 self, 

34 confirmation=Timestamp.create_now(), 

35 traceable_time=self.traceable_time.mark_for_update(), 

36 ) 

37 

38 @property 

39 def confirmed(self) -> bool: 

40 """Return True when this user recovery was confirmed.""" 

41 return not self.confirmation.empty 

42 

43 @property 

44 def is_expired(self) -> bool: 

45 """Return True when the user recovery is expired.""" 

46 return self.expiration.is_past 

47 

48 @property 

49 def mailed(self) -> bool: 

50 """Return True if the email has already been sent.""" 

51 return not self.mailed_at.empty 

52 

53 def mail_sent(self): 

54 """Set the timestamp when mail has been sent.""" 

55 return replace( 

56 self, 

57 mailed_at=Timestamp.create_now(), 

58 traceable_time=self.traceable_time.mark_for_update(), 

59 )