Coverage for src/kwai/modules/identity/users/user_account.py: 97%

1"""Module that implements a user account 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.password import Password 

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

10from kwai.modules.identity.exceptions import NotAllowedException 

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

12 

13 

14class UserAccountIdentifier(IntIdentifier): 

15 """Identifier for a user account.""" 

16 

17 

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

19class UserAccountEntity(DataclassEntity): 

20 """A user account entity. 

21 

22 Attributes: 

23 user: The associated user entity. 

24 password: The password of the user. 

25 logged_in: Whether the user is logged in. 

26 last_login: Timestamp of the last login. 

27 last_unsuccessful_login: Timestamp of the last unsuccessful login. 

28 revoked: Whether the user is revoked. 

29 admin: Whether the user is an administrator. 

30 """ 

31 

32 ID: ClassVar[Type] = UserAccountIdentifier 

33 

34 user: UserEntity 

35 password: Password 

36 logged_in: bool = False 

37 last_login: Timestamp = field(default_factory=Timestamp) 

38 last_unsuccessful_login: Timestamp = field(default_factory=Timestamp) 

39 revoked: bool = False 

40 admin: bool = False 

41 

42 def login(self, password: str) -> Self: 

43 """Check if the given password is correct. 

44 

45 When login succeeds, last_login will be updated. 

46 When login fails, last_unsuccessful_login will be updated. 

47 

48 Args: 

49 password: The password. 

50 """ 

51 if self.password.verify(password): 

52 return replace(self, last_login=Timestamp.create_now(), logged_in=True) 

53 

54 return replace( 

55 self, last_unsuccessful_login=Timestamp.create_now(), logged_in=False 

56 ) 

57 

58 def reset_password(self, password: Password) -> Self: 

59 """Reset the password of the user account. 

60 

61 Args: 

62 password: The new password. 

63 """ 

64 if self.revoked: 

65 raise NotAllowedException() 

66 

67 return replace( 

68 self, 

69 password=password, 

70 traceable_time=self.traceable_time.mark_for_update(), 

71 ) 

72 

73 def revoke(self) -> Self: 

74 """Revoke a user account.""" 

75 return replace( 

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

77 )