Coverage for src / lexigram / admin / integrations / features.py: 68%

1"""Features integration — gates UI elements behind feature flags.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING, Any 

6 

7if TYPE_CHECKING: 

8 from lexigram.contracts.core.di import ( 

9 ContainerRegistrarProtocol, 

10 ContainerResolverProtocol, 

11 ) 

12 

13 

14class _NoOpFeatures: 

15 def is_enabled(self, flag: str, context: Any = None) -> bool: 

16 return True 

17 

18 

19class FeaturesIntegration: 

20 """Adapter that checks feature flags for admin UI gating. 

21 

22 Gracefully no-ops (all flags pass) when ``lexigram-features`` is not 

23 installed or the integration is disabled. 

24 """ 

25 

26 def __init__(self, config: Any) -> None: 

27 self._config = config 

28 self._flags: Any = None 

29 self._enabled = False 

30 

31 def register(self, container: ContainerRegistrarProtocol) -> None: 

32 from lexigram.admin.config import FeaturesIntegrationConfig 

33 from lexigram.admin.integrations._optional import is_installed 

34 

35 cfg = self._config 

36 if not isinstance(cfg, FeaturesIntegrationConfig): 

37 cfg = FeaturesIntegrationConfig() 

38 if not cfg.enabled: 

39 self._flags = _NoOpFeatures() 

40 return 

41 if not is_installed("lexigram.features"): 

42 self._flags = _NoOpFeatures() 

43 return 

44 self._enabled = True 

45 

46 async def boot(self, container: ContainerResolverProtocol) -> None: 

47 if not self._enabled: 

48 return 

49 try: 

50 from lexigram.contracts.feature_flags import FlagManagerProtocol 

51 

52 self._flags = await container.resolve(FlagManagerProtocol) 

53 except Exception: # noqa: BLE001 

54 self._flags = _NoOpFeatures() 

55 

56 async def shutdown(self) -> None: 

57 pass 

58 

59 async def health_check(self) -> dict[str, Any]: 

60 return { 

61 "status": "healthy" 

62 if not isinstance(self._flags, _NoOpFeatures) 

63 else "noop" 

64 } 

65 

66 def is_enabled(self, flag: str, context: Any = None) -> bool: 

67 return ( 

68 self._flags.is_enabled(flag, context) 

69 if hasattr(self._flags, "is_enabled") 

70 else True 

71 ) 

72 

73 async def is_enabled_async(self, flag: str, context: Any = None) -> bool: 

74 return ( 

75 self._flags.is_enabled(flag, context) 

76 if hasattr(self._flags, "is_enabled") 

77 else True 

78 ) 

79 

80 

81__all__ = ["FeaturesIntegration"]