Coverage for pycommons / processes / shell_to_file.py: 99%

1r""" 

2A tool for invoking shell commands and piping their output to files. 

3 

4To do this more or less safely and reliably, we create a script that 

5invokes the original command. The script is created as temporary file 

6and will be deleted after the command completes. 

7 

8>>> from pycommons.io.temp import temp_dir 

9>>> cmd = Command(("echo", "123")) 

10 

11>>> with temp_dir() as td: 

12... so = td.resolve_inside("so.txt") 

13... se = td.resolve_inside("se.txt") 

14... to_files(cmd, so, se).execute() 

15... print(f"so: {so.read_all_str()}") 

16(None, None) 

17so: 123 

18<BLANKLINE> 

19 

20>>> with temp_dir() as td: 

21... so = td.resolve_inside("so.txt") 

22... to_files(cmd, so, so).execute() 

23... print(f"so: {so.read_all_str()}") 

24(None, None) 

25so: 123 

26<BLANKLINE> 

27 

28>>> try: 

29... to_files("a", "a", "b") 

30... except TypeError as te: 

31... print(str(te)[:10]) 

32command sh 

33 

34>>> try: 

35... to_files(cmd, 1, "b") 

36... except TypeError as te: 

37... print(te) 

38stdout should be an instance of any in {None, str} but is int, namely 1. 

39 

40>>> try: 

41... to_files(cmd, "a", 1) 

42... except TypeError as te: 

43... print(te) 

44stderr should be an instance of any in {None, str} but is int, namely 1. 

45 

46>>> try: 

47... to_files(cmd, None, None) 

48... except ValueError as ve: 

49... print(ve) 

50Either stdout or stderr must be specified. 

51""" 

52 

53import platform 

54from dataclasses import dataclass 

55from os import chmod 

56from stat import S_IRUSR, S_IWUSR, S_IXUSR # nosec 

57from subprocess import list2cmdline # nosec 

58from typing import Final, Iterable, Mapping 

59 

60from pycommons.io.console import logger 

61from pycommons.io.path import Path 

62from pycommons.io.temp import temp_file 

63from pycommons.processes.shell import STREAM_IGNORE, Command 

64from pycommons.types import type_error 

65 

66#: the internal operating system 

67_WINDOWS: Final[bool] = platform.system() == "Windows" 

68 

69 

70@dataclass(frozen=True, init=False, order=False, eq=False) 

71class __FileCommand(Command): 

72 """The internal file shell command data class.""" 

73 

74 #: the file to receive the standard output 

75 stdout_file: Path | None 

76 #: the file to receive the standard error output 

77 stderr_file: Path | None 

78 

79 def __init__(self, command: str | Iterable[str], 

80 working_dir: str | None = None, 

81 timeout: int | None = 3600, 

82 stdout: str | None = None, 

83 stderr: str | None = None, 

84 env: Mapping[str, str] | Iterable[tuple[ 

85 str, str]] | None = None) -> None: 

86 """ 

87 Initialize the file-based command object. 

88 

89 :param command: the command 

90 :param working_dir: the working directory 

91 :param timeout: the timeout, if any 

92 :param stdout: the file to capture the stdout, or `None` if 

93 stdout should be ignored 

94 :param stderr: the file to capture the stderr, or `None` if 

95 stderr should be ignored 

96 :param env: the environment variables to use 

97 """ 

98 super().__init__(command, working_dir, timeout, None, 

99 STREAM_IGNORE, STREAM_IGNORE, env) 

100 sof: Final[Path | None] = None if stdout is None else Path(stdout) 

101 sef: Path | None = None if stderr is None else Path(stderr) 

102 if (sof is not None) and (sef is not None) and (sof == sef): 

103 sef = sof 

104 object.__setattr__(self, "stdout_file", sof) 

105 object.__setattr__(self, "stderr_file", sef) 

106 

107 def __str__(self) -> str: 

108 """ 

109 Get the string representation of this command. 

110 

111 :return: the command's string representation 

112 

113 >>> cmd = Command(("echo", "123")) 

114 >>> str(to_files(cmd, "/tmp/x", "/tmp/x"))[-22:] 

115 "stdout+stderr>'/tmp/x'" 

116 >>> str(to_files(cmd, "/tmp/x", None))[-15:] 

117 "stdout>'/tmp/x'" 

118 >>> str(to_files(cmd, None, "/tmp/x"))[-15:] 

119 "stderr>'/tmp/x'" 

120 """ 

121 old: str = super().__str__() 

122 so: Final[Path | None] = self.stdout_file 

123 se: Final[Path | None] = self.stderr_file 

124 if so is se: 

125 return f"{old}, stdout+stderr>{so!r}" 

126 if so is not None: 

127 old = f"{old}, stdout>{so!r}" 

128 return old if se is None else f"{old}, stderr>{se!r}" 

129 

130 def execute(self, log_call: bool = True) -> tuple[None, None]: 

131 """ 

132 Execute the command. 

133 

134 :param log_call: shall we log the call? 

135 :return: always `(None, None)` 

136 """ 

137 text: Final[list[str]] = [str(list2cmdline(self.command))] 

138 with temp_file(directory=self.working_dir, 

139 suffix=".bat" if _WINDOWS else ".sh") as temp: 

140 if log_call: 

141 logger(f"Using temp file {temp!r} as execution " 

142 f"wrapper for {self}.") 

143 if not _WINDOWS: 

144 text.insert(0, "#!/bin/bash\n") 

145 sof: Final[Path | None] = self.stdout_file 

146 sef: Final[Path | None] = self.stderr_file 

147 if sof is not None: 

148 text.append(f' 1>"{sof}"') 

149 if sef is not None: 

150 text.append(" 2>&1" if sef is sof else f' 2>"{sef}"') 

151 temp.write_all_str("".join(text)) 

152 if not _WINDOWS: 

153 chmod(temp, S_IRUSR | S_IWUSR | S_IXUSR) 

154 Command(command=temp, 

155 working_dir=self.working_dir, 

156 timeout=self.timeout, 

157 stdin=None, 

158 stderr=STREAM_IGNORE, 

159 stdout=STREAM_IGNORE, 

160 env=self.env).execute(log_call) 

161 if sof is not None: 

162 sof.enforce_file() 

163 if sef is not None: 

164 sef.enforce_file() 

165 return None, None 

166 

167 

168def to_files(command: Command, stdout: str | None, 

169 stderr: str | None) -> Command: 

170 """ 

171 Take an existing command and forward its stdout and/or stderr to files. 

172 

173 Currently, providing text as standard input is not supported. 

174 You can provide either different or the same file for the standard output 

175 and standard error. If the same file is provided, then both streams will 

176 be merged into that file. 

177 Either way, the files you provide will be created and overwritten during 

178 the command execution. 

179 Notice that whatever original settings for standard error and standard 

180 output you provided in the original 

181 :class:`~pycommons.processes.shell.Command` instance `command` will be 

182 ignored. 

183 

184 :param command: the command 

185 :param stdout: the file to capture the stdout, or `None` if stdout 

186 should be ignored 

187 :param stderr: the file to capture the stderr, or `None` if stderr 

188 should be ignored 

189 :return: the new command 

190 """ 

191 if not isinstance(command, Command): 

192 raise type_error(command, "command", Command) 

193 if command.stdin is not None: 

194 raise ValueError("Stdin is not supported for file commands.") 

195 if (stdout is not None) and (not isinstance(stdout, str)): 

196 raise type_error(stdout, "stdout", (str, None)) 

197 if (stderr is not None) and (not isinstance(stderr, str)): 

198 raise type_error(stderr, "stderr", (str, None)) 

199 if (stdout is None) and (stderr is None): 

200 raise ValueError("Either stdout or stderr must be specified.") 

201 return __FileCommand(command=command.command, 

202 working_dir=command.working_dir, 

203 timeout=command.timeout, 

204 stdout=stdout, 

205 stderr=stderr, 

206 env=command.env)