Coverage for src/lexigram/admin/auth/store/app_principal.py: 73%

1"""AdminUserStoreProtocol over AdminPrincipalProviderProtocol. 

2 

3Lets the panel run entirely against the application's own users 

4(spec D3) — no admin_users table involved. Delegate shape: 

5 

6- get_user_by_email/id, list_users, get_admin_count -> provider lookups 

7 (mapping to AdminUserRecord with user_id=principal.user_id, roles, 

8 permissions, is_active; hashed_password="") 

9- create_user -> provider.create_principal(name, email, password=raw) 

10 (NOTE: AdminUserStoreProtocol.create_user receives a hashed_password; 

11 in app mode the adapter passes the raw value through as-is — the app's 

12 hashing policy owns verification; documented for implementers) 

13- update_user -> provider.update_principal + provider.sync_roles when 

14 roles changed (a non-empty hashed_password on the record is forwarded 

15 onto the principal so implementers can persist panel-side mutations) 

16- delete_user -> provider.delete_principal 

17- authenticate -> provider.authenticate 

18- ensure_schema -> provider.ensure_schema (no-op passthrough) 

19""" 

20 

21from __future__ import annotations 

22 

23from typing import Any 

24 

25from lexigram.admin.auth.errors import SetupAlreadyCompletedError 

26from lexigram.admin.auth.user import AdminUserRecord 

27from lexigram.contracts.admin import AdminPrincipal, AdminPrincipalProviderProtocol 

28from lexigram.result import Err, Ok, Result 

29 

30 

31class AppPrincipalUserStoreAdapter: 

32 """Adapter implementing the admin store seam over the app's principal bridge.""" 

33 

34 def __init__(self, provider: AdminPrincipalProviderProtocol) -> None: 

35 self._provider = provider 

36 

37 async def get_admin_count(self) -> int: 

38 """Return the count of principals exposed by the app (see protocol docs).""" 

39 return len(await self._provider.list_principals()) 

40 

41 async def ensure_schema(self) -> None: 

42 """Delegate schema ownership to the app provider (passthrough).""" 

43 await self._provider.ensure_schema() 

44 

45 async def list_users(self) -> list[Any]: 

46 """Return all principals as admin user records (see protocol docs).""" 

47 return [self._to_record(p) for p in await self._provider.list_principals()] 

48 

49 async def create_user( 

50 self, 

51 name: str, 

52 email: str, 

53 hashed_password: str, 

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

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

56 **kwargs: Any, 

57 ) -> Any: 

58 """Create a principal; the password value passes through as-is.""" 

59 del permissions, kwargs 

60 created = await self._provider.create_principal( 

61 name, email, hashed_password, roles=roles 

62 ) 

63 return self._to_record(created) 

64 

65 async def get_user_by_email(self, email: str) -> Any | None: 

66 """Look up a principal by email (see protocol docs).""" 

67 for p in await self._provider.list_principals(): 

68 if p.email == email: 

69 return self._to_record(p) 

70 return None 

71 

72 async def claim_first_admin( 

73 self, 

74 name: str, 

75 email: str, 

76 hashed_password: str, 

77 roles: list[str], 

78 ) -> Result[Any, SetupAlreadyCompletedError]: 

79 """Create the first admin principal when the app exposes none. 

80 

81 Atomicity is delegated to the app's principal provider — the app 

82 owns principal creation semantics in app mode. 

83 

84 Args: 

85 name: Display name. 

86 email: Unique email address — used as the login identifier. 

87 hashed_password: Pre-hashed credential (passed through as-is, 

88 mirroring :meth:`create_user`). 

89 roles: Role strings for the new account. 

90 

91 Returns: 

92 Ok(record) when this call created the first admin principal; 

93 ``Err(SetupAlreadyCompletedError)`` when the app already exposes 

94 at least one principal and nothing was created. 

95 """ 

96 if await self.get_admin_count() > 0: 

97 return Err(SetupAlreadyCompletedError()) 

98 created = await self._provider.create_principal( 

99 name, email, hashed_password, roles=roles 

100 ) 

101 return Ok(self._to_record(created)) 

102 

103 async def get_user_by_id(self, user_id: str) -> Any | None: 

104 """Look up a principal by id (see protocol docs).""" 

105 p = await self._provider.principal_for(user_id) 

106 return self._to_record(p) if p else None 

107 

108 async def get_by_id(self, user_id: str) -> Any | None: 

109 """Alias of :meth:`get_user_by_id` required by ``AdminAuthMiddleware``. 

110 

111 Internal stores expose ``get_by_id`` (via ``AuthStoreBase``); the 

112 session-loading middleware calls it, so the app-mode adapter mirrors 

113 the shape for parity (record with ``is_active`` and ``user_id``). 

114 """ 

115 return await self.get_user_by_id(user_id) 

116 

117 async def update_user(self, user: Any) -> None: 

118 """Persist principal changes and sync roles through the provider. 

119 

120 A non-empty ``hashed_password`` on the record is forwarded onto the 

121 principal (panel password changes/resets land it before this call), 

122 so implementers' ``update_principal`` can persist it — mirroring how 

123 ``create_principal`` receives the pre-hashed value as-is. 

124 """ 

125 hashed_password = getattr(user, "hashed_password", "") or "" 

126 await self._provider.update_principal( 

127 AdminPrincipal( 

128 user_id=user.user_id, 

129 name=getattr(user, "name", ""), 

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

131 roles=list(getattr(user, "roles", []) or []), 

132 permissions=list(getattr(user, "permissions", []) or []), 

133 is_active=bool(getattr(user, "is_active", True)), 

134 **({"hashed_password": hashed_password} if hashed_password else {}), 

135 ) 

136 ) 

137 if hasattr(user, "roles") and user.roles is not None: 

138 await self._provider.sync_roles(user.user_id, list(user.roles)) 

139 

140 async def delete_user(self, user_id: str) -> None: 

141 """Delete a principal (see protocol docs).""" 

142 await self._provider.delete_principal(user_id) 

143 

144 async def authenticate(self, email: str, password: str) -> Any | None: 

145 """Authenticate against the app's own users (see protocol docs).""" 

146 p = await self._provider.authenticate(email, password) 

147 return self._to_record(p) if p else None 

148 

149 @staticmethod 

150 def _to_record(p: AdminPrincipal) -> AdminUserRecord: 

151 """Map a principal onto the admin store record shape.""" 

152 return AdminUserRecord( 

153 user_id=p.user_id, 

154 email=p.email, 

155 name=p.name, 

156 hashed_password="", 

157 roles=list(p.roles), 

158 permissions=list(p.permissions), 

159 is_active=p.is_active, 

160 ) 

161 

162 

163__all__ = ["AppPrincipalUserStoreAdapter"]