Coverage for src/lexigram/admin/auth/store/memory.py: 52%
71 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 14:56 +0800
1"""
2In-memory admin user store implementation.
3"""
5from __future__ import annotations
7import asyncio
8from typing import TYPE_CHECKING, Any
9import uuid
11from lexigram.admin.auth.errors import SetupAlreadyCompletedError
12from lexigram.admin.auth.store.base import AbstractAdminUserStore
13from lexigram.admin.auth.user import AdminUserRecord
14from lexigram.logging import get_logger
15from lexigram.result import Err, Ok, Result
17if TYPE_CHECKING:
18 from lexigram.contracts import AuthenticatedUserProtocol
19 from lexigram.contracts.auth import PasswordHasherProtocol
20from lexigram.di.decorators import inject
22logger = get_logger(__name__)
25@inject
26class MemoryAdminUserStore(AbstractAdminUserStore):
27 """In-memory store for admin users with authentication support.
29 Attributes:
30 _users_by_id: Dictionary of users by ID
31 _users_by_email: Dictionary of users by email
32 _users_by_username: Dictionary of users by username
33 """
35 def __init__(self, config: Any, hasher: PasswordHasherProtocol | None = None):
36 """Initialize user store from configuration."""
37 self._hasher = hasher
38 self._users_by_id: dict[str, AdminUserRecord] = {}
39 self._users_by_email: dict[str, AdminUserRecord] = {}
40 self._users_by_username: dict[str, AdminUserRecord] = {}
41 self._claim_lock = asyncio.Lock()
43 # Handle both legacy and modern Pydantic config
44 users = getattr(config, "users", [])
46 # Load users from config
47 for user_data in users:
48 # Check if it's a Pydantic model (modern config)
49 if hasattr(user_data, "model_dump"):
50 # Map AuthUserConfig to User
51 user = AdminUserRecord(
52 user_id=user_data.username,
53 name=user_data.username,
54 email=user_data.email,
55 hashed_password=user_data.password_hash
56 or user_data.password, # Very basic mapping
57 roles=user_data.roles,
58 permissions=[], # Needs flattening from roles if strictly mimicking legacy
59 is_active=user_data.is_active,
60 is_verified=True,
61 )
62 elif isinstance(user_data, dict):
63 # Map dict to AdminUserRecord
64 user = AdminUserRecord(
65 user_id=user_data.get("username", ""),
66 name=user_data.get("username", ""),
67 email=user_data.get("email", ""),
68 hashed_password=user_data.get("password_hash")
69 or user_data.get("password"),
70 roles=user_data.get("roles", []),
71 permissions=user_data.get("permissions", []),
72 is_active=user_data.get("is_active", True),
73 is_verified=True,
74 )
75 else:
76 user = user_data
78 # Safe ID extraction
79 user_id = getattr(user, "user_id", None)
80 if not user_id:
81 user_id = getattr(user, "username", "")
83 if user_id:
84 self._users_by_id[user_id] = user
86 email = getattr(user, "email", "")
87 if email:
88 self._users_by_email[email.lower()] = user
90 username = getattr(user, "username", "")
91 if username:
92 self._users_by_username[username.lower()] = user
94 async def ensure_schema(self) -> None:
95 """No-op — the in-memory store has no table to create."""
96 return
98 async def get_by_id(self, user_id: str) -> AuthenticatedUserProtocol | None:
99 """Get user by ID.
101 Args:
102 user_id: User ID
104 Returns:
105 User if found, None otherwise
106 """
107 return self._users_by_id.get(user_id)
109 async def get_by_email(self, email: str) -> AuthenticatedUserProtocol | None:
110 """Get user by email address.
112 Args:
113 email: Email address (case-insensitive)
115 Returns:
116 User if found, None otherwise
117 """
118 return self._users_by_email.get(email.lower())
120 async def get_by_username(self, username: str) -> AuthenticatedUserProtocol | None:
121 """Get user by username.
123 Args:
124 username: Username (case-insensitive)
126 Returns:
127 User if found, None otherwise
128 """
129 return self._users_by_username.get(username.lower())
131 async def authenticate(
132 self, email: str, password: str
133 ) -> AuthenticatedUserProtocol | None:
134 """Authenticate user by email and password.
136 Args:
137 email: Email address
138 password: Plain text password
140 Returns:
141 User if authentication successful, None otherwise
142 """
143 user = await self.get_by_email(email)
145 if not user:
146 return None
148 if not user.is_active:
149 return None
151 if not user.hashed_password: # type: ignore[attr-defined]
152 return None
153 if self._hasher:
154 verified = await self._hasher.verify(password, user.hashed_password) # type: ignore[attr-defined]
155 else:
156 import hashlib
158 verified = (
159 hashlib.sha256(password.encode()).hexdigest() == user.hashed_password # type: ignore[attr-defined]
160 )
161 if not verified:
162 return None
164 return user
166 async def count(self) -> int:
167 """Get total number of users.
169 Returns:
170 Number of users in store
171 """
172 return len(self._users_by_id)
174 async def claim_first_admin(
175 self,
176 name: str,
177 email: str,
178 hashed_password: str,
179 roles: list[str],
180 ) -> Result[Any, SetupAlreadyCompletedError]:
181 """Atomically insert the first admin account when the store is empty.
183 The emptiness check and insert run under an ``asyncio.Lock``, so
184 concurrent first-run submissions cannot both insert.
186 Args:
187 name: Display name.
188 email: Unique email address — used as the login identifier.
189 hashed_password: Pre-hashed credential.
190 roles: Role strings for the new account.
192 Returns:
193 Ok(AdminUserRecord) when this call inserted the first admin
194 account; ``Err(SetupAlreadyCompletedError)`` when the store
195 already holds an admin account and nothing was inserted.
196 """
197 async with self._claim_lock:
198 if self._users_by_id:
199 return Err(SetupAlreadyCompletedError())
200 user = AdminUserRecord(
201 user_id=str(uuid.uuid4()),
202 name=name,
203 email=email,
204 hashed_password=hashed_password,
205 roles=roles,
206 permissions=[],
207 is_active=True,
208 )
209 self._users_by_id[user.user_id] = user
210 self._users_by_email[email.lower()] = user
211 self._users_by_username[name.lower()] = user
212 return Ok(user)