Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Keep track of <stderr> and <stdout> mix in CliRunner results #2523

Open
wants to merge 1 commit into
base: main
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
4 changes: 4 additions & 0 deletions CHANGES.rst
Expand Up @@ -31,6 +31,10 @@ Unreleased
- When generating a command's name from a decorated function's name, the
suffixes ``_command``, ``_cmd``, ``_group``, and ``_grp`` are removed.
:issue:`2322`
- Keep `<stdout>` and `<stderr>` streams independent in `CliRunner`. Always
collect `<stderr>` output and never raise an exception. Add a new
`<output>` stream to simulate what the user sees in its terminal. Removes
the ``mix_stderr`` parameter in ``CliRunner``. :issue:`2522` :pr:`2523`


Version 8.1.7
Expand Down
138 changes: 98 additions & 40 deletions src/click/testing.py
Expand Up @@ -17,6 +17,8 @@
from ._compat import _find_binary_reader

if t.TYPE_CHECKING:
from _typeshed import ReadableBuffer

from .core import Command


Expand Down Expand Up @@ -64,6 +66,39 @@ def _pause_echo(stream: EchoingStdin | None) -> cabc.Iterator[None]:
stream._paused = False


class BytesIOCopy(io.BytesIO):
"""Patch ``io.BytesIO`` to let the written stream be copied to another.

.. versionadded:: 8.2
"""

def __init__(self, copy_to: io.BytesIO) -> None:
super().__init__()
self.copy_to = copy_to

def flush(self) -> None:
super().flush()
self.copy_to.flush()

def write(self, b: ReadableBuffer) -> int:
self.copy_to.write(b)
return super().write(b)


class StreamMixer:
"""Mixes `<stdout>` and `<stderr>` streams.

The result is available in the ``output`` attribute.

.. versionadded:: 8.2
"""

def __init__(self) -> None:
self.output: io.BytesIO = io.BytesIO()
self.stdout: io.BytesIO = BytesIOCopy(copy_to=self.output)
self.stderr: io.BytesIO = BytesIOCopy(copy_to=self.output)


class _NamedTextIOWrapper(io.TextIOWrapper):
def __init__(
self, buffer: t.BinaryIO, name: str, mode: str, **kwargs: t.Any
Expand Down Expand Up @@ -108,7 +143,8 @@ def __init__(
self,
runner: CliRunner,
stdout_bytes: bytes,
stderr_bytes: bytes | None,
stderr_bytes: bytes,
output_bytes: bytes,
return_value: t.Any,
exit_code: int,
exception: BaseException | None,
Expand All @@ -119,8 +155,16 @@ def __init__(
self.runner = runner
#: The standard output as bytes.
self.stdout_bytes = stdout_bytes
#: The standard error as bytes, or None if not available
#: The standard error as bytes.
#:
#: .. versionchanged:: 8.2
#: No longer optional.
self.stderr_bytes = stderr_bytes
#: A mix of `stdout_bytes` and `stderr_bytes``, as the user would see
# it in its terminal.
#:
#: .. versionadded:: 8.2
self.output_bytes = output_bytes
#: The value returned from the invoked command.
#:
#: .. versionadded:: 8.0
Expand All @@ -134,8 +178,15 @@ def __init__(

@property
def output(self) -> str:
"""The (standard) output as unicode string."""
return self.stdout
"""The terminal output as unicode string, as the user would see it.

.. versionchanged:: 8.2
No longer a proxy for ``self.stdout``. Now has its own independent stream
that is mixing `<stdout>` and `<stderr>`, in the order they were written.
"""
return self.output_bytes.decode(self.runner.charset, "replace").replace(
"\r\n", "\n"
)

@property
def stdout(self) -> str:
Expand All @@ -146,9 +197,11 @@ def stdout(self) -> str:

@property
def stderr(self) -> str:
"""The standard error as unicode string."""
if self.stderr_bytes is None:
raise ValueError("stderr not separately captured")
"""The standard error as unicode string.

.. versionchanged:: 8.2
No longer raise an exception, always returns the `<stderr>` string.
"""
return self.stderr_bytes.decode(self.runner.charset, "replace").replace(
"\r\n", "\n"
)
Expand All @@ -166,28 +219,24 @@ class CliRunner:

:param charset: the character set for the input and output data.
:param env: a dictionary with environment variables for overriding.
:param echo_stdin: if this is set to `True`, then reading from stdin writes
to stdout. This is useful for showing examples in
:param echo_stdin: if this is set to `True`, then reading from `<stdin>` writes
to `<stdout>`. This is useful for showing examples in
some circumstances. Note that regular prompts
will automatically echo the input.
:param mix_stderr: if this is set to `False`, then stdout and stderr are
preserved as independent streams. This is useful for
Unix-philosophy apps that have predictable stdout and
noisy stderr, such that each may be measured
independently

.. versionchanged:: 8.2
``mix_stderr`` parameter has been removed.
"""

def __init__(
self,
charset: str = "utf-8",
env: cabc.Mapping[str, str | None] | None = None,
echo_stdin: bool = False,
mix_stderr: bool = True,
) -> None:
self.charset = charset
self.env: cabc.Mapping[str, str | None] = env or {}
self.echo_stdin = echo_stdin
self.mix_stderr = mix_stderr

def get_default_prog_name(self, cli: Command) -> str:
"""Given a command object it will return the default program name
Expand All @@ -211,22 +260,29 @@ def isolation(
input: str | bytes | t.IO[t.Any] | None = None,
env: cabc.Mapping[str, str | None] | None = None,
color: bool = False,
) -> cabc.Iterator[tuple[io.BytesIO, io.BytesIO | None]]:
) -> cabc.Iterator[tuple[io.BytesIO, io.BytesIO, io.BytesIO]]:
"""A context manager that sets up the isolation for invoking of a
command line tool. This sets up stdin with the given input data
command line tool. This sets up `<stdin>` with the given input data
and `os.environ` with the overrides from the given dictionary.
This also rebinds some internals in Click to be mocked (like the
prompt functionality).

This is automatically done in the :meth:`invoke` method.

:param input: the input stream to put into sys.stdin.
:param input: the input stream to put into `sys.stdin`.
:param env: the environment overrides as dictionary.
:param color: whether the output should contain color codes. The
application can still override this explicitly.

.. versionadded:: 8.2
An additional output stream is returned, which is a mix of
`<stdout>` and `<stderr>` streams.

.. versionchanged:: 8.2
Always returns the `<stderr>` stream.

.. versionchanged:: 8.0
``stderr`` is opened with ``errors="backslashreplace"``
`<stderr>` is opened with ``errors="backslashreplace"``
instead of the default ``"strict"``.

.. versionchanged:: 4.0
Expand All @@ -243,11 +299,11 @@ def isolation(

env = self.make_env(env)

bytes_output = io.BytesIO()
stream_mixer = StreamMixer()

if self.echo_stdin:
bytes_input = echo_input = t.cast(
t.BinaryIO, EchoingStdin(bytes_input, bytes_output)
t.BinaryIO, EchoingStdin(bytes_input, stream_mixer.stdout)
)

sys.stdin = text_input = _NamedTextIOWrapper(
Expand All @@ -260,21 +316,16 @@ def isolation(
text_input._CHUNK_SIZE = 1 # type: ignore

sys.stdout = _NamedTextIOWrapper(
bytes_output, encoding=self.charset, name="<stdout>", mode="w"
stream_mixer.stdout, encoding=self.charset, name="<stdout>", mode="w"
)

bytes_error = None
if self.mix_stderr:
sys.stderr = sys.stdout
else:
bytes_error = io.BytesIO()
sys.stderr = _NamedTextIOWrapper(
bytes_error,
encoding=self.charset,
name="<stderr>",
mode="w",
errors="backslashreplace",
)
sys.stderr = _NamedTextIOWrapper(
stream_mixer.stderr,
encoding=self.charset,
name="<stderr>",
mode="w",
errors="backslashreplace",
)

@_pause_echo(echo_input) # type: ignore
def visible_input(prompt: str | None = None) -> str:
Expand Down Expand Up @@ -329,7 +380,7 @@ def should_strip_ansi(
pass
else:
os.environ[key] = value
yield (bytes_output, bytes_error)
yield (stream_mixer.stdout, stream_mixer.stderr, stream_mixer.output)
finally:
for key, value in old_env.items():
if value is None:
Expand Down Expand Up @@ -378,6 +429,14 @@ def invoke(
:param color: whether the output should contain color codes. The
application can still override this explicitly.

.. versionadded:: 8.2
The result object has the ``output_bytes`` attribute with
the mix of ``stdout_bytes`` and ``stderr_bytes``, as the user would
see it in its terminal.

.. versionchanged:: 8.2
The result object always returns the ``stderr_bytes`` stream.

.. versionchanged:: 8.0
The result object has the ``return_value`` attribute with
the value returned from the invoked command.
Expand Down Expand Up @@ -434,15 +493,14 @@ def invoke(
finally:
sys.stdout.flush()
stdout = outstreams[0].getvalue()
if self.mix_stderr:
stderr = None
else:
stderr = outstreams[1].getvalue() # type: ignore
stderr = outstreams[1].getvalue()
output = outstreams[2].getvalue()

return Result(
runner=self,
stdout_bytes=stdout,
stderr_bytes=stderr,
output_bytes=output,
return_value=return_value,
exit_code=exit_code,
exception=exception,
Expand Down
2 changes: 1 addition & 1 deletion tests/test_termui.py
Expand Up @@ -246,7 +246,7 @@ def test_secho(runner):
("value", "expect"), [(123, b"\x1b[45m123\x1b[0m"), (b"test", b"test")]
)
def test_secho_non_text(runner, value, expect):
with runner.isolation() as (out, _):
with runner.isolation() as (out, _, _):
click.secho(value, nl=False, color=True, bg="magenta")
result = out.getvalue()
assert result == expect
Expand Down
31 changes: 11 additions & 20 deletions tests/test_testing.py
Expand Up @@ -303,32 +303,23 @@ def cli_env():
def test_stderr():
@click.command()
def cli_stderr():
click.echo("stdout")
click.echo("stderr", err=True)

runner = CliRunner(mix_stderr=False)

result = runner.invoke(cli_stderr)

assert result.output == "stdout\n"
assert result.stdout == "stdout\n"
assert result.stderr == "stderr\n"
click.echo("1 - stdout")
click.echo("2 - stderr", err=True)
click.echo("3 - stdout")
click.echo("4 - stderr", err=True)

runner_mix = CliRunner(mix_stderr=True)
runner_mix = CliRunner()
result_mix = runner_mix.invoke(cli_stderr)

assert result_mix.output == "stdout\nstderr\n"
assert result_mix.stdout == "stdout\nstderr\n"

with pytest.raises(ValueError):
assert result_mix.stderr # noqa B018
assert result_mix.output == "1 - stdout\n2 - stderr\n3 - stdout\n4 - stderr\n"
assert result_mix.stdout == "1 - stdout\n3 - stdout\n"
assert result_mix.stderr == "2 - stderr\n4 - stderr\n"

@click.command()
def cli_empty_stderr():
click.echo("stdout")

runner = CliRunner(mix_stderr=False)

runner = CliRunner()
result = runner.invoke(cli_empty_stderr)

assert result.output == "stdout\n"
Expand Down Expand Up @@ -412,9 +403,9 @@ def test_isolation_stderr_errors():
"""Writing to stderr should escape invalid characters instead of
raising a UnicodeEncodeError.
"""
runner = CliRunner(mix_stderr=False)
runner = CliRunner()

with runner.isolation() as (_, err):
with runner.isolation() as (_, err, _):
click.echo("\udce2", err=True, nl=False)

assert err.getvalue() == b"\\udce2"