Coverage for src/lexigram/admin/auth/store/protocols.py: 33%

1"""Admin user store protocol. 

2 

3Defines the formal contract for admin panel user store implementations. 

4Any class that satisfies these method signatures is a valid implementation. 

5""" 

6 

7from __future__ import annotations 

8 

9from typing import Any, Protocol, runtime_checkable 

10 

11from lexigram.admin.auth.errors import SetupAlreadyCompletedError 

12from lexigram.result import Result 

13 

14 

15@runtime_checkable 

16class AdminUserStoreProtocol(Protocol): 

17 """Protocol for admin panel user store operations. 

18 

19 This is the single authoritative contract for anything that stores and 

20 manages admin-panel user accounts (distinct from the application's own 

21 user store managed by lexigram-auth). 

22 

23 Implementations: 

24 - :class:`~lexigram.admin.auth.store.direct_sql.DirectSQLAdminUserStore` 

25 — production SQL backend (``admin_users`` table) 

26 - :class:`~lexigram.admin.auth.store.memory.MemoryAdminUserStore` 

27 — in-memory store for testing 

28 """ 

29 

30 async def get_admin_count(self) -> int: 

31 """Return the total number of admin-panel accounts. 

32 

33 Used by :class:`~lexigram.admin.middleware.setup.SetupMiddleware` to 

34 decide whether to redirect to the first-run setup wizard. 

35 

36 Returns: 

37 Non-negative integer count of admin users. 

38 """ 

39 ... 

40 

41 async def ensure_schema(self) -> None: 

42 """Create the admin_users table if it does not exist (idempotent). 

43 

44 Failures are logged and swallowed, never raised — matching the boot 

45 loop's swallow-and-log behavior. 

46 """ 

47 ... 

48 

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

50 """Return all admin users ordered by creation time.""" 

51 ... 

52 

53 async def create_user( 

54 self, 

55 name: str, 

56 email: str, 

57 hashed_password: str, 

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

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

60 **kwargs: Any, 

61 ) -> Any: 

62 """Create (or upsert) an admin-panel user account. 

63 

64 Args: 

65 name: Display name. 

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

67 hashed_password: Pre-hashed credential. 

68 roles: Optional list of role strings (e.g. ``["superadmin"]``). 

69 permissions: Optional list of explicit permission strings. 

70 **kwargs: Implementation-specific extras (ignored if unsupported). 

71 

72 Returns: 

73 A lightweight object exposing at least ``user_id``, ``name``, and 

74 ``email`` attributes. 

75 """ 

76 ... 

77 

78 async def claim_first_admin( 

79 self, 

80 name: str, 

81 email: str, 

82 hashed_password: str, 

83 roles: list[str], 

84 ) -> Result[Any, SetupAlreadyCompletedError]: 

85 """Atomically create the first admin account if none exists yet. 

86 

87 Implementations must make the emptiness-check plus insert a single 

88 atomic operation (e.g. ``INSERT ... SELECT ... WHERE NOT EXISTS`` 

89 or an ``asyncio.Lock`` around an in-memory insert), so that two 

90 concurrent ``POST /setup`` submissions cannot both succeed. 

91 

92 Args: 

93 name: Display name. 

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

95 hashed_password: Pre-hashed credential. 

96 roles: Role strings for the new account (e.g. ``["superadmin"]``). 

97 

98 Returns: 

99 Ok(user-like object) when this call created the first admin 

100 account; ``Err(SetupAlreadyCompletedError)`` when at least one 

101 admin account already exists and nothing was inserted. 

102 """ 

103 ... 

104 

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

106 """Look up an admin user by email address. 

107 

108 Args: 

109 email: Email to search for. 

110 

111 Returns: 

112 User object or ``None`` when no match exists. 

113 """ 

114 ... 

115 

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

117 """Look up an admin user by primary key. 

118 

119 Args: 

120 user_id: Unique identifier (UUID string). 

121 

122 Returns: 

123 User object or ``None`` when no match exists. 

124 """ 

125 ... 

126 

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

128 """Look up the authenticated user backing a panel session. 

129 

130 Called by :class:`~lexigram.admin.middleware.auth.AdminAuthMiddleware` 

131 while loading the user stored in ``admin_sessions``; the returned 

132 object must expose ``user_id`` and ``is_active`` for that path. 

133 

134 Args: 

135 user_id: Unique identifier (UUID string). 

136 

137 Returns: 

138 User object or ``None`` when no match exists. 

139 """ 

140 ... 

141 

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

143 """Persist changes to an existing admin user. 

144 

145 Args: 

146 user: User object carrying updated field values. Must expose at 

147 least ``user_id``, ``name``, ``email``, ``roles``, 

148 ``permissions``, ``hashed_password``, and ``is_active``. 

149 """ 

150 ... 

151 

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

153 """Permanently remove an admin user account. 

154 

155 Args: 

156 user_id: Unique identifier of the user to delete. 

157 """ 

158 ... 

159 

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

161 """Authenticate an admin user by email and password. 

162 

163 Args: 

164 email: Email address to look up. 

165 password: Plain-text password to verify. 

166 

167 Returns: 

168 User object when credentials are valid and account is active, 

169 ``None`` otherwise. 

170 """ 

171 ... 

172 

173 

174__all__ = ["AdminUserStoreProtocol"]