Coverage for src/lexigram/admin/di/mount/core.py: 0%

1"""Core mount phases: resource resolution and the admin settings service.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING, Any 

6 

7from lexigram.logging import get_logger 

8 

9if TYPE_CHECKING: 

10 from lexigram.admin.di.bundle_provider import AdminProvider 

11 from lexigram.admin.di.mount.context import MountContext 

12 

13_log = get_logger(__name__) 

14 

15 

16class AdminMountCoreMixin: 

17 """Mount phases that resolve admin resources and settings plumbing.""" 

18 

19 # Host attributes provided by AdminProvider. 

20 _config: Any 

21 _resources: list[Any] 

22 _controllers: list[Any] 

23 _mount_failures: dict[str, str] 

24 _authorizer_service: Any 

25 

26 _get_csrf_service: Any 

27 

28 

29 async def _mount_resources( 

30 self, resolver: Any, ctx: MountContext 

31 ) -> None: 

32 """Resolve registered resource classes into named instances. 

33 

34 Failures are recorded in ``_mount_failures`` and re-raised only when 

35 strict resource resolution is configured. 

36 

37 Args: 

38 resolver: The DI resolver for resource resolution. 

39 ctx: Mount pipeline state (``resources`` populated in place). 

40 """ 

41 from lexigram.admin.rbac.inventory import PermissionInventoryService 

42 

43 for resource_cls in self._resources: 

44 name = ( 

45 getattr(resource_cls, "name", None) 

46 or resource_cls.__name__.replace("Resource", "").lower() 

47 ) 

48 try: 

49 ctx.resources[name] = await resolver.resolve( 

50 resource_cls, 

51 bypass_visibility=True, 

52 ) 

53 except Exception as exc: 

54 _log.error( 

55 "admin.resource_resolution_failed", 

56 resource=resource_cls.__name__, 

57 error=str(exc), 

58 strict=self._config.strict_resource_resolution, 

59 ) 

60 self._mount_failures[f"resource:{resource_cls.__name__}"] = str(exc) 

61 if self._config.strict_resource_resolution: 

62 raise 

63 

64 # Populate the RBAC permission inventory from registered resources 

65 try: 

66 inventory = await resolver.resolve( 

67 PermissionInventoryService, 

68 bypass_visibility=True, 

69 ) 

70 inventory.register_resources(ctx.resources.keys()) 

71 except Exception as exc: # noqa: BLE001 — discovery is best-effort 

72 _log.warning( 

73 "admin.rbac_inventory_discovery_failed", 

74 error=str(exc), 

75 ) 

76 

77 async def _mount_settings_service( 

78 self, resolver: Any, ctx: MountContext 

79 ) -> None: 

80 """Create the shared settings service for runtime theme overrides. 

81 

82 Best-effort: without a database provider the service falls back to an 

83 in-memory instance so the panel still mounts. 

84 

85 Args: 

86 resolver: The DI resolver for settings dependencies. 

87 ctx: Mount pipeline state (``settings_service`` populated). 

88 """ 

89 admin_settings_service: Any = None 

90 try: 

91 from lexigram.admin.services.settings_service import ( 

92 AdminSettingsDbProvider, 

93 AdminSettingsService, 

94 ) 

95 from lexigram.contracts.data import DatabaseProviderProtocol 

96 

97 db_provider = await resolver.resolve( 

98 DatabaseProviderProtocol, 

99 bypass_visibility=True, 

100 ) 

101 config_provider = AdminSettingsDbProvider(db=db_provider) 

102 # Ensure the tenant_configs table is created eagerly at startup 

103 try: 

104 await config_provider._ensure_table() 

105 except Exception: 

106 _log.warning("admin.tenant_config_table_creation_failed") 

107 admin_settings_service = AdminSettingsService( 

108 config_provider=config_provider, 

109 ) 

110 from lexigram.admin.settings.panel.registry import ConfigRegistry 

111 from lexigram.admin.settings.store import TenantConfigStore 

112 

113 try: 

114 registry = await resolver.resolve( 

115 ConfigRegistry, 

116 bypass_visibility=True, 

117 ) 

118 registry.register_store("db", TenantConfigStore(admin_settings_service)) 

119 except Exception: 

120 _log.warning("admin.settings_store_registration_failed") 

121 except Exception: 

122 try: 

123 admin_settings_service = AdminSettingsService() 

124 except Exception as exc: # noqa: BLE001 — fallback is best-effort 

125 _log.warning("admin.settings_service_fallback_failed", reason=str(exc)) 

126 ctx.settings_service = admin_settings_service