Coverage for / home / runner / work / bijux-cli / bijux-cli / src / bijux_cli / cli / commands / status.py: 97%
93 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 `status` command for the Bijux CLI.
6This module provides a lightweight "liveness probe" for the CLI, designed for
7health checks and monitoring. In its default mode, it performs a quick check
8and returns a simple "ok" status. It also supports a continuous "watch" mode
9that emits status updates at a regular interval.
11Output Contract:
12 * Success: `{"status": "ok"}`
13 * Watch Mode Tick: `{"status": "ok", "ts": float, ...}`
14 * Watch Mode Stop: `{"status": "watch-stopped", ...}`
15 * Error: `{"error": str, "code": int}`
17Exit Codes:
18 * `0`: Success.
19 * `1`: Internal or fatal error during execution.
20 * `2`: Invalid argument (e.g., bad watch interval or format).
21 * `3`: ASCII encoding error.
22"""
24from __future__ import annotations
26import platform
27import signal
28import threading
29import time
30from types import FrameType
31from typing import Any
33import typer
35from bijux_cli.cli.core.command import (
36 ascii_safe,
37 new_run_command,
38 validate_common_flags,
39)
40from bijux_cli.cli.core.constants import (
41 OPT_FORMAT,
42 OPT_LOG_LEVEL,
43 OPT_PRETTY,
44 OPT_QUIET,
45)
46from bijux_cli.cli.core.help_text import (
47 HELP_FORMAT,
48 HELP_LOG_LEVEL,
49 HELP_NO_PRETTY,
50 HELP_QUIET,
51)
52from bijux_cli.core.di import DIContainer
53from bijux_cli.core.enums import ErrorType, LogLevel, OutputFormat
54from bijux_cli.core.exit_policy import ExitIntentError
55from bijux_cli.core.precedence import current_execution_policy, resolve_exit_intent
56from bijux_cli.core.runtime import AsyncTyper
57from bijux_cli.infra.contracts import Emitter
58from bijux_cli.services.contracts import TelemetryProtocol
60typer.core.rich = None # type: ignore[attr-defined]
62status_app = AsyncTyper(
63 name="status",
64 help="Show the CLI Status (Lean probe).",
65 rich_markup_mode=None,
66 context_settings={"help_option_names": ["-h", "--help"]},
67 no_args_is_help=False,
68)
71def _build_payload(include_runtime: bool) -> dict[str, object]:
72 """Constructs the status payload.
74 Args:
75 include_runtime (bool): If True, includes Python version and platform
76 information in the payload.
78 Returns:
79 Mapping[str, object]: A dictionary containing the status and optional
80 runtime details.
81 """
82 payload: dict[str, object] = {"status": "ok"}
83 if include_runtime:
84 payload.update(
85 {
86 "python": ascii_safe(platform.python_version(), "python_version"),
87 "platform": ascii_safe(platform.platform(), "platform"),
88 }
89 )
90 return payload
93def _run_watch_mode(
94 *,
95 command: str,
96 watch_interval: float,
97 fmt: OutputFormat,
98 quiet: bool,
99 effective_pretty: bool,
100 include_runtime: bool,
101 log_policy: Any,
102 telemetry: TelemetryProtocol,
103 emitter: Emitter,
104) -> None:
105 """Emits CLI status in a continuous watch mode.
107 This function enters a loop, emitting a JSON-formatted status payload at
108 the specified interval. It handles graceful shutdown on SIGINT (Ctrl+C).
110 Args:
111 command (str): The command name for telemetry and error contracts.
112 watch_interval (float): The polling interval in seconds.
113 fmt (str): The output format, which must be "json" for streaming.
114 quiet (bool): If True, suppresses all output except errors.
115 effective_pretty (bool): If True, pretty-prints the output.
116 include_runtime (bool): If True, includes Python and platform fields.
117 log_level (LogLevel): Logging level for diagnostics.
118 telemetry (TelemetryProtocol): The telemetry sink for reporting events.
119 emitter (Emitter): The output emitter instance.
121 Returns:
122 None:
124 Raises:
125 SystemExit: On an invalid format or an unrecoverable error during
126 the watch loop.
127 """
128 format_value = fmt
129 if format_value is not OutputFormat.JSON:
130 intent = resolve_exit_intent(
131 message="Only JSON output is supported in watch mode.",
132 code=2,
133 failure="watch_fmt",
134 command=command,
135 fmt=format_value,
136 quiet=quiet,
137 include_runtime=include_runtime,
138 error_type=ErrorType.USER_INPUT,
139 log_level=log_policy.level,
140 )
141 raise ExitIntentError(intent)
143 stop = False
144 emit_output = not quiet
145 emit_diagnostics = log_policy.show_internal and emit_output
147 def _sigint_handler(_sig: int, _frame: FrameType | None) -> None:
148 """Handles SIGINT to allow for a graceful shutdown of the watch loop.
150 Args:
151 _sig (int): The signal number (unused).
152 _frame (FrameType | None): The current stack frame (unused).
153 """
154 nonlocal stop
155 stop = True
157 old_handler = None
158 if threading.current_thread() is threading.main_thread(): 158 ↛ 163line 158 didn't jump to line 163 because the condition on line 158 was always true
159 try:
160 old_handler = signal.signal(signal.SIGINT, _sigint_handler)
161 except ValueError:
162 old_handler = None
163 try:
164 while not stop:
165 try:
166 payload = _build_payload(include_runtime)
167 payload["ts"] = time.time()
168 if emit_diagnostics:
169 emitter.emit(
170 payload,
171 fmt=OutputFormat.JSON,
172 pretty=effective_pretty,
173 level=LogLevel.DEBUG,
174 message=f"Debug: Emitting payload at ts={payload['ts']}",
175 output=None,
176 emit_output=False,
177 emit_diagnostics=True,
178 )
179 if emit_output:
180 emitter.emit(
181 payload,
182 fmt=OutputFormat.JSON,
183 pretty=effective_pretty,
184 level=LogLevel.INFO,
185 message="Status update",
186 output=None,
187 emit_output=emit_output,
188 emit_diagnostics=emit_diagnostics,
189 )
190 telemetry.event(
191 "COMMAND_SUCCESS",
192 {"command": command, "format": fmt.value, "mode": "watch"},
193 )
194 time.sleep(watch_interval)
195 except ValueError as exc:
196 intent = resolve_exit_intent(
197 message=str(exc),
198 code=3,
199 failure="ascii",
200 command=command,
201 fmt=fmt,
202 quiet=quiet,
203 include_runtime=include_runtime,
204 error_type=ErrorType.ASCII,
205 log_level=log_policy.level,
206 )
207 raise ExitIntentError(intent) from exc
208 except Exception as exc:
209 intent = resolve_exit_intent(
210 message=f"Watch mode failed: {exc}",
211 code=1,
212 failure="emit",
213 command=command,
214 fmt=fmt,
215 quiet=quiet,
216 include_runtime=include_runtime,
217 error_type=ErrorType.INTERNAL,
218 log_level=log_policy.level,
219 )
220 raise ExitIntentError(intent) from exc
221 finally:
222 if old_handler is not None: 222 ↛ 224line 222 didn't jump to line 224 because the condition on line 222 was always true
223 signal.signal(signal.SIGINT, old_handler)
224 try:
225 stop_payload = _build_payload(include_runtime)
226 stop_payload["status"] = "watch-stopped"
227 if emit_diagnostics:
228 emitter.emit(
229 stop_payload,
230 fmt=OutputFormat.JSON,
231 pretty=effective_pretty,
232 level=LogLevel.DEBUG,
233 message="Debug: Emitting watch-stopped payload",
234 output=None,
235 emit_output=False,
236 emit_diagnostics=True,
237 )
238 if emit_output:
239 emitter.emit(
240 stop_payload,
241 fmt=OutputFormat.JSON,
242 pretty=effective_pretty,
243 level=LogLevel.INFO,
244 message="Status watch stopped",
245 output=None,
246 emit_output=emit_output,
247 emit_diagnostics=emit_diagnostics,
248 )
249 telemetry.event(
250 "COMMAND_STOPPED",
251 {"command": command, "format": fmt.value, "mode": "watch"},
252 )
253 except (ValueError, Exception):
254 _ = None
257@status_app.callback(invoke_without_command=True)
258def status(
259 ctx: typer.Context,
260 watch: float | None = typer.Option(None, "--watch", help="Poll every N seconds"),
261 quiet: bool = typer.Option(False, *OPT_QUIET, help=HELP_QUIET),
262 fmt: str = typer.Option("json", *OPT_FORMAT, help=HELP_FORMAT),
263 pretty: bool = typer.Option(True, OPT_PRETTY, help=HELP_NO_PRETTY),
264 log_level: str = typer.Option("info", *OPT_LOG_LEVEL, help=HELP_LOG_LEVEL),
265) -> None:
266 """Defines the entrypoint and logic for the `bijux status` command.
268 This function orchestrates the status check. It validates flags and then
269 dispatches to either the single-run logic or the continuous watch mode
270 based on the presence of the `--watch` flag.
272 Args:
273 ctx (typer.Context): The Typer context for the CLI.
274 watch (float | None): If provided, enters watch mode, polling at this
275 interval in seconds. Must be a positive number.
276 quiet (bool): If True, suppresses all output except for errors.
277 output payload.
278 fmt (str): The output format, either "json" or "yaml". Watch mode only
279 supports "json".
280 pretty (bool): If True, pretty-prints the output for human readability.
281 log_level (str): The resolved log level name.
283 Returns:
284 None:
286 Raises:
287 SystemExit: Exits with a contract-compliant status code and payload
288 upon any error, such as an invalid watch interval.
289 """
290 if ctx.invoked_subcommand:
291 return
293 emitter = DIContainer.current().resolve(Emitter)
294 telemetry = DIContainer.current().resolve(TelemetryProtocol)
295 command = "status"
297 effective = current_execution_policy()
298 fmt_lower = validate_common_flags(
299 fmt,
300 command,
301 effective.quiet,
302 include_runtime=effective.include_runtime,
303 log_level=effective.log_level,
304 )
305 quiet = effective.quiet
306 log_policy = effective.log_policy
307 log_level_value = effective.log_level
308 pretty = effective.pretty
310 if watch is not None:
311 try:
312 interval = float(watch)
313 if interval <= 0:
314 raise ValueError
315 except (ValueError, TypeError):
316 intent = resolve_exit_intent(
317 message="Invalid watch interval: must be > 0",
318 code=2,
319 failure="interval",
320 command=command,
321 fmt=fmt_lower,
322 quiet=quiet,
323 include_runtime=effective.include_runtime,
324 error_type=ErrorType.USER_INPUT,
325 log_level=log_level_value,
326 )
327 raise ExitIntentError(intent) from None
329 _run_watch_mode(
330 command=command,
331 watch_interval=interval,
332 fmt=fmt_lower,
333 quiet=quiet,
334 effective_pretty=pretty,
335 include_runtime=effective.include_runtime,
336 log_policy=log_policy,
337 telemetry=telemetry,
338 emitter=emitter,
339 )
340 else:
341 new_run_command(
342 command_name=command,
343 payload_builder=lambda include: _build_payload(include),
344 quiet=quiet,
345 fmt=fmt_lower,
346 pretty=pretty,
347 log_level=log_level_value,
348 )