Coverage for src/lexigram/auth/di/bundle_provider.py: 83%

1"""Convenience provider that registers the full authentication + authorisation stack. 

2 

3Import and register :class:`AuthBundleProvider` when you want the complete 

4auth subsystem in a single line instead of adding four separate providers. 

5""" 

6 

7from __future__ import annotations 

8 

9from typing import TYPE_CHECKING, Any, Self 

10 

11from lexigram.auth.config import AuthConfig 

12from lexigram.auth.di.sub_providers.admin_provider import AuthAdminProvider 

13from lexigram.auth.di.sub_providers.authentication_provider import ( 

14 AuthenticationProvider, 

15) 

16from lexigram.auth.di.sub_providers.authorization_provider import AuthorizationProvider 

17from lexigram.auth.di.sub_providers.google_oauth_provider import GoogleOAuthProvider 

18from lexigram.auth.di.sub_providers.session_provider import SessionProvider 

19from lexigram.auth.di.sub_providers.token_provider import TokenProvider 

20from lexigram.contracts.core import HealthCheckResult, HealthStatus, ProviderPriority 

21from lexigram.di.provider import Provider 

22from lexigram.logging import get_logger 

23 

24if TYPE_CHECKING: 

25 from lexigram.contracts.core.di import ( 

26 BootContainerProtocol, 

27 ContainerRegistrarProtocol, 

28 ) 

29 

30logger = get_logger(__name__) 

31 

32__all__ = ["AuthBundleProvider"] 

33 

34 

35class AuthBundleProvider(Provider): 

36 """Composite provider that wires the full Lexigram auth stack. 

37 

38 Composes :class:`~lexigram.auth.di.AuthenticationProvider`, 

39 :class:`~lexigram.auth.di.sub_providers.token_provider.TokenProvider`, 

40 :class:`~lexigram.auth.di.sub_providers.session_provider.SessionProvider`, 

41 and :class:`~lexigram.auth.di.AuthorizationProvider` so that callers only 

42 need to register a single provider: 

43 

44 .. code-block:: python 

45 

46 container.add_provider(AuthBundleProvider(config=auth_config)) 

47 

48 Dependencies registered by each sub-provider are available in the 

49 container after :meth:`register` completes. 

50 

51 Args: 

52 config: Shared :class:`~lexigram.auth.config.AuthConfig` forwarded to 

53 every sub-provider. When ``None``, each sub-provider uses its own 

54 defaults. 

55 initial_roles: Optional initial RBAC roles forwarded to 

56 :class:`~lexigram.auth.di.AuthorizationProvider`. 

57 enable_passkeys: When ``True``, append 

58 :class:`~lexigram.auth.di.sub_providers.passkey_provider.PasskeyProvider` 

59 to the sub-provider list (requires the WebAuthn extra to be 

60 installed). 

61 kwargs: Extra keyword arguments forwarded to 

62 :class:`~lexigram.auth.di.AuthenticationProvider`. 

63 """ 

64 

65 config_key: str | None = "auth" 

66 config_model: type | None = AuthConfig 

67 

68 def __init__( 

69 self, 

70 config: AuthConfig | None = None, 

71 initial_roles: dict[str, Any] | None = None, 

72 enable_passkeys: bool = False, 

73 **kwargs: Any, 

74 ) -> None: 

75 super().__init__(name="auth_bundle", priority=ProviderPriority.SECURITY) 

76 self.config = config 

77 self._authn = AuthenticationProvider(config=config) 

78 self._token = TokenProvider(config=config) 

79 self._session = SessionProvider(config=config) 

80 self._authz = AuthorizationProvider( 

81 config=config, initial_roles=initial_roles or {} 

82 ) 

83 self._admin = AuthAdminProvider(config=config) 

84 self._sub_providers = [ 

85 self._authn, 

86 self._token, 

87 self._session, 

88 self._authz, 

89 self._admin, 

90 ] 

91 google_oauth_config = ( 

92 getattr(config, "oauth2_providers", {}).get("google", {}) 

93 if config is not None 

94 else {} 

95 ) 

96 if google_oauth_config: 

97 self._sub_providers.append( 

98 GoogleOAuthProvider(config=config, google_oauth=google_oauth_config), 

99 ) 

100 if enable_passkeys: 

101 try: 

102 from lexigram.auth.di.sub_providers.passkey_provider import ( 

103 PasskeyProvider, 

104 ) 

105 

106 self._sub_providers.append(PasskeyProvider(config=config)) 

107 except ImportError: 

108 logger.warning( 

109 "auth.passkeys_unavailable", 

110 reason="PasskeyProvider could not be imported", 

111 ) 

112 

113 @classmethod 

114 def from_config(cls, config: AuthConfig, **context: Any) -> Self: 

115 """Create provider from config object.""" 

116 return cls(config=config) 

117 

118 # ------------------------------------------------------------------ 

119 # Provider interface 

120 # ------------------------------------------------------------------ 

121 

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

123 """Register all auth sub-providers with the container. 

124 

125 Args: 

126 container: The DI container registrar. 

127 """ 

128 for provider in self._sub_providers: 

129 await provider.register(container) 

130 logger.info("auth_bundle.registered") 

131 

132 async def boot(self, container: BootContainerProtocol) -> None: 

133 """Boot all auth sub-providers in registration order. 

134 

135 Args: 

136 container: The DI container resolver. 

137 """ 

138 for provider in self._sub_providers: 

139 await provider.boot(container) 

140 logger.info("auth_bundle.booted") 

141 

142 async def shutdown(self) -> None: 

143 """Shut down all auth sub-providers in reverse registration order.""" 

144 for provider in reversed(self._sub_providers): 

145 await provider.shutdown() 

146 logger.info("auth_bundle.shutdown") 

147 

148 async def health_check(self, timeout: float = 5.0) -> HealthCheckResult: 

149 """Aggregate health check across all sub-providers. 

150 

151 Returns :attr:`~lexigram.contracts.core.HealthStatus.DEGRADED` if any 

152 sub-provider is unhealthy. 

153 

154 Args: 

155 timeout: Per-provider timeout budget in seconds. 

156 

157 Returns: 

158 An aggregated :class:`~lexigram.contracts.core.HealthCheckResult`. 

159 """ 

160 results = [await p.health_check(timeout=timeout) for p in self._sub_providers] 

161 overall = ( 

162 HealthStatus.HEALTHY 

163 if all(r.status == HealthStatus.HEALTHY for r in results) 

164 else HealthStatus.DEGRADED 

165 ) 

166 return HealthCheckResult( 

167 component=self.name, 

168 status=overall, 

169 details={r.component: r.status.value for r in results}, 

170 )