Coverage for / home / runner / work / bijux-cli / bijux-cli / src / bijux_cli / cli / core / command.py: 97%
132 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"""Command helpers for policy-aware execution and exit intents."""
6from __future__ import annotations
8import os
9from pathlib import Path
10import re
11import sys
12from typing import Any, NoReturn
14from bijux_cli.cli.core.constants import ENV_CONFIG, ENV_PREFIX
15from bijux_cli.core.enums import ErrorType, ExitCode, LogLevel, OutputFormat
16from bijux_cli.core.exit_policy import ExitIntent, ExitIntentError
17from bijux_cli.core.precedence import current_execution_policy, resolve_exit_intent
18from bijux_cli.infra.contracts import Emitter, Serializer
20_ALLOWED_CTRL = {"\n", "\r", "\t"}
21_ENV_LINE_RX = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*=[A-Za-z0-9_./\\-]*$")
24def record_history(command: str, exit_code: int) -> None:
25 """Record a history entry, ignoring failures."""
26 if command == "history":
27 return
28 try:
29 from bijux_cli.core.di import DIContainer
30 from bijux_cli.services.history.contracts import HistoryProtocol
32 hist = DIContainer.current().resolve(HistoryProtocol)
33 hist.add(
34 command=command,
35 params=[],
36 success=(exit_code == 0),
37 return_code=exit_code,
38 duration_ms=0.0,
39 )
40 except PermissionError as exc:
41 print(f"Permission denied writing history: {exc}", file=sys.stderr)
42 except OSError as exc:
43 import errno as _errno
45 if exc.errno in (_errno.EACCES, _errno.EPERM):
46 print(f"Permission denied writing history: {exc}", file=sys.stderr)
47 elif exc.errno in (_errno.ENOSPC, _errno.EDQUOT):
48 print(
49 f"No space left on device while writing history: {exc}",
50 file=sys.stderr,
51 )
52 else:
53 print(f"Error writing history: {exc}", file=sys.stderr)
54 except Exception as exc:
55 print(f"Error writing history: {exc}", file=sys.stderr)
58def new_run_command(
59 command_name: str,
60 payload_builder: Any,
61 quiet: bool,
62 fmt: OutputFormat,
63 pretty: bool,
64 log_level: str,
65 exit_code: int = 0,
66) -> NoReturn:
67 """Build a payload and raise an ExitIntentError with resolved behavior."""
68 from bijux_cli.core.di import DIContainer
69 from bijux_cli.infra.contracts import Emitter
70 from bijux_cli.services.contracts import TelemetryProtocol
72 _ = (quiet, fmt, pretty, log_level)
73 DIContainer.current().resolve(Emitter)
74 DIContainer.current().resolve(TelemetryProtocol)
76 resolved = current_execution_policy()
77 include_runtime = resolved.include_runtime
78 output_format = validate_common_flags(
79 fmt,
80 command_name,
81 resolved.quiet,
82 include_runtime=include_runtime,
83 log_level=resolved.log_level,
84 )
85 effective_pretty = resolved.pretty
86 try:
87 payload = payload_builder(include_runtime)
88 except ValueError as exc:
89 intent = resolve_exit_intent(
90 message=str(exc),
91 code=2,
92 failure="ascii",
93 command=command_name,
94 fmt=output_format,
95 quiet=resolved.quiet,
96 include_runtime=include_runtime,
97 error_type=ErrorType.ASCII,
98 log_level=resolved.log_level,
99 )
100 raise ExitIntentError(intent) from exc
102 record_history(command_name, exit_code)
104 if resolved.quiet:
105 intent = ExitIntent(
106 code=ExitCode(exit_code),
107 stream=None,
108 payload=None,
109 fmt=output_format,
110 pretty=effective_pretty,
111 show_traceback=False,
112 )
113 raise ExitIntentError(intent)
115 intent = ExitIntent(
116 code=ExitCode(exit_code),
117 stream="stdout",
118 payload=payload,
119 fmt=output_format,
120 pretty=effective_pretty,
121 show_traceback=False,
122 )
123 raise ExitIntentError(intent)
126def raise_exit_intent(*args: Any, **kwargs: Any) -> NoReturn:
127 """Raise an ExitIntentError from resolved error intent."""
128 if args: 128 ↛ 132line 128 didn't jump to line 132 because the condition on line 128 was always true
129 if len(args) != 1:
130 raise TypeError("raise_exit_intent accepts at most one positional arg")
131 kwargs["message"] = args[0]
132 raise ExitIntentError(resolve_exit_intent(**kwargs))
135def resolve_serializer() -> Serializer:
136 """Resolve the serializer adapter."""
137 from bijux_cli.core.di import DIContainer
139 serializer = DIContainer.current().resolve(Serializer)
140 if not hasattr(serializer, "dumps"):
141 raise RuntimeError("Serializer does not implement dumps()")
142 return serializer
145def resolve_emitter() -> Emitter | None:
146 """Resolve the emitter adapter or return None."""
147 from bijux_cli.core.di import DIContainer
149 try:
150 return DIContainer.current().resolve(Emitter)
151 except Exception:
152 return None
155def emit_payload(
156 payload: object,
157 *,
158 serializer: Serializer,
159 emitter: Emitter | None,
160 fmt: OutputFormat,
161 pretty: bool,
162 stream: str,
163) -> None:
164 """Emit a payload to the requested stream."""
165 out = sys.stdout if stream == "stdout" else sys.stderr
166 _ = emitter
167 output = serializer.dumps(payload, fmt=fmt, pretty=pretty).rstrip("\n")
168 print(output, file=out, flush=True)
171def ascii_safe(text: Any, _field: str = "") -> str:
172 """Return a printable ASCII-only string."""
173 text_str = text if isinstance(text, str) else str(text)
174 return "".join(
175 ch if (32 <= ord(ch) <= 126) or ch in _ALLOWED_CTRL else "?" for ch in text_str
176 )
179def normalize_format(fmt: str | OutputFormat | None) -> OutputFormat | None:
180 """Normalize a format value into OutputFormat."""
181 if isinstance(fmt, OutputFormat):
182 return fmt
183 if isinstance(fmt, str):
184 value = fmt.strip().lower()
185 if value in ("json", "yaml"):
186 return OutputFormat(value)
187 return None
190def contains_non_ascii_env() -> bool:
191 """Return True when config env or file contents are non-ASCII."""
192 config_path_str = os.environ.get(ENV_CONFIG)
193 if config_path_str:
194 if not config_path_str.isascii():
195 return True
196 try:
197 config_path = Path(config_path_str)
198 except NotImplementedError:
199 return False
200 if config_path.exists(): 200 ↛ 208line 200 didn't jump to line 208 because the condition on line 200 was always true
201 try:
202 config_path.read_text(encoding="ascii")
203 except UnicodeDecodeError:
204 return True
205 except (IsADirectoryError, PermissionError, FileNotFoundError, OSError):
206 pass
208 for k, v in os.environ.items():
209 if k.startswith(ENV_PREFIX) and not v.isascii():
210 return True
211 return False
214def validate_common_flags(
215 fmt: str | OutputFormat,
216 command: str,
217 quiet: bool,
218 include_runtime: bool = False,
219 log_level: LogLevel = LogLevel.INFO,
220) -> OutputFormat:
221 """Validate output format and ASCII environment."""
222 format_value = normalize_format(fmt)
223 if format_value is None:
224 intent = resolve_exit_intent(
225 message=f"Unsupported format: {fmt}",
226 code=ExitCode.USAGE,
227 failure="format",
228 command=command,
229 fmt=OutputFormat.JSON,
230 quiet=quiet,
231 include_runtime=include_runtime,
232 error_type=ErrorType.USAGE,
233 log_level=log_level,
234 )
235 raise ExitIntentError(intent)
236 if format_value not in (OutputFormat.JSON, OutputFormat.YAML): 236 ↛ 237line 236 didn't jump to line 237 because the condition on line 236 was never true
237 intent = resolve_exit_intent(
238 message="Invalid output format.",
239 code=ExitCode.USAGE,
240 failure="format",
241 command=command,
242 fmt=format_value,
243 quiet=quiet,
244 include_runtime=include_runtime,
245 error_type=ErrorType.USAGE,
246 log_level=log_level,
247 )
248 raise ExitIntentError(intent)
250 if contains_non_ascii_env():
251 intent = resolve_exit_intent(
252 message="Non-ASCII in configuration or environment",
253 code=ExitCode.ASCII,
254 failure="ascii",
255 command=command,
256 fmt=format_value,
257 quiet=quiet,
258 include_runtime=include_runtime,
259 error_type=ErrorType.ASCII,
260 log_level=log_level,
261 )
262 raise ExitIntentError(intent)
264 return format_value
267def validate_env_file_if_present(path_str: str) -> None:
268 """Validate env file format if present."""
269 if not path_str or not Path(path_str).exists():
270 return
271 try:
272 text = Path(path_str).read_text(encoding="utf-8", errors="strict")
273 except Exception as exc:
274 raise ValueError(f"Cannot read config file: {exc}") from exc
276 for i, line in enumerate(text.splitlines(), start=1):
277 s = line.strip()
278 if s and not s.startswith("#") and not _ENV_LINE_RX.match(s):
279 raise ValueError(f"Malformed line {i} in config: {line!r}")
282__all__ = [
283 "ascii_safe",
284 "contains_non_ascii_env",
285 "emit_payload",
286 "new_run_command",
287 "normalize_format",
288 "record_history",
289 "resolve_emitter",
290 "resolve_serializer",
291 "raise_exit_intent",
292 "validate_common_flags",
293 "validate_env_file_if_present",
294]