# Copyright Kevin Deldycke <kevin@deldycke.com> and contributors.
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
"""CLI testing and simulation of their execution."""
from __future__ import annotations
import inspect
import logging
import os
import re
import shutil
import subprocess
import sys
import tempfile
from contextlib import contextmanager, nullcontext
from dataclasses import dataclass
from functools import cached_property, partial
from itertools import zip_longest
from textwrap import indent
import click
import click.testing
from boltons.strutils import strip_ansi
from boltons.tbutils import ExceptionInfo
from cloup import Color
from ._utils import patch_attr
# The CLI-invocation serialization and disclosure atoms moved to
# click_extra.execution once production code (subprocess wrappers, not just
# tests) started depending on them. They are imported under private names so
# this module's namespace does not re-export them: their canonical home is
# click_extra.execution.
from .execution import (
INDENT as _INDENT,
args_cleanup as _args_cleanup,
format_cli_prompt as _format_cli_prompt,
)
from .styling import Style
TYPE_CHECKING = False
if TYPE_CHECKING:
from collections.abc import Iterable, Iterator
from contextlib import AbstractContextManager
from typing import IO, Any, Literal
from .envvar import TEnvVars
from .execution import TArg, TNestedArgs
OUTPUT_LABEL = "<output>"
"""Label for the merged stream, where stdout and stderr are interleaved."""
STDOUT_LABEL = "<stdout>"
"""Label for the standard output stream."""
STDERR_LABEL = "<stderr>"
"""Label for the standard error stream."""
EXIT_CODE_LABEL = "<exit_code>"
"""Label for the process exit code."""
STREAM_FIELDS = {
"output_": (OUTPUT_LABEL, "output"),
"stdout_": (STDOUT_LABEL, "stdout"),
"stderr_": (STDERR_LABEL, "stderr"),
}
"""Maps a test-case field prefix to its stream label and {class}`StreamView` attribute.
`output_*` directives target the merged stream; `stdout_*` and `stderr_*`
target the separate streams. Both {func}`render_cli_run` and
{meth}`click_extra.test_suite.CLITestCase.run_cli_test` read this single table so the
rendered trace and the assertion loop agree on labels and stream selection.
"""
[docs]
@dataclass(frozen=True)
class StreamView:
"""Normalized view of a CLI run's captured streams and exit code.
Both runners produce one of these so the renderer and the assertion loop read a
single shape, regardless of whether the run was driven in-process (Click's
{class}`click.testing.Result`) or as a black-box subprocess
({class}`subprocess.CompletedProcess`).
A run captures either the merged stream (`output`) or the separate `stdout` and
`stderr` streams, never both: the unused fields stay empty.
"""
stdout: str = ""
"""Captured standard output, or empty when the merged stream was captured."""
stderr: str = ""
"""Captured standard error, or empty when the merged stream was captured."""
output: str = ""
"""Captured merged stream (stdout and stderr interleaved), or empty when the
separate streams were captured."""
exit_code: int | None = None
"""Process exit code, or `None` when unavailable."""
[docs]
@classmethod
def from_result(cls, result: click.testing.Result) -> StreamView:
"""Build a view from an in-process {class}`click.testing.Result`.
Click always exposes `stdout`, `stderr` and the interleaved `output`
together, so all three are carried over verbatim.
"""
return cls(
stdout=result.stdout,
stderr=result.stderr,
output=result.output,
exit_code=result.exit_code,
)
[docs]
@classmethod
def from_completed_process(cls, result: subprocess.CompletedProcess) -> StreamView:
"""Build a view from a black-box {class}`subprocess.CompletedProcess`.
A subprocess run with stderr merged into stdout (`stderr=STDOUT`) reports
`result.stderr` as `None`: that case is rendered as the interleaved
`output` stream. Otherwise the two streams are kept separate.
"""
if result.stderr is None:
return cls(output=result.stdout or "", exit_code=result.returncode)
return cls(
stdout=result.stdout,
stderr=result.stderr,
exit_code=result.returncode,
)
[docs]
def render_cli_run(
args: Iterable[str],
result: click.testing.Result | subprocess.CompletedProcess,
env: TEnvVars | None = None,
) -> str:
"""Generates the full simulation of CLI execution, including output.
Mostly used to print debug traces to user or in test results.
"""
prompt = _format_cli_prompt(args, env)
if isinstance(result, click.testing.Result):
view = StreamView.from_result(result)
else:
view = StreamView.from_completed_process(result)
# Per-stream colors for the rendered trace, keyed by the shared stream labels.
stream_colors = {
OUTPUT_LABEL: Color.blue,
STDOUT_LABEL: Color.green,
STDERR_LABEL: Color.red,
}
# Render the execution trace.
trace = [prompt]
for label, attr in STREAM_FIELDS.values():
content = getattr(view, attr)
if content:
trace.append(f"{Style(fg=stream_colors[label])(label)} stream:")
trace.append(indent(content, _INDENT))
if view.exit_code is not None:
trace.append(f"{Style(fg=Color.yellow)(EXIT_CODE_LABEL)}: {view.exit_code}")
return "\n".join(trace)
def _print_cli_run(
args: Iterable[str],
result: click.testing.Result | subprocess.CompletedProcess,
env: TEnvVars | None = None,
) -> None:
"""Prints the full simulation of CLI execution, including output.
The trace embeds the CLI's captured streams verbatim, and those are whatever
bytes a subprocess happened to write rather than text this library produced.
Printing them can fail on its own: a `PYTHONIOENCODING` carrying no error
handler leaves `sys.stdout` strict, and a byte the codec refuses then raises
from the `print` rather than from anything under test.
So a failed write is retried against a sanitized copy, marking each
unprintable character rather than losing the trace. A debugging aid must
never be the reason a test fails, and this one is reached from every
`CliRunner` invocation.
"""
trace = render_cli_run(args, result, env)
try:
print(trace)
except (UnicodeError, OSError):
# OSError covers the platform-level refusal, which macOS reports as
# EILSEQ from the write itself instead of raising from the codec.
encoding = getattr(sys.stdout, "encoding", None) or "UTF-8"
print(trace.encode(encoding, "replace").decode(encoding, "replace"))
INVOKE_ARGS = set(inspect.getfullargspec(click.testing.CliRunner.invoke).args)
"""Parameter IDs of `click.testing.CliRunner.invoke()`.
We need to collect them to help us identify which extra parameters passed to
`invoke()` collides with its original signature.
```{warning}
This has been [reported upstream to Click project](https://github.com/pallets/click/issues/2110) but has been rejected and not
considered an issue worth fixing.
```
"""
[docs]
class Result(click.testing.Result):
"""A `Result` subclass with automatic traceback formatting.
Enhances `__repr__` so that pytest assertion failures show the full
traceback instead of just the exception type.
"""
def __repr__(self) -> str:
if self.formatted_exception:
return f"<{type(self).__name__}\n{self.formatted_exception}>"
exc_str = repr(self.exception) if self.exception else "okay"
return f"<{type(self).__name__} {exc_str}>"
[docs]
@contextmanager
def isolated_filesystem(
temp_dir: str | os.PathLike[str] | None = None,
) -> Iterator[str]:
"""Create a temporary directory and change the working directory to it.
Reimplements {meth}`click.testing.CliRunner.isolated_filesystem`, which Click
deprecated in `8.5.0` and removes in `9.0`. Click points at
{class}`tempfile.TemporaryDirectory` and pytest's `tmp_path` fixture, and
neither replaces it: what isolates a CLI here is the *working directory*,
which a command reads through relative paths and configuration-file globs.
The `runner` pytest fixture and the `click:run` Sphinx directive both call
this on every test and every documented example, so the deprecated version
buries a suite run under thousands of warnings, then stops working.
```{warning}
The isolation is process-global, because it calls {func}`os.chdir`.
Parallelize with processes, never with threads.
```
:param temp_dir: create the temporary directory inside this one. When given,
the directory is left in place on exit.
:return: path of the directory the working directory was changed to.
"""
previous_cwd = os.getcwd()
target = tempfile.mkdtemp(dir=temp_dir)
os.chdir(target)
try:
yield target
finally:
os.chdir(previous_cwd)
if temp_dir is None:
shutil.rmtree(target, ignore_errors=True)
[docs]
class CliRunner(click.testing.CliRunner):
"""Augment {class}`click.testing.CliRunner` with extra features and bug fixes."""
force_color: bool = False
"""Global class attribute to override the `color` parameter in `invoke`."""
[docs]
def invoke( # type: ignore[override]
self,
cli: click.Command,
*args: TArg | TNestedArgs,
input: str | bytes | IO | None = None,
env: TEnvVars | None = None,
catch_exceptions: bool = True,
color: bool | Literal["forced"] | None = None,
**extra: Any,
) -> Result:
"""Same as `click.testing.CliRunner.invoke()` with extra features.
- The first positional parameter is the CLI to invoke. The remaining positional
parameters of the function are the CLI arguments. All other parameters are
required to be named.
- The CLI arguments can be nested iterables of arbitrary depth. This is
[useful for argument composition of test cases with @pytest.mark.parametrize](https://docs.pytest.org/en/stable/example/parametrize.html).
- Allow forcing of the `color` property at the class-level via
`force_color` attribute.
- Adds a special case in the form of `color="forced"` parameter, which allows
colored output to be kept, while forcing the initialization of
`Context.color = True`. This is [not allowed in current implementation](https://github.com/pallets/click/issues/2110) of
`click.testing.CliRunner.invoke()` because of colliding parameters.
- Strips all ANSI codes from results if `color` was explicitly set to
`False`.
- Always prints a simulation of the CLI execution as the user would see it in
its terminal. Including colors.
- Pretty-prints a formatted exception traceback if the command fails.
:param cli: CLI to invoke.
:param args: can be nested iterables composed of `str`,
:py:class:`pathlib.Path` objects and `None` values. The nested structure
will be flattened and `None` values will be filtered out. Then all
elements will be cast to `str`. See
{func}`~click_extra.execution.args_cleanup` for details.
:param input: same as `click.testing.CliRunner.invoke()`.
:param env: same as `click.testing.CliRunner.invoke()`.
:param catch_exceptions: same as `click.testing.CliRunner.invoke()`.
:param color: If a boolean, the parameter will be passed as-is to
`click.testing.CliRunner.isolation()`. If `"forced"`, the parameter
will be passed as `True` to `click.testing.CliRunner.isolation()` and
an extra `color=True` parameter will be passed to the invoked CLI.
:param extra: same as `click.testing.CliRunner.invoke()`, but colliding
parameters are allowed and properly passed on to the invoked CLI.
"""
# Pop out the `args` parameter from `extra` and append it to the positional
# arguments. This handles the case where `args` is passed as a keyword
# argument, as in vanilla Click's `CliRunner.invoke()` API.
cli_args = list(args)
if "args" in extra:
cli_args.extend(extra.pop("args"))
# Flatten and filters out CLI arguments.
clean_args = _args_cleanup(*cli_args)
if color == "forced":
# Pass the color argument as an extra parameter to the invoked CLI.
# This works around Click issue #2110: `CliRunner.invoke(color=True)`
# controls the test "terminal" but cannot simultaneously pass `color`
# through to `Context`.
extra["color"] = True
# The class attribute `force_color` overrides the `color` parameter.
if self.force_color:
isolation_color = True
# Cast to `bool` to avoid passing `None` or `"forced"` to `invoke()`.
else:
isolation_color = bool(color)
# No-op context manager without any effects.
extra_params_bypass: AbstractContextManager = nullcontext()
# Capture a simulated prog name before `extra` is consumed: an explicit
# `prog_name` (how multicall personalities are simulated) is what the
# printed command line should show, not the CLI's own name.
simulated_prog_name = extra.get("prog_name") or self.get_default_prog_name(cli)
# If `extra` contains parameters that collide with the original `invoke()`
# parameters, we need to remove them from `extra`, then use a monkeypatch to
# properly pass them to the CLI.
colliding_params = INVOKE_ARGS.intersection(extra)
if colliding_params:
# Transfer colliding parameters from `extra` to `extra_bypass`.
extra_bypass = {pid: extra.pop(pid) for pid in colliding_params}
# Monkeypatch the original command's `main()` call to pass extra
# parameter for `Context` initialization. Because we cannot simply add
# colliding parameter IDs to `**extra`.
extra_params_bypass = patch_attr(
cli,
"main",
partial(cli.main, **extra_bypass),
)
with extra_params_bypass:
result = super().invoke(
cli=cli,
args=clean_args,
input=input,
env=env,
catch_exceptions=catch_exceptions,
color=isolation_color,
**extra,
)
# Upgrade the result to our subclass for automatic traceback formatting.
result.__class__ = Result
extra_result: Result = result # type: ignore[assignment]
# `color` has been explicitly set to `False`, so strip all ANSI codes.
if color is False:
extra_result.stdout_bytes = strip_ansi(extra_result.stdout_bytes) # type: ignore[assignment,arg-type]
extra_result.stderr_bytes = strip_ansi(extra_result.stderr_bytes) # type: ignore[assignment,arg-type]
extra_result.output_bytes = strip_ansi(extra_result.output_bytes) # type: ignore[assignment,arg-type]
_print_cli_run(
[simulated_prog_name, *clean_args],
extra_result,
env=env,
)
if extra_result.formatted_exception:
print(extra_result.formatted_exception)
return extra_result
[docs]
@contextmanager
def isolated_filesystem(
self, temp_dir: str | os.PathLike[str] | None = None
) -> Iterator[str]:
"""Same as {func}`isolated_filesystem`, minus Click's deprecation warning.
Overrides {meth}`click.testing.CliRunner.isolated_filesystem` so a runner
keeps the helper once Click `9.0` removes it.
"""
# Resolves to the module-level function above: a class attribute never
# shadows a global inside a method body.
with isolated_filesystem(temp_dir) as target:
yield target
[docs]
def unescape_regex(text: str) -> str:
"""De-obfuscate a regex for better readability.
This is like the reverse of `re.escape()`.
"""
char_map = {
escaped_char: chr(single_char)
for single_char, escaped_char in (
re._special_chars_map.items() # type: ignore[attr-defined]
)
}
char_map.update({r"\x1b": "\x1b"})
for escaped, char in char_map.items():
text = text.replace(escaped, char)
return text
[docs]
class RegexLineMismatch(AssertionError):
"""Raised when a regex line does not match the corresponding content line."""
def __init__(self, regex_line: str, content_line: str, line_number: int) -> None:
# De-obfuscate the regex to allow for comparison with the output.
self.regex_line = unescape_regex(regex_line)
self.content_line = content_line
self.line_number = line_number
message = (
f"Line #{self.line_number} does not match.\n"
f"Regex : {self.regex_line!r}\n"
f"Output: {self.content_line!r}"
)
super().__init__(message)
REGEX_NEWLINE = "\\n"
"""Newline token used to split a multi-line regex pattern for line-by-line
matching."""
[docs]
def regex_fullmatch_line_by_line(regex: re.Pattern | str, content: str) -> None:
"""Check that the `content` matches the given `regex`.
If the `regex` does not fully match the `content`, raise an `AssertionError`,
with a message showing the first mismatching line.
This is useful when comparing large walls of text, such as CLI output.
"""
# If the regex fully match the output right away, no need for a custom message.
if re.fullmatch(regex, content):
return
content_lines = content.splitlines(keepends=True)
if isinstance(regex, str):
regex_lines = [line + REGEX_NEWLINE for line in regex.split(REGEX_NEWLINE)]
else:
regex_lines = regex.pattern.splitlines(keepends=True)
# Pad the shorter side with empty lines so a length mismatch is reported as
# a regular line mismatch instead of crashing on an out-of-range index.
for i, (regex_line, content_line) in enumerate(
zip_longest(regex_lines, content_lines, fillvalue=""),
):
try:
matched = re.fullmatch(regex_line, content_line)
except re.error:
# Multi-line groups like (?:...\n)* get broken when the regex is
# split on \n for line-by-line comparison. Fall back to reporting
# the overall content so the mismatch is diagnosable.
msg = f"Full regex does not match output.\nContent:\n{content}"
raise AssertionError(msg) from None
if matched:
logging.debug(
f"Line #{i + 1} match.\n"
f"Regex : {regex_line!r}\n"
f"Output: {content_line!r}"
)
else:
raise RegexLineMismatch(regex_line, content_line, i + 1)