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

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 async def _mount_resources( 

20 self: AdminProvider, resolver: Any, ctx: MountContext 

21 ) -> None: 

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

23 

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

25 strict resource resolution is configured. 

26 

27 Args: 

28 resolver: The DI resolver for resource resolution. 

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

30 """ 

31 from lexigram.admin.rbac.inventory import PermissionInventoryService 

32 

33 for resource_cls in self._resources: 

34 name = ( 

35 getattr(resource_cls, "name", None) 

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

37 ) 

38 try: 

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

40 resource_cls, 

41 bypass_visibility=True, 

42 ) 

43 except Exception as exc: 

44 _log.error( 

45 "admin.resource_resolution_failed", 

46 resource=resource_cls.__name__, 

47 error=str(exc), 

48 strict=self._config.strict_resource_resolution, 

49 ) 

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

51 if self._config.strict_resource_resolution: 

52 raise 

53 

54 # Populate the RBAC permission inventory from registered resources 

55 try: 

56 inventory = await resolver.resolve( 

57 PermissionInventoryService, 

58 bypass_visibility=True, 

59 ) 

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

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

62 _log.warning( 

63 "admin.rbac_inventory_discovery_failed", 

64 error=str(exc), 

65 ) 

66 

67 async def _mount_settings_service( 

68 self: AdminProvider, resolver: Any, ctx: MountContext 

69 ) -> None: 

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

71 

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

73 in-memory instance so the panel still mounts. 

74 

75 Args: 

76 resolver: The DI resolver for settings dependencies. 

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

78 """ 

79 admin_settings_service: Any = None 

80 try: 

81 from lexigram.admin.services.settings_service import ( 

82 AdminSettingsDbProvider, 

83 AdminSettingsService, 

84 ) 

85 from lexigram.contracts.data import DatabaseProviderProtocol 

86 

87 db_provider = await resolver.resolve( 

88 DatabaseProviderProtocol, 

89 bypass_visibility=True, 

90 ) 

91 config_provider = AdminSettingsDbProvider(db=db_provider) 

92 # Ensure the tenant_configs table is created eagerly at startup 

93 try: 

94 await config_provider._ensure_table() 

95 except Exception: 

96 _log.warning("admin.tenant_config_table_creation_failed") 

97 admin_settings_service = AdminSettingsService( 

98 config_provider=config_provider, 

99 ) 

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

101 from lexigram.admin.settings.store import TenantConfigStore 

102 

103 try: 

104 registry = await resolver.resolve( 

105 ConfigRegistry, 

106 bypass_visibility=True, 

107 ) 

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

109 except Exception: 

110 _log.warning("admin.settings_store_registration_failed") 

111 except Exception: 

112 try: 

113 admin_settings_service = AdminSettingsService() 

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

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

116 ctx.settings_service = admin_settings_service