Coverage for /builds/svesch/cikit/src/cikit/command.py: 88%
149 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-10-09 14:17 +0000
1# SPDX-License-Identifier: GPL-3.0-or-later
2"""Command helpers for cikit.
4This module describes external commands and provides a small execution helper
5that can stream stdout and stderr live while keeping both streams separated
6for later inspection.
8Author:
9 Sven Schlender
10License:
11 SPDX-License-Identifier: GPL-3.0-or-later
12Date:
13 2026-02-22
14"""
16import logging
17import os
18import queue
19import signal
20import subprocess
21import threading
22import time
23from pathlib import Path
24from dataclasses import dataclass, field
26from cikit.errors import CiToolsError
29LOGGER = logging.getLogger(__name__)
30READER_JOIN_TIMEOUT = 1.0
31PROCESS_STOP_TIMEOUT = 5.0
34class CommandArgv(list):
35 """Command arguments with an additional masked representation.
37 The object behaves like the executable command argument list itself.
38 A second representation with masked secret values is available through
39 the ``masked`` attribute for logging and persisted reports.
41 Args:
42 exec_argv: Fully resolved command arguments used for execution.
43 masked_argv: Display-safe command arguments with secret values masked.
44 """
46 def __init__(self, exec_argv: list[str], masked_argv: list[str]):
47 """Initialize command arguments.
49 Args:
50 exec_argv: Fully resolved command arguments used for execution.
51 masked_argv: Display-safe command arguments with secret values masked.
52 """
53 super().__init__(exec_argv)
54 self.masked = masked_argv
56 @property
57 def exec(self) -> list[str]:
58 """Return executable command arguments.
60 Returns:
61 Executable command argument list.
62 """
63 return self
66class Command:
67 """Describe a command builder for one executable with optional subcommands.
69 This class stores metadata for one executable and optional subcommands.
70 It can build a plain argument list for direct execution. Environment
71 placeholders are intentionally not resolved here.
73 Argument strings may contain "${VAR}" placeholders. These placeholders are
74 resolved by the environment-aware execution layer.
76 Args:
77 name: Executable name, e.g. "ruff".
78 common_args: Arguments applied to every invocation unless disabled.
79 use_exec_tool: If True, CiEnvironment should prefix EXEC_TOOL.
80 """
81 def __init__(self, name: str, common_args: list = None, *, use_exec_tool: bool = False):
82 """Initialize a Command definition.
84 The constructor stores metadata for one executable and prepares
85 internal copies of common arguments to avoid accidental external
86 mutation. Subcommands can be registered with add_subcommand().
88 Args:
89 name: Executable name, e.g. "ruff".
90 common_args: Arguments applied to every invocation unless disabled.
91 use_exec_tool: If True, CiEnvironment should prefix EXEC_TOOL.
92 """
93 self.name = name
94 self.common_args = list(common_args or [])
95 self.use_exec_tool = use_exec_tool
96 self._subcommands = {}
98 def add_subcommand(self, name: str, args: list = None):
99 """Register a subcommand.
101 Args:
102 name: Subcommand name, e.g. "check".
103 args: Argument list for this subcommand (may be empty).
105 """
106 self._subcommands[name] = list(args or [])
108 def argv(self, subcommand: str = None, extra_args: list = None,
109 no_common: bool = False) -> list[str]:
110 """Build the argv list for this command invocation.
112 The resulting argv is suitable for subprocess execution (argv[0] is the
113 executable). Argument strings may contain "${VAR}" placeholders which are
114 expanded later by the execution layer (e.g. CiEnvironment.run()).
116 Args:
117 subcommand: Optional subcommand name registered via add_subcommand().
118 extra_args: Optional list of additional arguments appended at the end.
119 no_common: If True, do not include common_args.
121 Returns:
122 List of argv items (strings), starting with the executable name.
124 Raises:
125 CiToolsError: If the specified subcommand is unknown.
126 """
127 argv = [self.name]
129 if not no_common:
130 argv.extend([str(x) for x in self.common_args])
132 if subcommand:
133 if subcommand not in self._subcommands:
134 raise CiToolsError(f"Unknown subcommand for {self.name}: {subcommand}")
135 argv.extend([str(x) for x in self._subcommands[subcommand]])
137 if extra_args:
138 argv.extend([str(x) for x in extra_args])
140 return argv
143@dataclass
144class CommandOutput:
145 """Store one line of command output.
147 Args:
148 stream: Stream name, usually ``stdout`` or ``stderr``.
149 line: Output line without trailing newline.
150 """
152 stream: str
153 line: str
154 timestamp: float = field(default_factory=time.monotonic)
157@dataclass
158class CommandResult:
159 """Store the completed result of a command execution.
161 Args:
162 argv: Executed command argument list object.
163 returncode: Process return code.
164 output: Optional list of CommandOutput events.
165 """
167 argv: CommandArgv
168 returncode: int
169 output: list = field(default_factory=list)
171 def __post_init__(self):
172 """Build derived output helper strings.
174 The command output is stored as ordered CommandOutput events. This method
175 derives separated stdout and stderr strings once after initialization so
176 callers can inspect both streams directly.
177 """
178 self.stdout = self._stream_text("stdout")
179 self.stderr = self._stream_text("stderr")
181 def _stream_text(self, stream: str) -> str:
182 """Return all lines for one stream as text.
184 Args:
185 stream: Stream name to extract.
187 Returns:
188 Joined stream output.
189 """
190 return "\n".join(item.line for item in self.output if item.stream == stream)
192 def check_returncode(self):
193 """Raise CiToolsError if the command failed.
195 Raises:
196 CiToolsError: If the return code is non-zero.
197 """
198 if not self.returncode:
199 return
201 message = [
202 f"Command '{self.argv[0]}' failed with exit code {self.returncode}.",
203 f"Command: {' '.join(self.argv.masked)}",
204 ]
206 if self.stdout:
207 message.extend(["", "STDOUT:", self.stdout])
209 if self.stderr:
210 message.extend(["", "STDERR:", self.stderr])
212 raise CiToolsError("\n".join(message), exit_code=self.returncode)
214 def write_report(self, output_file: Path, mark_stderr: bool = True,
215 include_header: bool = True):
216 """Write captured command output to a text report.
218 Output events are written in their captured order so stdout and stderr
219 remain interleaved as they occurred during command execution. Standard
220 error lines can optionally be marked, and a header with the masked command
221 line and exit code can be included.
223 Args:
224 output_file: Target report file.
225 mark_stderr: Prefix standard error lines with ``[stderr]``.
226 include_header: Include masked command arguments and exit code before
227 the captured output.
229 Raises:
230 CiToolsError: If the report file cannot be written.
231 """
232 lines = []
234 if include_header:
235 lines.extend([f"Command: {' '.join(self.argv.masked)}", f"Exit code: {self.returncode}",
236 ""])
238 for item in self.output:
239 line = item.line
240 if mark_stderr and item.stream == "stderr":
241 line = f"[stderr] {line}"
242 lines.append(line)
244 try:
245 output_file.parent.mkdir(parents=True, exist_ok=True)
246 output_file.write_text("\n".join(lines) + "\n", encoding="utf-8")
247 except OSError as exp:
248 raise CiToolsError(f"Failed to write command report '{output_file}': {exp}") from exp
250 LOGGER.debug("Written %s", output_file)
253class CommandStreamReader(threading.Thread):
254 """Read one command output stream in a daemon thread.
256 The reader continuously consumes one subprocess output stream and forwards
257 received lines as CommandOutput events into a shared queue. If a logging
258 level is configured, lines are also logged immediately for live output.
260 Args:
261 stream: Text stream returned by subprocess.Popen.
262 stream_name: Logical stream name, e.g. ``stdout`` or ``stderr``.
263 output_queue: Queue receiving CommandOutput events.
264 log_level: Optional logging level for live output. If None, no
265 live logging is performed.
266 """
268 def __init__(self, stream, stream_name: str, output_queue: queue.Queue, log_level: int = None):
269 """Initialize a command stream reader.
271 The reader starts itself immediately as a daemon thread and forwards
272 received stream lines as CommandOutput events into the shared output
273 queue. Live logging is enabled if a log level is configured.
275 Args:
276 stream: Text stream returned by subprocess.Popen.
277 stream_name: Logical stream name, e.g. ``stdout`` or ``stderr``.
278 output_queue: Queue receiving CommandOutput events.
279 log_level: Optional logging level for live output. If None,
280 live logging is disabled.
281 """
282 super().__init__(name=f"CommandStreamReader-{stream_name}", daemon=True)
283 self.stream = stream
284 self.stream_name = stream_name
285 self.output_queue = output_queue
286 self.log_level = log_level
287 self.start()
289 def run(self):
290 """Consume the assigned subprocess stream.
292 The method continuously reads lines from the configured stream until
293 the stream reaches EOF. Every received line is converted into a
294 CommandOutput event and forwarded into the shared queue.
296 Live logging is performed immediately if a log level is configured.
297 The stream is always closed before the thread terminates.
298 """
299 try:
300 for line in iter(self.stream.readline, ""):
301 item = CommandOutput(self.stream_name, line.rstrip("\n"))
302 self.output_queue.put(item)
304 if self.log_level is not None:
305 LOGGER.log(self.log_level, "%s", item.line)
306 finally:
307 self.stream.close()
310def _terminate_process(process):
311 """Terminate a process and kill it if required.
313 The process is first asked to terminate gracefully. On POSIX systems the
314 signal is sent to the whole process group, because run_command() starts the
315 child in a new session. If the process does not stop in time, it is killed.
317 Args:
318 process: Active subprocess instance.
319 """
320 try:
321 if os.name == "nt": 321 ↛ 322line 321 didn't jump to line 322 because the condition on line 321 was never true
322 process.terminate()
323 else:
324 os.killpg(process.pid, signal.SIGTERM)
325 except ProcessLookupError:
326 return
328 try:
329 process.wait(timeout=PROCESS_STOP_TIMEOUT)
330 return
331 except subprocess.TimeoutExpired:
332 pass
334 try:
335 if os.name == "nt":
336 process.kill()
337 else:
338 os.killpg(process.pid, signal.SIGKILL)
339 except ProcessLookupError:
340 return
342 try:
343 process.wait(timeout=PROCESS_STOP_TIMEOUT)
344 except subprocess.TimeoutExpired:
345 LOGGER.warning("Process did not terminate after SIGKILL: %s", process.pid)
348def _drain_queue(output_queue: queue.Queue) -> list:
349 """Return all currently queued output events.
351 The stream reader threads push CommandOutput events into the shared queue.
352 This helper drains the queue after the process has finished and the readers
353 had a chance to stop, preserving queue order for the result object.
355 Args:
356 output_queue: Queue containing CommandOutput events.
358 Returns:
359 List of collected CommandOutput events.
360 """
361 output = []
362 while True:
363 try:
364 output.append(output_queue.get_nowait())
365 except queue.Empty:
366 break
367 return output
370def run_command(argv: list, *, cwd: Path = None, timeout: int = None,
371 check_exit_code: bool = True, live_output: bool = True) -> CommandResult:
372 """Run a command and return its completed result.
374 The function starts the process, reads stdout and stderr in separate
375 reader threads and stores all output as CommandOutput events. On POSIX
376 systems the command is started in a new session so that interrupts can
377 terminate the whole process group.
379 Args:
380 argv: Command argument list.
381 cwd: Optional working directory.
382 timeout: Optional timeout in seconds.
383 check_exit_code: If True, non-zero exit codes raise CiToolsError.
384 live_output: If True, log process output while running.
386 Returns:
387 Completed command result.
389 Raises:
390 CiToolsError: If the process cannot be started, times out, is
391 interrupted, or returns a failing exit code.
392 """
393 if not isinstance(argv, CommandArgv):
394 argv = CommandArgv(argv, argv)
396 LOGGER.debug("Running command <%s> (cwd=%s)", " ".join(argv), cwd or os.getcwd())
398 try:
399 with subprocess.Popen(argv, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
400 stdin=subprocess.DEVNULL, text=True, cwd=cwd,
401 start_new_session=True) as process:
402 output_queue = queue.Queue()
403 stdout = CommandStreamReader(process.stdout, "stdout", output_queue,
404 logging.DEBUG if live_output else None)
405 stderr = CommandStreamReader(process.stderr, "stderr", output_queue,
406 logging.WARNING if live_output else None)
408 try:
409 process.wait(timeout=timeout)
410 except subprocess.TimeoutExpired as exp:
411 _terminate_process(process)
412 raise CiToolsError(f"Command '{argv[0]}' timed out.") from exp
413 except KeyboardInterrupt as exp:
414 _terminate_process(process)
415 raise CiToolsError(f"Command '{argv[0]}' interrupted.") from exp
417 stdout.join(timeout=READER_JOIN_TIMEOUT)
418 stderr.join(timeout=READER_JOIN_TIMEOUT)
420 result = CommandResult(argv, process.returncode, _drain_queue(output_queue))
422 except OSError as exp:
423 raise CiToolsError(f"Failed to start command '{argv[0]}': {exp}") from exp
425 if check_exit_code:
426 result.check_returncode()
428 return result