Coverage for src / lexigram / admin / auth / store / memory.py: 27%

1""" 

2In-memory admin user store implementation. 

3""" 

4 

5from __future__ import annotations 

6 

7from typing import TYPE_CHECKING, Any 

8 

9from lexigram.admin.auth.store.base import AbstractAdminUserStore 

10from lexigram.admin.auth.user import AdminUserRecord 

11from lexigram.logging import get_logger 

12 

13if TYPE_CHECKING: 

14 from lexigram.contracts import AuthenticatedUserProtocol 

15 from lexigram.contracts.auth import PasswordHasherProtocol 

16from lexigram.di.decorators import inject 

17 

18logger = get_logger(__name__) 

19 

20 

21@inject 

22class MemoryAdminUserStore(AbstractAdminUserStore): 

23 """In-memory store for admin users with authentication support. 

24 

25 Attributes: 

26 _users_by_id: Dictionary of users by ID 

27 _users_by_email: Dictionary of users by email 

28 _users_by_username: Dictionary of users by username 

29 """ 

30 

31 def __init__(self, config: Any, hasher: PasswordHasherProtocol | None = None): 

32 """Initialize user store from configuration.""" 

33 self._hasher = hasher 

34 self._users_by_id: dict[str, AdminUserRecord] = {} 

35 self._users_by_email: dict[str, AdminUserRecord] = {} 

36 self._users_by_username: dict[str, AdminUserRecord] = {} 

37 

38 # Handle both legacy and modern Pydantic config 

39 users = getattr(config, "users", []) 

40 

41 # Load users from config 

42 for user_data in users: 

43 # Check if it's a Pydantic model (modern config) 

44 if hasattr(user_data, "model_dump"): 

45 # Map AuthUserConfig to User 

46 user = AdminUserRecord( 

47 user_id=user_data.username, 

48 name=user_data.username, 

49 email=user_data.email, 

50 hashed_password=user_data.password_hash 

51 or user_data.password, # Very basic mapping 

52 roles=user_data.roles, 

53 permissions=[], # Needs flattening from roles if strictly mimicking legacy 

54 is_active=user_data.is_active, 

55 is_verified=True, 

56 ) 

57 elif isinstance(user_data, dict): 

58 # Map dict to AdminUserRecord 

59 user = AdminUserRecord( 

60 user_id=user_data.get("username", ""), 

61 name=user_data.get("username", ""), 

62 email=user_data.get("email", ""), 

63 hashed_password=user_data.get("password_hash") 

64 or user_data.get("password"), 

65 roles=user_data.get("roles", []), 

66 permissions=user_data.get("permissions", []), 

67 is_active=user_data.get("is_active", True), 

68 is_verified=True, 

69 ) 

70 else: 

71 user = user_data 

72 

73 # Safe ID extraction 

74 user_id = getattr(user, "user_id", None) 

75 if not user_id: 

76 user_id = getattr(user, "username", "") 

77 

78 if user_id: 

79 self._users_by_id[user_id] = user 

80 

81 email = getattr(user, "email", "") 

82 if email: 

83 self._users_by_email[email.lower()] = user 

84 

85 username = getattr(user, "username", "") 

86 if username: 

87 self._users_by_username[username.lower()] = user 

88 

89 async def get_by_id(self, user_id: str) -> AuthenticatedUserProtocol | None: 

90 """Get user by ID. 

91 

92 Args: 

93 user_id: User ID 

94 

95 Returns: 

96 User if found, None otherwise 

97 """ 

98 return self._users_by_id.get(user_id) 

99 

100 async def get_by_email(self, email: str) -> AuthenticatedUserProtocol | None: 

101 """Get user by email address. 

102 

103 Args: 

104 email: Email address (case-insensitive) 

105 

106 Returns: 

107 User if found, None otherwise 

108 """ 

109 return self._users_by_email.get(email.lower()) 

110 

111 async def get_by_username(self, username: str) -> AuthenticatedUserProtocol | None: 

112 """Get user by username. 

113 

114 Args: 

115 username: Username (case-insensitive) 

116 

117 Returns: 

118 User if found, None otherwise 

119 """ 

120 return self._users_by_username.get(username.lower()) 

121 

122 async def authenticate( 

123 self, email: str, password: str 

124 ) -> AuthenticatedUserProtocol | None: 

125 """Authenticate user by email and password. 

126 

127 Args: 

128 email: Email address 

129 password: Plain text password 

130 

131 Returns: 

132 User if authentication successful, None otherwise 

133 """ 

134 user = await self.get_by_email(email) 

135 

136 if not user: 

137 return None 

138 

139 if not user.is_active: 

140 return None 

141 

142 if not user.hashed_password: # type: ignore[attr-defined] 

143 return None 

144 if self._hasher: 

145 verified = await self._hasher.verify(password, user.hashed_password) # type: ignore[attr-defined] 

146 else: 

147 import hashlib 

148 

149 verified = ( 

150 hashlib.sha256(password.encode()).hexdigest() == user.hashed_password # type: ignore[attr-defined] 

151 ) 

152 if not verified: 

153 return None 

154 

155 return user 

156 

157 async def count(self) -> int: 

158 """Get total number of users. 

159 

160 Returns: 

161 Number of users in store 

162 """ 

163 return len(self._users_by_id)