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 | None = None) -> 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 Use None for password, when the user logs in using OIDC. 

48 

49 Args: 

50 password: The password. 

51 """ 

52 if password and not self.password.verify(password): 

53 return replace( 

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

55 ) 

56 

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

58 

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

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

61 

62 Args: 

63 password: The new password. 

64 """ 

65 if self.revoked: 

66 raise NotAllowedException() 

67 

68 return replace( 

69 self, 

70 password=password, 

71 traceable_time=self.traceable_time.mark_for_update(), 

72 ) 

73 

74 def revoke(self) -> Self: 

75 """Revoke a user account.""" 

76 return replace( 

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

78 ) 

79 

80 def enact(self) -> Self: 

81 """Reactivate a user account.""" 

82 return replace( 

83 self, revoked=False, traceable_time=self.traceable_time.mark_for_update() 

84 )