# 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 textwrap import indent
import click
import click.testing
from boltons.strutils import strip_ansi
from boltons.tbutils import ExceptionInfo
from . import Color, Style
# The CLI-invocation serialization and disclosure atoms moved to
# click_extra.execution once production code (subprocess wrappers, not just tests)
# started depending on them. This module still consumes them, which keeps their
# historical click_extra.testing import path working.
from .execution import (
INDENT,
args_cleanup,
format_cli_prompt,
)
from .parameters import patch_attr
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 explicirely 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 casted to ``str``. See
:func:`~click_extra.testing.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)
line_indexes = range(max(len(regex_lines), len(content_lines)))
for i in line_indexes:
regex_line = regex_lines[i]
content_line = content_lines[i]
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)