|
16 | 16 | import sys |
17 | 17 | import typing as t |
18 | 18 | from collections.abc import Iterable, Mapping, MutableMapping, Sequence |
| 19 | +from typing import IO, Any |
19 | 20 |
|
20 | 21 | from libvcs import exc |
21 | 22 | from libvcs._internal.types import StrOrBytesPath |
|
24 | 25 |
|
25 | 26 | console_encoding = sys.stdout.encoding |
26 | 27 |
|
| 28 | +PIPE_NOT_SET_ERR = ( |
| 29 | + "File descriptor is None, which should not happen as we set PIPE as default" |
| 30 | +) |
| 31 | + |
27 | 32 |
|
28 | 33 | def console_to_str(s: bytes) -> str: |
29 | 34 | """From pypa/pip project, pip.backwardwardcompat. License MIT.""" |
@@ -104,7 +109,7 @@ def __call__(self, output: t.AnyStr, timestamp: datetime.datetime) -> None: |
104 | 109 | ] |
105 | 110 |
|
106 | 111 | _CMD = t.Union[StrOrBytesPath, Sequence[StrOrBytesPath]] |
107 | | -_FILE: TypeAlias = t.Optional[t.Union[int, t.IO[t.Any]]] |
| 112 | +_FILE: TypeAlias = t.Optional[t.Union[int, IO[Any]]] |
108 | 113 |
|
109 | 114 |
|
110 | 115 | def run( |
@@ -226,12 +231,23 @@ def progress_cb(output: t.AnyStr, timestamp: datetime.datetime) -> None: |
226 | 231 | if callback and callable(callback): |
227 | 232 | callback(output="\r", timestamp=datetime.datetime.now()) |
228 | 233 |
|
229 | | - lines = filter(None, (line.strip() for line in proc.stdout.readlines())) |
230 | | - all_output = console_to_str(b"\n".join(lines)) |
| 234 | + if proc.stdout is None: |
| 235 | + raise RuntimeError(PIPE_NOT_SET_ERR) |
| 236 | + |
| 237 | + stdout_lines: list[bytes] = [ |
| 238 | + line.strip() for line in proc.stdout.readlines() if line |
| 239 | + ] |
| 240 | + all_output = [console_to_str(b"\n".join(stdout_lines))] |
| 241 | + |
231 | 242 | if code: |
232 | | - stderr_lines = filter(None, (line.strip() for line in proc.stderr.readlines())) |
233 | | - all_output = console_to_str(b"".join(stderr_lines)) |
| 243 | + if proc.stderr is None: |
| 244 | + raise RuntimeError(PIPE_NOT_SET_ERR) |
| 245 | + stderr_lines: list[bytes] = [ |
| 246 | + line.strip() for line in proc.stderr.readlines() if line |
| 247 | + ] |
| 248 | + all_output = [console_to_str(b"".join(stderr_lines))] |
| 249 | + |
234 | 250 | output = "".join(all_output) |
235 | 251 | if code != 0 and check_returncode: |
236 | | - raise exc.CommandError(output=output, returncode=code, cmd=args) |
| 252 | + raise exc.CommandError(output=output, returncode=code, cmd=str(args)) |
237 | 253 | return output |
0 commit comments