Coverage for src / mafw / tools / shell_tools.py: 100%

1# Copyright 2025–2026 European Union 

2# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu) 

3# SPDX-License-Identifier: EUPL-1.2 

4""" 

5Shell execution utilities. 

6 

7This module provides unified functions for executing external commands via subprocess, 

8with support for dry-runs, environment merging, and standardized output. 

9""" 

10 

11import os 

12import shlex 

13import subprocess 

14from pathlib import Path 

15from typing import Any 

16 

17from rich.console import Console 

18 

19CONSOLE = Console() 

20"""Shared console for standardized output.""" 

21 

22 

23def _to_command_line(command: str | list[str]) -> str: 

24 """ 

25 Convert a command into a printable shell-like string. 

26 

27 :param command: Command expressed as a string or tokenized list. 

28 :type command: str | list[str] 

29 :return: Human-readable command line. 

30 :rtype: str 

31 """ 

32 if isinstance(command, str): 

33 return command 

34 return shlex.join(command) 

35 

36 

37def run( 

38 command: str | list[str], 

39 *, 

40 cwd: Path | str | None = None, 

41 dry_run: bool = False, 

42 env: dict[str, str] | None = None, 

43 quiet: bool = False, 

44 **kwargs: Any, 

45) -> subprocess.CompletedProcess[Any]: 

46 """ 

47 Execute a subprocess command and return a completed process object. 

48 

49 The command line is always printed with a standard prefix. In dry-run mode, 

50 execution is skipped and a successful synthetic ``CompletedProcess`` is returned. 

51 

52 .. versionadded:: 2.2 

53 Add the `quiet` parameter to do not display the command being executed 

54 

55 :param command: Command to execute. 

56 :type command: str | list[str] 

57 :param cwd: Working directory for command execution. 

58 :type cwd: Path | str | None 

59 :param dry_run: If ``True``, print the command without executing it. 

60 :type dry_run: bool 

61 :param env: Optional environment variables to merge with the current environment. 

62 :type env: dict[str, str] | None 

63 :param quiet: Suppress the standard console banner when ``True``. 

64 :type quiet: bool 

65 :param kwargs: Additional keyword arguments forwarded to ``subprocess.run``. 

66 :type kwargs: Any 

67 :return: Completed process produced by the command execution. 

68 :rtype: subprocess.CompletedProcess[Any] 

69 """ 

70 command_str = _to_command_line(command) 

71 if not quiet: 

72 CONSOLE.print(f'🧩 Running: {command_str}') 

73 

74 if dry_run: 

75 return subprocess.CompletedProcess(args=command, returncode=0, stdout='', stderr='') 

76 

77 kwargs.setdefault('check', True) 

78 kwargs.setdefault('text', True) 

79 

80 if env is not None: 

81 merged_env = os.environ.copy() 

82 merged_env.update(env) 

83 kwargs['env'] = merged_env 

84 

85 if isinstance(command, str): 

86 kwargs.setdefault('shell', True) 

87 

88 return subprocess.run(command, cwd=cwd, **kwargs) 

89 

90 

91def run_stdout(command: str | list[str], *, dry_run: bool = False, quiet: bool = False, **kwargs: Any) -> str: 

92 """ 

93 Execute a command and return stripped standard output. 

94 

95 .. versionadded:: 2.2 

96 Add the `quiet` parameter to do not display the command being executed 

97 

98 :param command: Command to execute. 

99 :type command: str | list[str] 

100 :param dry_run: If ``True``, do not execute the command. 

101 :type dry_run: bool 

102 :param quiet: Suppress the standard console banner when ``True``. 

103 :type quiet: bool 

104 :param kwargs: Additional keyword arguments forwarded to ``run``. 

105 :type kwargs: Any 

106 :return: Standard output stripped of leading and trailing whitespace. 

107 :rtype: str 

108 """ 

109 kwargs['capture_output'] = True 

110 result = run(command, dry_run=dry_run, quiet=quiet, **kwargs) 

111 stdout = result.stdout 

112 if isinstance(stdout, bytes): 

113 return stdout.decode().strip() 

114 return (stdout or '').strip()