Skip to content
Snippets Groups Projects
Unverified Commit d4ea7807 authored by Newbyte's avatar Newbyte :snowflake:
Browse files

pmb.helpers.run_core: Properly type pipe_read() (MR 2464)

I think the "these types make no sense" comment were about
output_return_buffer, which defaulted to False but was to be set to
list[bytes] if used. I changed it so that it defaults to None, which
probably is more conventional. Other than that I didn't notice anything
weird about the types here.

Also check for None as necessary to appease mypy.
parent 47c84925
Branches
Tags
1 merge request!2464Add lots of type hints, fix some bugs, and clean up some code
......@@ -14,6 +14,7 @@ import sys
import threading
import time
from collections.abc import Sequence
from typing import overload, Literal
import pmb.helpers.run
"""For a detailed description of all output modes, read the description of
......@@ -94,14 +95,33 @@ def pipe(cmd, working_dir=None):
return ret
# FIXME (#2324): These types make no sense at all
@overload
def pipe_read(
process,
output_to_stdout=False,
output_log=True,
output_return=False,
output_return_buffer=False,
):
process: subprocess.Popen,
output_to_stdout: bool = ...,
output_log: bool = ...,
output_return: Literal[False] = ...,
output_return_buffer: None = ...,
) -> None: ...
@overload
def pipe_read(
process: subprocess.Popen,
output_to_stdout: bool = ...,
output_log: bool = ...,
output_return: Literal[True] = ...,
output_return_buffer: list[bytes] = ...,
) -> None: ...
def pipe_read(
process: subprocess.Popen,
output_to_stdout: bool = False,
output_log: bool = True,
output_return: bool = False,
output_return_buffer: list[bytes] | None = None,
) -> None:
"""Read all output from a subprocess, copy it to the log and optionally stdout and a buffer variable.
This is only meant to be called by foreground_pipe() below.
......@@ -115,13 +135,18 @@ def pipe_read(
"""
while True:
# Copy available output
out = process.stdout.readline()
process_stdout = process.stdout
if process_stdout is None:
raise RuntimeError("subprocess has no stdout?")
out = process_stdout.readline()
if len(out):
if output_log:
pmb.helpers.logging.logfd.buffer.write(out)
if output_to_stdout:
sys.stdout.buffer.write(out)
if output_return:
if output_return_buffer is None:
raise AssertionError
output_return_buffer.append(out)
continue
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment