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
« prev ^ index » next coverage.py v7.6.10, created at 2024-01-01 00:00 +0000
1"""Module that implements a user recovery entity."""
3from dataclasses import dataclass, field, replace
4from typing import ClassVar, Self, Type
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
13class UserRecoveryIdentifier(IntIdentifier):
14 """Identifier for a user recovery entity."""
17@dataclass(kw_only=True, eq=False, slots=True, frozen=True)
18class UserRecoveryEntity(DataclassEntity):
19 """A user recovery entity."""
21 ID: ClassVar[Type] = UserRecoveryIdentifier
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)
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 )
38 @property
39 def confirmed(self) -> bool:
40 """Return True when this user recovery was confirmed."""
41 return not self.confirmation.empty
43 @property
44 def is_expired(self) -> bool:
45 """Return True when the user recovery is expired."""
46 return self.expiration.is_past
48 @property
49 def mailed(self) -> bool:
50 """Return True if the email has already been sent."""
51 return not self.mailed_at.empty
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 )