Coverage for bc/kwai-bc-identity/src/kwai_bc_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 

10 

11from kwai_bc_identity.exceptions import NotAllowedException 

12from kwai_bc_identity.users.user import UserEntity 

13 

14 

15class UserAccountIdentifier(IntIdentifier): 

16 """Identifier for a user account.""" 

17 

18 

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

20class UserAccountEntity(DataclassEntity): 

21 """A user account entity. 

22 

23 Attributes: 

24 user: The associated user entity. 

25 password: The password of the user. 

26 logged_in: Whether the user is logged in. 

27 last_login: Timestamp of the last login. 

28 last_unsuccessful_login: Timestamp of the last unsuccessful login. 

29 revoked: Whether the user is revoked. 

30 admin: Whether the user is an administrator. 

31 """ 

32 

33 ID: ClassVar[Type] = UserAccountIdentifier 

34 

35 user: UserEntity 

36 password: Password 

37 logged_in: bool = False 

38 last_login: Timestamp = field(default_factory=Timestamp) 

39 last_unsuccessful_login: Timestamp = field(default_factory=Timestamp) 

40 revoked: bool = False 

41 admin: bool = False 

42 

43 def login(self, password: str | None = None) -> Self: 

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

45 

46 When login succeeds, last_login will be updated. 

47 When login fails, last_unsuccessful_login will be updated. 

48 Use None for password, when the user logs in using OIDC. 

49 

50 Args: 

51 password: The password. 

52 """ 

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

54 return replace( 

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

56 ) 

57 

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

59 

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

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

62 

63 Args: 

64 password: The new password. 

65 """ 

66 if self.revoked: 

67 raise NotAllowedException() 

68 

69 return replace( 

70 self, 

71 password=password, 

72 traceable_time=self.traceable_time.mark_for_update(), 

73 ) 

74 

75 def revoke(self) -> Self: 

76 """Revoke a user account.""" 

77 return replace( 

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

79 ) 

80 

81 def enact(self) -> Self: 

82 """Reactivate a user account.""" 

83 return replace( 

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

85 )