Coverage for / home / runner / work / bijux-cli / bijux-cli / src / bijux_cli / cli / plugins / commands / check.py: 98%
80 statements
« prev ^ index » next coverage.py v7.13.2, created at 2026-01-26 17:59 +0000
1# SPDX-License-Identifier: Apache-2.0
2# Copyright © 2025 Bijan Mousavi
4"""Implements the `plugins check` subcommand for the Bijux CLI.
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.
11Output Contract:
12 * Healthy: `{"plugin": str, "status": "healthy"}`
13 * Unhealthy: `{"plugin": str, "status": "unhealthy"}` (exits with code 1)
14 * Error: `{"error": "...", "code": int}` (for pre-check failures)
16Exit Codes:
17 * `0`: The plugin is healthy.
18 * `1`: The plugin is unhealthy, could not be found, or an error occurred
19 during import or execution.
20 * `2`: An invalid flag was provided (e.g., bad format).
21 * `3`: An ASCII or encoding error was detected in the environment.
22"""
24from __future__ import annotations
26import asyncio
27from collections.abc import Mapping
28import importlib.util
29import inspect
30import platform
31import sys
32import types
33from typing import Any
35import anyio
36import typer
38from bijux_cli.cli.core.command import (
39 ascii_safe,
40 new_run_command,
41 raise_exit_intent,
42 validate_common_flags,
43)
44from bijux_cli.cli.core.constants import (
45 OPT_FORMAT,
46 OPT_LOG_LEVEL,
47 OPT_PRETTY,
48 OPT_QUIET,
49)
50from bijux_cli.cli.core.help_text import (
51 HELP_FORMAT,
52 HELP_LOG_LEVEL,
53 HELP_NO_PRETTY,
54 HELP_QUIET,
55)
56from bijux_cli.core.precedence import current_execution_policy
57from bijux_cli.plugins.metadata import get_plugin_metadata
60async def check_plugin(
61 name: str = typer.Argument(..., help="Plugin name"),
62 quiet: bool = typer.Option(False, *OPT_QUIET, help=HELP_QUIET),
63 fmt: str = typer.Option("json", *OPT_FORMAT, help=HELP_FORMAT),
64 pretty: bool = typer.Option(True, OPT_PRETTY, help=HELP_NO_PRETTY),
65 log_level: str = typer.Option("info", *OPT_LOG_LEVEL, help=HELP_LOG_LEVEL),
66) -> None:
67 """Runs a health check on a specific installed plugin.
69 This function validates a plugin's structure, dynamically imports its
70 `plugin.py` file, and executes its `health()` hook to determine its
71 operational status. The final status is emitted as a structured payload.
73 Args:
74 name (str): The name of the plugin to check.
75 quiet (bool): If True, suppresses all output except for errors.
76 fmt (str): The output format, "json" or "yaml".
77 pretty (bool): If True, pretty-prints the output.
78 log_level (str): Logging level for diagnostics.
80 Returns:
81 None:
83 Raises:
84 SystemExit: Always exits with a contract-compliant status code and
85 payload, indicating the health status or detailing an error.
86 """
87 command = "plugins check"
89 policy = current_execution_policy()
90 quiet = policy.quiet
91 include_runtime = policy.include_runtime
92 log_level_value = policy.log_level
93 pretty = policy.pretty
94 fmt_lower = validate_common_flags(
95 fmt,
96 command,
97 quiet,
98 include_runtime=include_runtime,
99 log_level=log_level_value,
100 )
102 try:
103 meta = await anyio.to_thread.run_sync(get_plugin_metadata, name)
104 except Exception as exc:
105 raise_exit_intent(
106 str(exc),
107 code=1,
108 failure="metadata_error",
109 command=command,
110 fmt=fmt_lower,
111 quiet=quiet,
112 include_runtime=include_runtime,
113 log_level=log_level_value,
114 )
116 if not meta.path: 116 ↛ 117line 116 didn't jump to line 117 because the condition on line 116 was never true
117 raise_exit_intent(
118 f'Plugin "{name}" has no local health hook',
119 code=1,
120 failure="health_unavailable",
121 command=command,
122 fmt=fmt_lower,
123 quiet=quiet,
124 include_runtime=include_runtime,
125 log_level=log_level_value,
126 )
128 plug_dir = meta.path
129 plug_py = plug_dir / "plugin.py"
130 if not plug_py.is_file():
131 raise_exit_intent(
132 f'Plugin "{name}" not found',
133 code=1,
134 failure="not_found",
135 command=command,
136 fmt=fmt_lower,
137 quiet=quiet,
138 include_runtime=include_runtime,
139 log_level=log_level_value,
140 extra={"plugin": name},
141 )
143 mod_name = f"_bijux_cli_plugin_{name}"
144 try:
145 spec = importlib.util.spec_from_file_location(mod_name, plug_py)
146 if not spec or not spec.loader:
147 raise ImportError("Cannot create import spec")
148 module = types.ModuleType(mod_name)
149 sys.modules[mod_name] = module
150 spec.loader.exec_module(module)
151 except Exception as exc:
152 err = f"Import error: {exc}"
153 raise_exit_intent(
154 err,
155 code=1,
156 failure="import_error",
157 command=command,
158 fmt=fmt_lower,
159 quiet=quiet,
160 include_runtime=include_runtime,
161 log_level=log_level_value,
162 )
164 async def _run_health() -> dict[str, Any]:
165 """Isolates and executes the plugin's `health()` hook.
167 This function finds and calls the `health()` function within the
168 imported plugin module. It handles both synchronous and asynchronous
169 hooks, validates their signatures, and safely captures any exceptions
170 during execution.
172 Returns:
173 dict[str, Any]: A dictionary containing the health check result,
174 which includes the plugin name and a status ('healthy' or
175 'unhealthy'), or an error message.
176 """
177 hook = getattr(module, "health", None)
178 if not callable(hook):
179 return {"plugin": name, "error": "No health() hook"}
180 try:
181 sig = inspect.signature(hook)
182 if len(sig.parameters) != 1:
183 return {
184 "plugin": name,
185 "error": "health() hook must take exactly one argument (di)",
186 }
187 except Exception as exc1:
188 return {"plugin": name, "error": f"health() signature error: {exc1}"}
189 try:
190 if asyncio.iscoroutinefunction(hook):
191 res = await hook(None)
192 else:
193 res = await asyncio.to_thread(hook, None)
194 except BaseException as exc2:
195 return {"plugin": name, "error": str(exc2) or exc2.__class__.__name__}
197 if res is True:
198 return {"plugin": name, "status": "healthy"}
199 if res is False:
200 return {"plugin": name, "status": "unhealthy"}
201 if isinstance(res, dict) and res.get("status") in ("healthy", "unhealthy"):
202 return {"plugin": name, "status": res["status"]}
203 return {"plugin": name, "status": "unhealthy"}
205 result = await _run_health()
206 sys.modules.pop(mod_name, None)
207 exit_code = 1 if result.get("status") == "unhealthy" else 0
209 if result.get("error"):
210 raise_exit_intent(
211 result["error"],
212 code=1,
213 failure="health_error",
214 command=command,
215 fmt=fmt_lower,
216 quiet=quiet,
217 include_runtime=include_runtime,
218 log_level=log_level_value,
219 )
221 def _build_payload(include: bool) -> Mapping[str, object]:
222 """Constructs the final result payload.
224 Args:
225 include (bool): If True, adds Python and platform info to the payload.
227 Returns:
228 Mapping[str, object]: The payload containing the health check
229 result and optional runtime metadata.
230 """
231 payload = result
232 if include:
233 payload["python"] = ascii_safe(platform.python_version(), "python_version")
234 payload["platform"] = ascii_safe(platform.platform(), "platform")
235 return payload
237 new_run_command(
238 command_name=command,
239 payload_builder=_build_payload,
240 quiet=quiet,
241 fmt=fmt_lower,
242 pretty=pretty,
243 log_level=log_level_value,
244 exit_code=exit_code,
245 )