Coverage for /home/runner/work/bijux-cli/bijux-cli/src/bijux_cli/commands/plugins/check.py: 100%

1# SPDX-License-Identifier: MIT 

2# Copyright © 2025 Bijan Mousavi 

3 

4"""Implements the `plugins check` subcommand for the Bijux CLI. 

5 

6This module contains the logic for performing a health check on a specific 

7installed plugin. It validates the plugin's files, dynamically imports its 

8code, and executes a `health()` hook function if available. The result is 

9reported in a structured, machine-readable format. 

10 

11Output Contract: 

12 * Healthy: `{"plugin": str, "status": "healthy"}` 

13 * Unhealthy: `{"plugin": str, "status": "unhealthy"}` (exits with code 1) 

14 * Verbose: Adds `{"python": str, "platform": str}` to the payload. 

15 * Error: `{"error": "...", "code": int}` (for pre-check failures) 

16 

17Exit Codes: 

18 * `0`: The plugin is healthy. 

19 * `1`: The plugin is unhealthy, could not be found, or an error occurred 

20 during import or execution. 

21 * `2`: An invalid flag was provided (e.g., bad format). 

22 * `3`: An ASCII or encoding error was detected in the environment. 

23""" 

24 

25from __future__ import annotations 

26 

27import asyncio 

28from collections.abc import Mapping 

29import importlib.util 

30import inspect 

31import json 

32import platform 

33import sys 

34import traceback 

35import types 

36from typing import Any 

37 

38import typer 

39 

40from bijux_cli.commands.utilities import ( 

41 ascii_safe, 

42 emit_error_and_exit, 

43 new_run_command, 

44 validate_common_flags, 

45) 

46from bijux_cli.core.constants import ( 

47 HELP_DEBUG, 

48 HELP_FORMAT, 

49 HELP_NO_PRETTY, 

50 HELP_QUIET, 

51 HELP_VERBOSE, 

52) 

53from bijux_cli.services.plugins import get_plugins_dir 

54 

55 

56def check_plugin( 

57 name: str = typer.Argument(..., help="Plugin name"), 

58 quiet: bool = typer.Option(False, "-q", "--quiet", help=HELP_QUIET), 

59 verbose: bool = typer.Option(False, "-v", "--verbose", help=HELP_VERBOSE), 

60 fmt: str = typer.Option("json", "-f", "--format", help=HELP_FORMAT), 

61 pretty: bool = typer.Option(True, "--pretty/--no-pretty", help=HELP_NO_PRETTY), 

62 debug: bool = typer.Option(False, "-d", "--debug", help=HELP_DEBUG), 

63) -> None: 

64 """Runs a health check on a specific installed plugin. 

65 

66 This function validates a plugin's structure, dynamically imports its 

67 `plugin.py` file, and executes its `health()` hook to determine its 

68 operational status. The final status is emitted as a structured payload. 

69 

70 Args: 

71 name (str): The name of the plugin to check. 

72 quiet (bool): If True, suppresses all output except for errors. 

73 verbose (bool): If True, includes Python/platform details in the output. 

74 fmt (str): The output format, "json" or "yaml". 

75 pretty (bool): If True, pretty-prints the output. 

76 debug (bool): If True, enables debug diagnostics. 

77 

78 Returns: 

79 None: 

80 

81 Raises: 

82 SystemExit: Always exits with a contract-compliant status code and 

83 payload, indicating the health status or detailing an error. 

84 """ 

85 command = "plugins check" 

86 

87 fmt_lower = validate_common_flags(fmt, command, quiet) 

88 

89 plug_dir = get_plugins_dir() / name 

90 plug_py = plug_dir / "plugin.py" 

91 meta_json = plug_dir / "plugin.json" 

92 

93 if not plug_py.is_file(): 

94 emit_error_and_exit( 

95 f'Plugin "{name}" not found', 

96 code=1, 

97 failure="not_found", 

98 command=command, 

99 fmt=fmt_lower, 

100 quiet=quiet, 

101 include_runtime=verbose, 

102 debug=debug, 

103 extra={"plugin": name}, 

104 ) 

105 

106 if not meta_json.is_file(): 

107 emit_error_and_exit( 

108 f'Plugin "{name}" metadata (plugin.json) is missing', 

109 code=1, 

110 failure="metadata_missing", 

111 command=command, 

112 fmt=fmt_lower, 

113 quiet=quiet, 

114 include_runtime=verbose, 

115 debug=debug, 

116 ) 

117 

118 try: 

119 meta = json.loads(meta_json.read_text("utf-8")) 

120 if not (isinstance(meta, dict) and meta.get("name") and meta.get("desc")): 

121 raise ValueError("Incomplete metadata") 

122 except Exception as exc: 

123 emit_error_and_exit( 

124 f'Plugin "{name}" metadata is corrupt: {exc}', 

125 code=1, 

126 failure="metadata_corrupt", 

127 command=command, 

128 fmt=fmt_lower, 

129 quiet=quiet, 

130 include_runtime=verbose, 

131 debug=debug, 

132 ) 

133 

134 mod_name = f"_bijux_cli_plugin_{name}" 

135 try: 

136 spec = importlib.util.spec_from_file_location(mod_name, plug_py) 

137 if not spec or not spec.loader: 

138 raise ImportError("Cannot create import spec") 

139 module = types.ModuleType(mod_name) 

140 sys.modules[mod_name] = module 

141 spec.loader.exec_module(module) 

142 except Exception as exc: 

143 err = f"Import error: {exc}" 

144 if debug: 

145 err += "\n" + traceback.format_exc() 

146 emit_error_and_exit( 

147 err, 

148 code=1, 

149 failure="import_error", 

150 command=command, 

151 fmt=fmt_lower, 

152 quiet=quiet, 

153 include_runtime=verbose, 

154 debug=debug, 

155 ) 

156 

157 async def _run_health() -> dict[str, Any]: 

158 """Isolates and executes the plugin's `health()` hook. 

159 

160 This function finds and calls the `health()` function within the 

161 imported plugin module. It handles both synchronous and asynchronous 

162 hooks, validates their signatures, and safely captures any exceptions 

163 during execution. 

164 

165 Returns: 

166 dict[str, Any]: A dictionary containing the health check result, 

167 which includes the plugin name and a status ('healthy' or 

168 'unhealthy'), or an error message. 

169 """ 

170 hook = getattr(module, "health", None) 

171 if not callable(hook): 

172 return {"plugin": name, "error": "No health() hook"} 

173 try: 

174 sig = inspect.signature(hook) 

175 if len(sig.parameters) != 1: 

176 return { 

177 "plugin": name, 

178 "error": "health() hook must take exactly one argument (di)", 

179 } 

180 except Exception as exc1: 

181 return {"plugin": name, "error": f"health() signature error: {exc1}"} 

182 try: 

183 if asyncio.iscoroutinefunction(hook): 

184 res = await hook(None) 

185 else: 

186 loop = asyncio.get_running_loop() 

187 res = await loop.run_in_executor(None, hook, None) 

188 except BaseException as exc2: 

189 return {"plugin": name, "error": str(exc2) or exc2.__class__.__name__} 

190 

191 if res is True: 

192 return {"plugin": name, "status": "healthy"} 

193 if res is False: 

194 return {"plugin": name, "status": "unhealthy"} 

195 if isinstance(res, dict) and res.get("status") in ("healthy", "unhealthy"): 

196 return {"plugin": name, "status": res["status"]} 

197 return {"plugin": name, "status": "unhealthy"} 

198 

199 result = asyncio.run(_run_health()) 

200 sys.modules.pop(mod_name, None) 

201 exit_code = 1 if result.get("status") == "unhealthy" else 0 

202 

203 if result.get("error"): 

204 emit_error_and_exit( 

205 result["error"], 

206 code=1, 

207 failure="health_error", 

208 command=command, 

209 fmt=fmt_lower, 

210 quiet=quiet, 

211 include_runtime=verbose, 

212 debug=debug, 

213 ) 

214 

215 def _build_payload(include: bool) -> Mapping[str, object]: 

216 """Constructs the final result payload. 

217 

218 Args: 

219 include (bool): If True, adds Python and platform info to the payload. 

220 

221 Returns: 

222 Mapping[str, object]: The payload containing the health check 

223 result and optional runtime metadata. 

224 """ 

225 payload = result 

226 if include: 

227 payload["python"] = ascii_safe(platform.python_version(), "python_version") 

228 payload["platform"] = ascii_safe(platform.platform(), "platform") 

229 return payload 

230 

231 new_run_command( 

232 command_name=command, 

233 payload_builder=_build_payload, 

234 quiet=quiet, 

235 verbose=verbose, 

236 fmt=fmt_lower, 

237 pretty=pretty, 

238 debug=debug, 

239 exit_code=exit_code, 

240 )