Coverage for src/lexigram/admin/di/mount/controllers.py: 0%
211 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-24 23:18 +0800
1"""Mount phase that resolves and wires every admin controller."""
3from __future__ import annotations
5from typing import TYPE_CHECKING, Any
7from lexigram.logging import get_logger
9if TYPE_CHECKING:
10 from lexigram.admin.di.mount.context import MountContext
12_log = get_logger(__name__)
15class AdminMountControllersMixin:
16 """Resolves built-in admin controllers and wires their private services."""
18 # Host attributes provided by AdminProvider.
19 _config: Any
20 _resources: list[Any]
21 _controllers: list[Any]
22 _mount_failures: dict[str, str]
23 _authorizer_service: Any
24 _get_csrf_service: Any
26 async def _resolve_audit_service(self, resolver: Any) -> Any | None:
27 """Best-effort resolution of the optional audit-log service."""
29 try:
30 from lexigram.admin.auth.protocols import (
31 AdminAuditLogServiceProtocol,
32 )
34 return await resolver.resolve(
35 AdminAuditLogServiceProtocol,
36 bypass_visibility=True,
37 )
38 except Exception:
39 return None
41 async def _mount_controllers(self, resolver: Any, ctx: MountContext) -> None:
42 """Resolve every built-in controller, best-effort per controller.
44 Failures are recorded in ``_mount_failures`` and only abort the mount
45 when strict resource resolution is configured. Controllers receive
46 runtime wiring (settings, audit, CSRF, user store) as available.
48 Args:
49 resolver: The DI resolver for controller resolution.
50 ctx: Mount pipeline state (``controllers`` populated in place).
51 """
52 # Create shared settings_service for runtime theme overrides (best-effort)
53 admin_settings_service = ctx.settings_service
55 for controller_cls in self._controllers:
56 try:
57 instance = await resolver.resolve(
58 controller_cls,
59 bypass_visibility=True,
60 )
61 ctx.controllers.append(instance)
62 if admin_settings_service is not None and hasattr(
63 instance, "_settings_service"
64 ):
65 instance._settings_service = admin_settings_service
66 except Exception as exc:
67 _log.error(
68 "admin.controller_resolution_failed",
69 controller=controller_cls.__name__,
70 error=str(exc),
71 strict=self._config.strict_resource_resolution,
72 )
73 self._mount_failures[f"controller:{controller_cls.__name__}"] = str(exc)
74 if self._config.strict_resource_resolution:
75 raise
77 # Resolve built-in WidgetController (best-effort)
78 try:
79 from lexigram.admin.controllers.widgets import WidgetController
81 widget_controller = await resolver.resolve(
82 WidgetController,
83 bypass_visibility=True,
84 )
85 ctx.controllers.append(widget_controller)
86 if admin_settings_service is not None and hasattr(
87 widget_controller, "_settings_service"
88 ):
89 widget_controller._settings_service = admin_settings_service
90 audit_service = await self._resolve_audit_service(resolver)
91 if audit_service is not None and hasattr(
92 widget_controller, "_audit_service"
93 ):
94 widget_controller._audit_service = audit_service
95 csrf_service = await self._get_csrf_service(resolver)
96 if hasattr(widget_controller, "_csrf_service"):
97 widget_controller._csrf_service = csrf_service
98 except Exception as exc:
99 _log.error(
100 "admin.widget_controller_resolution_failed",
101 error=str(exc),
102 strict=self._config.strict_resource_resolution,
103 )
104 self._mount_failures["controller:WidgetController"] = str(exc)
105 if self._config.strict_resource_resolution:
106 raise
108 # Resolve built-in ImpersonationController (best-effort)
109 try:
110 from lexigram.admin.controllers.impersonation import (
111 ImpersonationController,
112 )
113 from lexigram.admin.services.impersonation import ImpersonationService
115 impersonation_service = await resolver.resolve(
116 ImpersonationService,
117 bypass_visibility=True,
118 )
119 if getattr(impersonation_service, "_audit", None) is None:
120 audit_service = await self._resolve_audit_service(resolver)
121 if audit_service is not None:
122 impersonation_service._audit = audit_service
124 impersonation_controller = await resolver.resolve(
125 ImpersonationController,
126 bypass_visibility=True,
127 )
128 ctx.controllers.append(impersonation_controller)
129 except Exception as exc:
130 _log.error(
131 "admin.impersonation_controller_resolution_failed",
132 error=str(exc),
133 strict=self._config.strict_resource_resolution,
134 )
135 self._mount_failures["controller:ImpersonationController"] = str(exc)
136 if self._config.strict_resource_resolution:
137 raise
139 # Resolve built-in TenancyController (best-effort)
140 try:
141 from lexigram.admin.controllers.tenancy import TenancyController
142 from lexigram.admin.multitenancy.adapter import TenantProviderRegistry
144 tenancy_controller = await resolver.resolve(
145 TenancyController,
146 bypass_visibility=True,
147 )
148 ctx.controllers.append(tenancy_controller)
149 if self._config.tenancy.enabled:
150 try:
151 tenancy_controller._registry = await resolver.resolve(
152 TenantProviderRegistry,
153 bypass_visibility=True,
154 )
155 except Exception:
156 tenancy_controller._registry = None
157 audit_service = await self._resolve_audit_service(resolver)
158 if audit_service is not None:
159 tenancy_controller._audit_service = audit_service
160 except Exception as exc:
161 _log.error(
162 "admin.tenancy_controller_resolution_failed",
163 error=str(exc),
164 strict=self._config.strict_resource_resolution,
165 )
166 self._mount_failures["controller:TenancyController"] = str(exc)
167 if self._config.strict_resource_resolution:
168 raise
170 # Resolve built-in DashboardController (best-effort)
171 try:
172 from lexigram.admin.controllers.dashboard import DashboardController
174 dashboard_controller = await resolver.resolve(
175 DashboardController,
176 bypass_visibility=True,
177 )
178 ctx.controllers.append(dashboard_controller)
179 if admin_settings_service is not None:
180 dashboard_controller._settings_service = admin_settings_service
181 except Exception as exc:
182 _log.error(
183 "admin.dashboard_controller_resolution_failed",
184 error=str(exc),
185 strict=self._config.strict_resource_resolution,
186 )
187 self._mount_failures["controller:DashboardController"] = str(exc)
188 if self._config.strict_resource_resolution:
189 raise
191 # Resolve built-in AuthController (login, logout)
192 try:
193 from lexigram.admin.controllers.auth import AuthController
195 auth_controller = await resolver.resolve(
196 AuthController,
197 bypass_visibility=True,
198 )
199 ctx.controllers.append(auth_controller)
200 if admin_settings_service is not None:
201 auth_controller._settings_service = admin_settings_service
203 # Wire self-service registration (opt-in via config). Best-effort:
204 # without a resolvable user store, registration stays disabled.
205 try:
206 from lexigram.admin.auth.store.protocols import (
207 AdminUserStoreProtocol,
208 )
210 auth_controller._user_store = await resolver.resolve(
211 AdminUserStoreProtocol,
212 bypass_visibility=True,
213 )
214 except Exception:
215 auth_controller._user_store = None
216 registration = getattr(self._config.auth, "registration", None)
217 auth_controller._registration_enabled = bool(
218 registration and registration.enabled
219 )
220 auth_controller._registration_default_role = (
221 str(registration.default_role) if registration else "admin"
222 )
223 auth_controller._registration_domains = (
224 list(registration.allowed_email_domains) if registration else []
225 )
226 except Exception as exc:
227 _log.error(
228 "admin.auth_controller_resolution_failed",
229 error=str(exc),
230 strict=self._config.strict_resource_resolution,
231 )
232 self._mount_failures["controller:AuthController"] = str(exc)
233 if self._config.strict_resource_resolution:
234 raise
236 # Resolve built-in ProfileController (profile page, password change)
237 try:
238 from lexigram.admin.controllers.profile import ProfileController
240 profile_controller = await resolver.resolve(
241 ProfileController,
242 bypass_visibility=True,
243 )
244 ctx.controllers.append(profile_controller)
245 if admin_settings_service is not None:
246 profile_controller._settings_service = admin_settings_service
247 try:
248 from lexigram.admin.auth.store.protocols import (
249 AdminUserStoreProtocol,
250 )
252 profile_controller._user_store = await resolver.resolve(
253 AdminUserStoreProtocol,
254 bypass_visibility=True,
255 )
256 except Exception:
257 profile_controller._user_store = None
258 except Exception as exc:
259 _log.error(
260 "admin.profile_controller_resolution_failed",
261 error=str(exc),
262 strict=self._config.strict_resource_resolution,
263 )
264 self._mount_failures["controller:ProfileController"] = str(exc)
265 if self._config.strict_resource_resolution:
266 raise
268 # Resolve built-in SetupController (first-run wizard)
269 try:
270 from lexigram.admin.controllers.setup import SetupController
272 setup_controller = await resolver.resolve(
273 SetupController,
274 bypass_visibility=True,
275 )
276 ctx.controllers.append(setup_controller)
277 if admin_settings_service is not None:
278 setup_controller._settings_service = admin_settings_service
279 except Exception as exc:
280 _log.error(
281 "admin.setup_controller_resolution_failed",
282 error=str(exc),
283 strict=self._config.strict_resource_resolution,
284 )
285 self._mount_failures["controller:SetupController"] = str(exc)
286 if self._config.strict_resource_resolution:
287 raise
289 # Mount ErrorController (styled error pages) — best-effort
290 try:
291 from lexigram.admin.controllers.error import ErrorController
293 error_controller = await resolver.resolve(
294 ErrorController,
295 bypass_visibility=True,
296 )
297 ctx.controllers.append(error_controller)
298 except Exception as exc:
299 _log.error(
300 "admin.error_controller_resolution_failed",
301 error=str(exc),
302 strict=self._config.strict_resource_resolution,
303 )
304 self._mount_failures["controller:ErrorController"] = str(exc)
305 if self._config.strict_resource_resolution:
306 raise
308 # Mount PoolHealthController (connection pool monitoring) — best-effort.
309 # pool_manager/task_manager are optional: without them the endpoints
310 # respond 503 instead of failing resolution.
311 try:
312 from lexigram.admin.controllers.pool_health import PoolHealthController
314 pool_health_controller = await resolver.resolve(
315 PoolHealthController,
316 bypass_visibility=True,
317 )
318 ctx.controllers.append(pool_health_controller)
319 except Exception as exc:
320 _log.error(
321 "admin.pool_health_controller_resolution_failed",
322 error=str(exc),
323 strict=self._config.strict_resource_resolution,
324 )
325 self._mount_failures["controller:PoolHealthController"] = str(exc)
326 if self._config.strict_resource_resolution:
327 raise
329 # Mount ProgressController (SSE/status progress tracking) — best-effort.
330 # Tries an integrator-registered tracker first; falls back to the
331 # admin-owned LocalProgressTracker (no dependency on optional
332 # integration packages — without one the controller mounts with the
333 # in-process tracker instead of being skipped).
334 try:
335 from lexigram.admin.controllers.progress import (
336 LocalProgressTracker,
337 ProgressController,
338 )
340 try:
341 progress_controller = await resolver.resolve(
342 ProgressController,
343 bypass_visibility=True,
344 )
345 except Exception:
346 progress_controller = ProgressController(tracker=LocalProgressTracker())
347 ctx.controllers.append(progress_controller)
348 except ModuleNotFoundError as exc:
349 _log.info(
350 "admin.progress_controller_skipped",
351 reason="progress_controller_unavailable",
352 error=str(exc),
353 )
354 except Exception as exc:
355 _log.error(
356 "admin.progress_controller_resolution_failed",
357 error=str(exc),
358 strict=self._config.strict_resource_resolution,
359 )
360 self._mount_failures["controller:ProgressController"] = str(exc)
361 if self._config.strict_resource_resolution:
362 raise
364 # Mount SettingsController (theme & branding settings)
365 try:
366 from lexigram.admin.controllers.settings import SettingsController
367 from lexigram.admin.engine.renderer import AdminRenderer
368 from lexigram.admin.settings.panel.registry import ConfigRegistry
370 settings_csrf = await self._get_csrf_service(resolver)
372 settings_registry: ConfigRegistry | None = None
373 try:
374 settings_registry = await resolver.resolve(
375 ConfigRegistry,
376 bypass_visibility=True,
377 )
378 except Exception as exc: # noqa: BLE001 — settings registry is optional
379 _log.warning("admin.config_registry_unavailable", reason=str(exc))
381 settings_audit = await self._resolve_audit_service(resolver)
383 renderer = await resolver.resolve(
384 AdminRenderer,
385 bypass_visibility=True,
386 )
387 settings_controller = SettingsController(
388 renderer=renderer,
389 settings_service=admin_settings_service,
390 csrf_service=settings_csrf,
391 audit_service=settings_audit,
392 registry=settings_registry,
393 rbac_config=self._config.rbac,
394 )
395 ctx.controllers.append(settings_controller)
396 except Exception as exc:
397 _log.warning(
398 "admin.settings_controller_skipped",
399 error=str(exc),
400 )
402 # Mount InfrastructureController (cluster landing page)
403 try:
404 from lexigram.admin.clusters import Cluster, ClusterRegistry
405 from lexigram.admin.controllers.clusters import ClusterCenterController
406 from lexigram.admin.controllers.infrastructure import (
407 InfrastructureController,
408 )
409 from lexigram.admin.engine.renderer import AdminRenderer
411 infra_renderer = await resolver.resolve(
412 AdminRenderer,
413 bypass_visibility=True,
414 )
415 ctx.controllers.append(InfrastructureController(renderer=infra_renderer))
417 # Build the cluster registry (built-in + config-declared extras)
418 # and mount a generic center controller per extra cluster.
419 cluster_registry = ClusterRegistry.with_defaults()
420 extra_specs = getattr(self._config, "clusters", None)
421 for spec in (extra_specs.extra if extra_specs else []) or []:
422 cluster = Cluster(
423 name=spec.name,
424 label=spec.label,
425 icon=spec.icon,
426 order=spec.order,
427 collapsible=spec.collapsible,
428 collapsed_by_default=spec.collapsed_by_default,
429 slug=spec.slug,
430 group=spec.group,
431 description=spec.description,
432 )
433 cluster_registry.register(cluster)
434 ctx.controllers.append(
435 ClusterCenterController(
436 renderer=infra_renderer,
437 cluster=cluster,
438 )
439 )
440 ctx.cluster_registry = cluster_registry
441 except Exception as exc:
442 _log.warning(
443 "admin.infrastructure_controller_skipped",
444 error=str(exc),
445 )
447 # Mount PluginsController (plugin listing & toggles) — best-effort.
448 try:
449 from lexigram.admin.controllers.plugins import PluginsController
450 from lexigram.admin.engine.renderer import AdminRenderer
452 plugins_csrf_service = await self._get_csrf_service(resolver)
454 plugins_audit_service = await self._resolve_audit_service(resolver)
456 plugins_renderer = await resolver.resolve(
457 AdminRenderer,
458 bypass_visibility=True,
459 )
460 ctx.controllers.append(
461 PluginsController(
462 renderer=plugins_renderer,
463 csrf_service=plugins_csrf_service,
464 audit_service=plugins_audit_service,
465 rbac_config=self._config.rbac,
466 )
467 )
468 except Exception as exc:
469 _log.warning(
470 "admin.plugins_controller_skipped",
471 error=str(exc),
472 )