# 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 chdir, 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 cloup import Color
from ._utils import patch_attr
from .ansi import strip_ansi
# Imported under private names, so this module does not re-export them: their
# home is click_extra.execution, where production code reaches them too.
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 Generator, Iterable
from contextlib import AbstractContextManager
from typing import IO, Any
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."""
REGEX_NEWLINE = "\\n"
"""Newline token used to split a multi-line regex pattern for line-by-line
matching."""
INVOKE_ARGS = set(inspect.getfullargspec(click.testing.CliRunner.invoke).args)
"""Parameter names of {meth}`click.testing.CliRunner.invoke`.
An extra keyword of {meth}`CliRunner.invoke` named like one of them collides with
Click's own parameter, and reaches `Context` through a patched `Command.main()`
call instead ([pallets/click#2110](https://github.com/pallets/click/issues/2110)).
```{todo}
[pallets/click#3896](https://github.com/pallets/click/pull/3896) adds a
`context_settings` parameter to `click.testing.CliRunner.invoke()`. It sets the
context color without the `Command.main()` patch in
{meth}`~click_extra.testing.CliRunner.invoke`.
On a Click release that carries it, this constant also catches
`context_settings`, and the patch sends it to `Context`, which rejects it. On
such a release, keep `context_settings` out of the collision set and pass
`force_color` through it. Drop this constant and the patch once the Click floor
of this package reaches that release.
```
"""
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.
"""
_STREAM_COLORS = {
OUTPUT_LABEL: Color.blue,
STDOUT_LABEL: Color.green,
STDERR_LABEL: Color.red,
}
"""Color of each stream label in the trace {func}`render_cli_run` draws."""
[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:
"""Render the full simulation of a CLI execution, output included.
{meth}`CliRunner.invoke` prints it after every run, so a failing test shows
the session that produced it.
"""
if isinstance(result, click.testing.Result):
view = StreamView.from_result(result)
else:
view = StreamView.from_completed_process(result)
trace = [_format_cli_prompt(args, env)]
for label, attr in STREAM_FIELDS.values():
content = getattr(view, attr)
if content:
trace.extend((
f"{Style(fg=_STREAM_COLORS[label])(label)} stream:",
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:
"""Print the full simulation of a CLI execution, output included.
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"))
[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,
) -> Generator[str, None, None]:
"""Create a temporary directory and change the working directory to it.
Replaces {meth}`click.testing.CliRunner.isolated_filesystem`, which Click
deprecates 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, where Click's version
would raise a deprecation warning each time.
```{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.
"""
target = tempfile.mkdtemp(dir=temp_dir)
try:
with chdir(target):
yield target
finally:
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
"""Default of the `force_color` argument of {meth}`invoke`, for every run.
A call passing `color=False` still gets uncolored output.
"""
[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 | None = None,
color: bool | None = None,
force_color: bool | 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).
- Adds `force_color`, which keeps colored output while initializing
`Context.color = True`. `click.testing.CliRunner.invoke()` cannot express
it, its `color` parameter colliding with the `Context` one
([pallets/click#2110](https://github.com/pallets/click/issues/2110)).
The {attr}`force_color` class attribute sets it for every run.
- 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`,
{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()`:
`None` (the default) follows the runner's own `catch_exceptions`.
:param color: whether the captured streams keep ANSI codes, passed to
`click.testing.CliRunner.isolation()`. `False` also scrubs the
result bytes.
:param force_color: keep the ANSI codes *and* initialize the invoked
CLI's `Context.color` to `True`, so it takes its colored branch.
`None` (the default) follows the {attr}`force_color` class
attribute.
:param extra: same as `click.testing.CliRunner.invoke()`, but colliding
parameters are allowed and properly passed on to the invoked CLI.
:raises ValueError: if `force_color=True` is passed with `color=False`.
"""
# 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 force_color is None:
# An explicit color=False on one call wins over the class default.
force_color = self.force_color and color is not False
elif force_color and color is False:
raise ValueError("color=False contradicts force_color=True.")
if force_color:
# `CliRunner.invoke(color=True)` controls the test "terminal" but
# cannot pass `color` through to `Context` as well: send it as an
# extra parameter of the invoked CLI (pallets/click#2110).
extra["color"] = True
isolation_color = force_color or bool(color)
# 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)
# An extra parameter named like one of `invoke()`'s own cannot travel
# through `**extra`: take it out, and hand it to the `Context` through a
# patched `main()` call instead.
bypass = {name: extra.pop(name) for name in INVOKE_ARGS.intersection(extra)}
main_patch: AbstractContextManager = (
patch_attr(cli, "main", partial(cli.main, **bypass))
if bypass
else nullcontext()
)
with main_patch:
result = super().invoke(
cli=cli,
args=clean_args,
input=input,
env=env,
catch_exceptions=catch_exceptions,
color=isolation_color,
**extra,
)
# Upgrade the result to the subclass formatting its traceback.
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)
extra_result.stderr_bytes = strip_ansi(extra_result.stderr_bytes)
extra_result.output_bytes = strip_ansi(extra_result.output_bytes)
_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
) -> Generator[str, None, None]:
"""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)
[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 number, (regex_line, content_line) in enumerate(
zip_longest(regex_lines, content_lines, fillvalue=""), start=1
):
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 not matched:
raise RegexLineMismatch(regex_line, content_line, number)
logging.debug(
"Line #%d match.\nRegex : %r\nOutput: %r", number, regex_line, content_line
)