Coverage for /home/runner/work/bijux-cli/bijux-cli/src/bijux_cli/commands/status.py: 100%
71 statements
« prev ^ index » next coverage.py v7.10.4, created at 2025-08-19 23:36 +0000
1# SPDX-License-Identifier: MIT
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 * Verbose: Adds `{"python": str, "platform": str}` to the payload.
14 * Watch Mode Tick: `{"status": "ok", "ts": float, ...}`
15 * Watch Mode Stop: `{"status": "watch-stopped", ...}`
16 * Error: `{"error": str, "code": int}`
18Exit Codes:
19 * `0`: Success.
20 * `1`: Internal or fatal error during execution.
21 * `2`: Invalid argument (e.g., bad watch interval or format).
22 * `3`: ASCII encoding error.
23"""
25from __future__ import annotations
27from collections.abc import Mapping
28import platform
29import signal
30import sys
31import time
32from types import FrameType
34import typer
36from bijux_cli.commands.utilities import (
37 ascii_safe,
38 emit_error_and_exit,
39 new_run_command,
40 validate_common_flags,
41)
42from bijux_cli.contracts import EmitterProtocol, TelemetryProtocol
43from bijux_cli.core.constants import (
44 HELP_DEBUG,
45 HELP_FORMAT,
46 HELP_NO_PRETTY,
47 HELP_QUIET,
48 HELP_VERBOSE,
49)
50from bijux_cli.core.di import DIContainer
51from bijux_cli.core.enums import OutputFormat
53typer.core.rich = None # type: ignore[attr-defined,assignment]
55status_app = typer.Typer( # pytype: skip-file
56 name="status",
57 help="Show the CLI Status (Lean probe).",
58 rich_markup_mode=None,
59 context_settings={"help_option_names": ["-h", "--help"]},
60 no_args_is_help=False,
61)
64def _build_payload(include_runtime: bool) -> Mapping[str, object]:
65 """Constructs the status payload.
67 Args:
68 include_runtime (bool): If True, includes Python version and platform
69 information in the payload.
71 Returns:
72 Mapping[str, object]: A dictionary containing the status and optional
73 runtime details.
74 """
75 payload: dict[str, object] = {"status": "ok"}
76 if include_runtime:
77 payload["python"] = ascii_safe(platform.python_version(), "python_version")
78 payload["platform"] = ascii_safe(platform.platform(), "platform")
79 return payload
82def _run_watch_mode(
83 *,
84 command: str,
85 watch_interval: float,
86 fmt: str,
87 quiet: bool,
88 verbose: bool,
89 debug: bool,
90 effective_pretty: bool,
91 include_runtime: bool,
92 telemetry: TelemetryProtocol,
93 emitter: EmitterProtocol,
94) -> None:
95 """Emits CLI status in a continuous watch mode.
97 This function enters a loop, emitting a JSON-formatted status payload at
98 the specified interval. It handles graceful shutdown on SIGINT (Ctrl+C).
100 Args:
101 command (str): The command name for telemetry and error contracts.
102 watch_interval (float): The polling interval in seconds.
103 fmt (str): The output format, which must be "json" for streaming.
104 quiet (bool): If True, suppresses all output except errors.
105 verbose (bool): If True, includes verbose fields in the payload.
106 debug (bool): If True, enables diagnostic output to stderr.
107 effective_pretty (bool): If True, pretty-prints the output.
108 include_runtime (bool): If True, includes Python and platform fields.
109 telemetry (TelemetryProtocol): The telemetry sink for reporting events.
110 emitter (EmitterProtocol): The output emitter instance.
112 Returns:
113 None:
115 Raises:
116 SystemExit: On an invalid format or an unrecoverable error during
117 the watch loop.
118 """
119 if fmt != "json":
120 emit_error_and_exit(
121 "Only JSON output is supported in watch mode.",
122 code=2,
123 failure="watch_fmt",
124 command=command,
125 fmt=fmt,
126 quiet=quiet,
127 include_runtime=include_runtime,
128 )
130 stop = False
132 def _sigint_handler(_sig: int, _frame: FrameType | None) -> None:
133 """Handles SIGINT to allow for a graceful shutdown of the watch loop.
135 Args:
136 _sig (int): The signal number (unused).
137 _frame (FrameType | None): The current stack frame (unused).
138 """
139 nonlocal stop
140 stop = True
142 old_handler = signal.signal(signal.SIGINT, _sigint_handler)
143 try:
144 while not stop:
145 try:
146 payload = dict(_build_payload(include_runtime))
147 payload["ts"] = time.time()
148 if debug and not quiet:
149 print(
150 f"Debug: Emitting payload at ts={payload['ts']}",
151 file=sys.stderr,
152 )
153 if not quiet:
154 emitter.emit(
155 payload,
156 fmt=OutputFormat.JSON,
157 pretty=effective_pretty,
158 )
159 telemetry.event(
160 "COMMAND_SUCCESS",
161 {"command": command, "format": fmt, "mode": "watch"},
162 )
163 time.sleep(watch_interval)
164 except ValueError as exc:
165 emit_error_and_exit(
166 str(exc),
167 code=3,
168 failure="ascii",
169 command=command,
170 fmt=fmt,
171 quiet=quiet,
172 include_runtime=include_runtime,
173 )
174 except Exception as exc:
175 emit_error_and_exit(
176 f"Watch mode failed: {exc}",
177 code=1,
178 failure="emit",
179 command=command,
180 fmt=fmt,
181 quiet=quiet,
182 include_runtime=include_runtime,
183 )
184 finally:
185 signal.signal(signal.SIGINT, old_handler)
186 try:
187 stop_payload = dict(_build_payload(include_runtime))
188 stop_payload["status"] = "watch-stopped"
189 if debug and not quiet:
190 print("Debug: Emitting watch-stopped payload", file=sys.stderr)
191 if not quiet:
192 emitter.emit(
193 stop_payload,
194 fmt=OutputFormat.JSON,
195 pretty=effective_pretty,
196 level="info",
197 )
198 telemetry.event(
199 "COMMAND_STOPPED",
200 {"command": command, "format": fmt, "mode": "watch"},
201 )
202 except (ValueError, Exception):
203 _ = None
206@status_app.callback(invoke_without_command=True)
207def status(
208 ctx: typer.Context,
209 watch: float | None = typer.Option(None, "--watch", help="Poll every N seconds"),
210 quiet: bool = typer.Option(False, "-q", "--quiet", help=HELP_QUIET),
211 verbose: bool = typer.Option(False, "-v", "--verbose", help=HELP_VERBOSE),
212 fmt: str = typer.Option("json", "-f", "--format", help=HELP_FORMAT),
213 pretty: bool = typer.Option(True, "--pretty/--no-pretty", help=HELP_NO_PRETTY),
214 debug: bool = typer.Option(False, "-d", "--debug", help=HELP_DEBUG),
215) -> None:
216 """Defines the entrypoint and logic for the `bijux status` command.
218 This function orchestrates the status check. It validates flags and then
219 dispatches to either the single-run logic or the continuous watch mode
220 based on the presence of the `--watch` flag.
222 Args:
223 ctx (typer.Context): The Typer context for the CLI.
224 watch (float | None): If provided, enters watch mode, polling at this
225 interval in seconds. Must be a positive number.
226 quiet (bool): If True, suppresses all output except for errors.
227 verbose (bool): If True, includes Python and platform details in the
228 output payload.
229 fmt (str): The output format, either "json" or "yaml". Watch mode only
230 supports "json".
231 pretty (bool): If True, pretty-prints the output for human readability.
232 debug (bool): If True, enables debug diagnostics, implying `verbose`
233 and `pretty`.
235 Returns:
236 None:
238 Raises:
239 SystemExit: Exits with a contract-compliant status code and payload
240 upon any error, such as an invalid watch interval.
241 """
242 if ctx.invoked_subcommand:
243 return
245 emitter = DIContainer.current().resolve(EmitterProtocol)
246 telemetry = DIContainer.current().resolve(TelemetryProtocol)
247 command = "status"
249 fmt_lower = validate_common_flags(fmt, command, quiet)
251 if watch is not None:
252 try:
253 interval = float(watch)
254 if interval <= 0:
255 raise ValueError
256 except (ValueError, TypeError):
257 emit_error_and_exit(
258 "Invalid watch interval: must be > 0",
259 code=2,
260 failure="interval",
261 command=command,
262 fmt=fmt_lower,
263 quiet=quiet,
264 include_runtime=verbose,
265 debug=debug,
266 )
268 _run_watch_mode(
269 command=command,
270 watch_interval=interval,
271 fmt=fmt_lower,
272 quiet=quiet,
273 verbose=verbose,
274 debug=debug,
275 effective_pretty=pretty,
276 include_runtime=verbose,
277 telemetry=telemetry,
278 emitter=emitter,
279 )
280 else:
281 new_run_command(
282 command_name=command,
283 payload_builder=lambda include: _build_payload(include),
284 quiet=quiet,
285 verbose=verbose,
286 fmt=fmt_lower,
287 pretty=pretty,
288 debug=debug,
289 )