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