Coverage for src / lexigram / admin / domain / aggregate.py: 0%

1"""AdminUserAggregate — aggregate root for admin user management.""" 

2 

3from __future__ import annotations 

4 

5from dataclasses import dataclass, field 

6from datetime import UTC, datetime 

7from typing import Any 

8 

9from lexigram.admin.events import ( 

10 UserCreated, 

11 UserDeactivated, 

12 UserDeleted, 

13 UserUpdated, 

14) 

15from lexigram.domain.models.aggregate import AggregateRoot 

16 

17 

18@dataclass 

19class AdminUserAggregate(AggregateRoot): 

20 """Aggregate root representing an admin-managed user. 

21 

22 Encapsulates all state transitions for a user managed through the 

23 admin panel, recording domain events on each mutation. 

24 """ 

25 

26 username: str = "" 

27 email: str = "" 

28 hashed_password: str = "" 

29 roles: list[str] = field(default_factory=list) 

30 permissions: list[str] = field(default_factory=list) 

31 is_active: bool = True 

32 deleted_at: datetime | None = None 

33 created_at: datetime = field(default_factory=lambda: datetime.now(UTC)) 

34 

35 # ------------------------------------------------------------------ 

36 # Factory 

37 # ------------------------------------------------------------------ 

38 

39 @classmethod 

40 def create( 

41 cls, 

42 user_id: str, 

43 username: str, 

44 email: str, 

45 hashed_password: str, 

46 roles: list[str] | None = None, 

47 permissions: list[str] | None = None, 

48 actor_id: str = "", 

49 ) -> AdminUserAggregate: 

50 """Create a new AdminUserAggregate and record a UserCreated event. 

51 

52 Args: 

53 user_id: Unique identifier for the user. 

54 username: Login username. 

55 email: Email address. 

56 hashed_password: Pre-hashed password string. 

57 roles: Initial role assignments. 

58 permissions: Initial permission grants. 

59 actor_id: Identity of the principal performing the action. 

60 

61 Returns: 

62 A fully initialised aggregate with one pending domain event. 

63 """ 

64 aggregate = cls( 

65 id=user_id, 

66 username=username, 

67 email=email, 

68 hashed_password=hashed_password, 

69 roles=roles or [], 

70 permissions=permissions or [], 

71 ) 

72 aggregate._record_event( 

73 UserCreated( 

74 user_id=user_id, 

75 email=email, 

76 actor_id=actor_id or None, 

77 ) 

78 ) 

79 return aggregate 

80 

81 # ------------------------------------------------------------------ 

82 # Mutations 

83 # ------------------------------------------------------------------ 

84 

85 def update(self, changes: dict[str, Any], actor_id: str = "") -> None: 

86 """Apply a partial update and record a UserUpdated event. 

87 

88 Args: 

89 changes: Mapping of field names to new values. 

90 actor_id: Identity of the principal performing the action. 

91 """ 

92 for key, value in changes.items(): 

93 if hasattr(self, key): 

94 setattr(self, key, value) 

95 self._record_event( 

96 UserUpdated( 

97 user_id=str(self.id), 

98 changes=changes, 

99 actor_id=actor_id or None, 

100 ) 

101 ) 

102 

103 def deactivate(self, actor_id: str = "") -> None: 

104 """Deactivate this user and record a UserDeactivated event. 

105 

106 Args: 

107 actor_id: Identity of the principal performing the action. 

108 """ 

109 self.is_active = False 

110 self._record_event( 

111 UserDeactivated( 

112 user_id=str(self.id), 

113 actor_id=actor_id or None, 

114 ) 

115 ) 

116 

117 def soft_delete(self, actor_id: str = "") -> None: 

118 """Soft-delete this user and record a UserDeleted event. 

119 

120 Sets ``deleted_at`` to now and ``is_active`` to False. 

121 

122 Args: 

123 actor_id: Identity of the principal performing the action. 

124 """ 

125 self.deleted_at = datetime.now(UTC) 

126 self.is_active = False 

127 self._record_event( 

128 UserDeleted( 

129 user_id=str(self.id), 

130 actor_id=actor_id or None, 

131 ) 

132 )