# 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 re
import subprocess
from contextlib import 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
# 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 here under private names
# for this module's own use; their historical public click_extra.testing names
# (INDENT, PROMPT, args_cleanup, format_cli_prompt) still resolve through the
# deprecated-alias __getattr__ hook at the bottom of this module. See
# click_extra._deprecated.
from .execution import (
INDENT as _INDENT,
args_cleanup as _args_cleanup,
format_cli_prompt as _format_cli_prompt,
)
from .parameters import patch_attr
from .styling import Style
TYPE_CHECKING = False
if TYPE_CHECKING:
from collections.abc import Iterable
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."""
print(render_cli_run(args, result, env))
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]
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()
# 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(
[self.get_default_prog_name(cli), *clean_args],
extra_result,
env=env,
)
if extra_result.formatted_exception:
print(extra_result.formatted_exception)
return extra_result
[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)
def __getattr__(name: str) -> Any:
"""Resolve deprecated `testing` symbols via the PEP 562 `__getattr__` hook.
INDENT, PROMPT, args_cleanup and format_cli_prompt moved to
{mod}`click_extra.execution`. Fires only for names not defined in this
module. See {mod}`click_extra._deprecated`.
"""
from ._deprecated import resolve_deprecated
return resolve_deprecated(__name__, name)