Coverage for  / home / runner / work / bijux-cli / bijux-cli / src / bijux_cli / core / engine.py: 99%

1# SPDX-License-Identifier: Apache-2.0 

2# Copyright © 2025 Bijan Mousavi 

3 

4"""Provides the core runtime engine for the Bijux CLI. 

5 

6This module defines the `Engine` class, which is responsible for orchestrating 

7the application's runtime environment after initial setup. Its key 

8responsibilities include: 

9 

10 * Initializing and registering all default services with the Dependency 

11 Injection (DI) container. 

12 * Discovering, loading, and registering all external plugins. 

13 * Providing a central method for dispatching commands to plugins. 

14 * Managing the graceful shutdown of services. 

15 

16The engine acts as the bridge between the CLI command layer and the 

17underlying services and plugins. 

18""" 

19 

20from __future__ import annotations 

21 

22import asyncio 

23import inspect 

24import os 

25from typing import TYPE_CHECKING, Any 

26 

27if TYPE_CHECKING: 

28 from bijux_cli.core.di import DIContainer 

29 

30from bijux_cli.core.enums import ColorMode, LogLevel, OutputFormat 

31from bijux_cli.core.errors import PluginError 

32from bijux_cli.core.precedence import resolve_output_flags 

33from bijux_cli.plugins import get_plugins_dir, load_plugin 

34from bijux_cli.plugins.contracts import RegistryProtocol 

35from bijux_cli.plugins.services import register_plugin_services 

36from bijux_cli.services import register_default_services 

37from bijux_cli.services.history import History 

38from bijux_cli.services.logging.contracts import LoggingConfig 

39from bijux_cli.services.logging.observability import Observability 

40 

41 

42class Engine: 

43 """Orchestrates the CLI's runtime services and plugin lifecycle. 

44 

45 Attributes: 

46 _di (DIContainer): The dependency injection container. 

47 _format (OutputFormat): The default output format. 

48 _quiet (bool): The quiet mode flag. 

49 _registry (RegistryProtocol): The plugin registry service. 

50 """ 

51 

52 def __init__( 

53 self, 

54 di: Any = None, 

55 *, 

56 log_level: LogLevel = LogLevel.INFO, 

57 fmt: OutputFormat = OutputFormat.JSON, 

58 quiet: bool = False, 

59 logging_config: LoggingConfig | None = None, 

60 ) -> None: 

61 """Initializes the engine and its core services. 

62 

63 This sets up the DI container, registers default services, and loads 

64 all discoverable plugins. 

65 

66 Args: 

67 di (Any, optional): An existing dependency injection container. If 

68 None, the global singleton instance is used. Defaults to None. 

69 log_level (LogLevel): The default log level for services. 

70 fmt (OutputFormat): The default output format for services. 

71 quiet (bool): If True, suppresses output from services. 

72 logging_config (LoggingConfig | None): Optional logging configuration 

73 override for service registration. 

74 """ 

75 from bijux_cli.core.di import DIContainer 

76 

77 self._di = di or DIContainer.current() 

78 self._format = fmt 

79 self._quiet = quiet 

80 if logging_config is None: 80 ↛ 92line 80 didn't jump to line 92 because the condition on line 80 was always true

81 resolved = resolve_output_flags( 

82 quiet=quiet, 

83 pretty=True, 

84 log_level=log_level, 

85 color=ColorMode.AUTO, 

86 ) 

87 logging_config = LoggingConfig( 

88 quiet=quiet, 

89 log_level=resolved.log_level, 

90 color=resolved.color, 

91 ) 

92 register_default_services( 

93 self._di, 

94 logging_config=logging_config, 

95 output_format=fmt, 

96 ) 

97 register_plugin_services(self._di) 

98 self._di.register(Engine, self) 

99 self._registry: RegistryProtocol = self._di.resolve(RegistryProtocol) 

100 self._register_plugins() 

101 

102 async def run_command(self, name: str, *args: Any, **kwargs: Any) -> Any: 

103 """Executes a plugin's command with a configured timeout. 

104 

105 Args: 

106 name (str): The name of the command or plugin to run. 

107 *args (Any): Positional arguments to pass to the plugin's `execute` 

108 method. 

109 **kwargs (Any): Keyword arguments to pass to the plugin's `execute` 

110 method. 

111 

112 Returns: 

113 Any: The result of the command's execution. 

114 

115 Raises: 

116 PluginError: If the plugin is not found, its `execute` method 

117 is invalid, or if it fails during execution. 

118 """ 

119 plugin = self._registry.get(name) 

120 execute = getattr(plugin, "execute", None) 

121 if not callable(execute): 

122 raise PluginError( 

123 f"Plugin '{name}' has no callable 'execute' method.", http_status=404 

124 ) 

125 if not inspect.iscoroutinefunction(execute): 

126 raise PluginError( 

127 f"Plugin '{name}' 'execute' is not async/coroutine.", http_status=400 

128 ) 

129 try: 

130 return await asyncio.wait_for(execute(*args, **kwargs), self._timeout()) 

131 except Exception as exc: # pragma: no cover 

132 raise PluginError(f"Failed to run plugin '{name}': {exc}") from exc 

133 

134 async def run_repl(self) -> None: 

135 """Runs the interactive shell (REPL). 

136 

137 Note: This is a placeholder for future REPL integration. 

138 """ 

139 pass 

140 

141 async def shutdown(self) -> None: 

142 """Gracefully shuts down the engine and all resolved services. 

143 

144 This method orchestrates the termination sequence for the application's 

145 runtime. It first attempts to flush any buffered command history to 

146 disk and then proceeds to shut down the main dependency injection 

147 container, which in turn cleans up all resolved services. 

148 

149 Returns: 

150 None: 

151 """ 

152 try: 

153 self._di.resolve(History).flush() 

154 except KeyError: 

155 pass 

156 finally: 

157 await self._di.shutdown() 

158 

159 def _register_plugins(self) -> None: 

160 """Discovers, loads, and registers all plugins from the filesystem. 

161 

162 This method scans the plugins directory for valid plugin subdirectories. 

163 For each one found, it dynamically imports the `plugin.py` file, 

164 executes an optional `startup(di)` hook if present, and registers the 

165 plugin with the application's plugin registry. Errors encountered while 

166 loading a single plugin are logged and suppressed to allow other 

167 plugins to load. 

168 

169 Returns: 

170 None: 

171 """ 

172 plugins_dir = get_plugins_dir() 

173 plugins_dir.mkdir(parents=True, exist_ok=True) 

174 telemetry = self._di.resolve(Observability) 

175 for folder in plugins_dir.iterdir(): 

176 if not folder.is_dir(): 

177 continue 

178 path = folder / "src" / folder.name.replace("-", "_") / "plugin.py" 

179 if not path.exists(): 

180 continue 

181 module_name = ( 

182 folder.name.replace("-", "_") 

183 if folder.name.startswith("bijux_plugin_") 

184 else f"bijux_plugin_{folder.name.replace('-', '_')}" 

185 ) 

186 try: 

187 plugin = load_plugin(path, module_name) 

188 if startup := getattr(plugin, "startup", None): 

189 startup(self._di) 

190 self._registry.register( 

191 plugin.name, plugin, alias=None, version=plugin.version 

192 ) 

193 except Exception as e: # pragma: no cover 

194 telemetry.log("error", f"Loading plugin {folder.name} failed: {e}") 

195 

196 def _timeout(self) -> float: 

197 """Retrieves the command timeout from the configuration service. 

198 

199 Returns: 

200 float: The command timeout duration in seconds. 

201 

202 Raises: 

203 ValueError: If the timeout value in the configuration is malformed. 

204 """ 

205 from bijux_cli.cli.core.constants import ENV_COMMAND_TIMEOUT 

206 

207 raw = os.getenv(ENV_COMMAND_TIMEOUT, "30.0") 

208 try: 

209 return float(raw) 

210 except (TypeError, ValueError) as err: 

211 raise ValueError(f"Invalid timeout configuration: {raw!r}") from err 

212 

213 @property 

214 def di(self) -> DIContainer: 

215 """Read-only access to the DI container.""" 

216 return self._di 

217 

218 

219__all__ = ["Engine"]