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

1# SPDX-License-Identifier: MIT 

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 

24from typing import TYPE_CHECKING, Any 

25 

26if TYPE_CHECKING: 

27 from bijux_cli.core.di import DIContainer 

28 

29from bijux_cli.contracts import ConfigProtocol, RegistryProtocol 

30from bijux_cli.core.enums import OutputFormat 

31from bijux_cli.core.exceptions import CommandError 

32from bijux_cli.infra.observability import Observability 

33from bijux_cli.services import register_default_services 

34from bijux_cli.services.history import History 

35from bijux_cli.services.plugins import get_plugins_dir, load_plugin 

36 

37 

38class Engine: 

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

40 

41 Attributes: 

42 _di (DIContainer): The dependency injection container. 

43 _debug (bool): The debug mode flag. 

44 _format (OutputFormat): The default output format. 

45 _quiet (bool): The quiet mode flag. 

46 _registry (RegistryProtocol): The plugin registry service. 

47 """ 

48 

49 def __init__( 

50 self, 

51 di: Any = None, 

52 *, 

53 debug: bool = False, 

54 fmt: OutputFormat = OutputFormat.JSON, 

55 quiet: bool = False, 

56 ) -> None: 

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

58 

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

60 all discoverable plugins. 

61 

62 Args: 

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

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

65 debug (bool): If True, enables debug mode for services. 

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

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

68 """ 

69 from bijux_cli.core.di import DIContainer 

70 

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

72 self._debug = debug 

73 self._format = fmt 

74 self._quiet = quiet 

75 self._di.register(Observability, lambda: Observability(debug=debug)) 

76 register_default_services(self._di, debug=debug, output_format=fmt, quiet=quiet) 

77 self._di.register(Engine, self) 

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

79 self._register_plugins() 

80 

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

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

83 

84 Args: 

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

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

87 method. 

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

89 method. 

90 

91 Returns: 

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

93 

94 Raises: 

95 CommandError: If the plugin is not found, its `execute` method 

96 is invalid, or if it fails during execution. 

97 """ 

98 plugin = self._registry.get(name) 

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

100 if not callable(execute): 

101 raise CommandError( 

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

103 ) 

104 if not inspect.iscoroutinefunction(execute): 

105 raise CommandError( 

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

107 ) 

108 try: 

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

110 except Exception as exc: # pragma: no cover 

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

112 

113 async def run_repl(self) -> None: 

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

115 

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

117 """ 

118 pass 

119 

120 async def shutdown(self) -> None: 

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

122 

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

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

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

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

127 

128 Returns: 

129 None: 

130 """ 

131 try: 

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

133 except KeyError: 

134 pass 

135 finally: 

136 await self._di.shutdown() 

137 

138 def _register_plugins(self) -> None: 

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

140 

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

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

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

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

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

146 plugins to load. 

147 

148 Returns: 

149 None: 

150 """ 

151 plugins_dir = get_plugins_dir() 

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

153 telemetry = self._di.resolve(Observability) 

154 for folder in plugins_dir.iterdir(): 

155 if not folder.is_dir(): 

156 continue 

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

158 if not path.exists(): 

159 continue 

160 module_name = ( 

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

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

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

164 ) 

165 try: 

166 plugin = load_plugin(path, module_name) 

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

168 startup(self._di) 

169 self._registry.register(plugin.name, plugin, version=plugin.version) 

170 except Exception as e: # pragma: no cover 

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

172 

173 def _timeout(self) -> float: 

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

175 

176 Returns: 

177 float: The command timeout duration in seconds. 

178 

179 Raises: 

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

181 """ 

182 try: 

183 cfg = self._di.resolve(ConfigProtocol) 

184 raw = cfg.get("BIJUXCLI_COMMAND_TIMEOUT", default=30.0) 

185 except KeyError: 

186 raw = 30.0 

187 value = raw.get("value", 30.0) if isinstance(raw, dict) else raw 

188 try: 

189 return float(value) 

190 except (TypeError, ValueError) as err: 

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

192 

193 @property 

194 def di(self) -> DIContainer: 

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

196 return self._di 

197 

198 

199__all__ = ["Engine"]