# 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.
"""Fixtures, configuration and helpers for tests."""
from __future__ import annotations
import io
import os
import subprocess
import threading
import time
import click
import pytest
import requests
from extra_platforms.pytest import skip_windows
from click_extra.color import COLOR_ENVVARS
from click_extra.envvar import temporary_env
from click_extra.pytest import ( # noqa: F401
assert_output_regex,
create_config,
invoke,
runner,
)
from click_extra.theme import THEME_ENVVAR
TYPE_CHECKING = False
if TYPE_CHECKING:
from collections.abc import Callable, Iterator
[docs]
@pytest.fixture(scope="session")
def httpserver_listen_address():
"""Bind the local HTTP server to the loopback address, not to a name.
Overrides ``pytest-httpserver``'s default of ``("localhost", 0)``. Resolving
that name is the one thing the ``tests/test_config.py`` cases serving a
configuration file over HTTP need from the host, and a build sandbox is
exactly where it is unavailable: the Nix one on macOS denies the lookup, so
every one of them errors out with ``socket.gaierror: [Errno 8] nodename nor
servname provided, or not known``.
Binding the literal address asks nothing of the resolver. Whether a sandbox
additionally gates the loopback socket is its own policy, and a separate
question: this only removes the lookup that failed first.
"""
return ("127.0.0.1", 0)
[docs]
@pytest.fixture(scope="session")
def requires_process_table():
"""Skip a test where the process table does not list the test's own process.
The kill paths of `run_cli` find a command's descendants in the process table.
Linux reads it from `/proc`. Elsewhere it comes from `ps`, which a build
sandbox can deny: the MacPorts test phase gets an empty listing, so the kill
tests would fail on what the sandbox hides rather than on the code.
The probe runs `ps` itself instead of `_posix_process_table()`, so a
regression that empties the table still fails those tests.
"""
if os.path.exists("/proc/self/stat"):
return
try:
listing = subprocess.run(
("ps", "-A", "-o", "pid="),
capture_output=True,
encoding="UTF-8",
timeout=10,
check=False,
)
except (OSError, subprocess.TimeoutExpired):
pytest.skip("No ps command answers here.")
if str(os.getpid()) not in listing.stdout.split():
pytest.skip("The process table does not list the test process.")
@pytest.fixture(autouse=True)
def _isolate_color_envvars():
"""Remove output-affecting environment variables so tests are deterministic.
Variables like ``NO_COLOR`` and ``LLM`` are commonly set by shells, editors,
and AI tooling. Their presence overrides ``ColorOption``'s default, making
color-dependent tests fail in developer environments. ``ACCESSIBLE`` is
isolated for the same reason: it lowers the ``--color`` and ``--table-format``
defaults. ``CLICK_EXTRA_THEME`` is meant to be exported from a shell profile,
so a developer running the suite is precisely who is likely to have it set,
and it would repaint every help screen the assertions pin.
"""
with temporary_env(unset_vars=(*COLOR_ENVVARS, "ACCESSIBLE", THEME_ENVVAR)):
yield
skip_windows_colors = skip_windows(reason="Click overstrip colors on Windows")
"""Skips color tests on Windows as ``click.testing.invoke`` overzealously strips colors.
See:
- https://github.com/pallets/click/issues/2111
- https://github.com/pallets/click/issues/2110
"""
#: Status codes a host answers with when it is refusing *this* request rather
#: than reporting something about the resource: a rate limit, a proxy hiccup,
#: or a service that is briefly down.
TRANSIENT_STATUS_CODES = frozenset({408, 425, 429, 500, 502, 503, 504})
[docs]
def fetch_or_skip(url: str, timeout: float = 60) -> requests.Response:
"""Fetch `url`, skipping the test when the failure says nothing about it.
A network-dependent test asserts something about what a host serves. It
cannot assert it while the host is rate-limiting, timing out, or down, and
a bare `assert response.ok` there reports `assert False`: a red run that
looks like the finding the test exists to make, with nothing naming the
cause. GitHub throttles anonymous archive downloads, so a full-suite run
hits this on its own schedule and reads as an order-dependent flake.
A response that *is* about the resource still fails, loudly and with its
status: a 404 means the URL these tests build no longer resolves, which is
the finding, not the weather.
"""
try:
response = requests.get(url, timeout=timeout)
except requests.RequestException as error:
pytest.skip(f"cannot reach {url}: {error}")
if response.status_code in TRANSIENT_STATUS_CODES:
pytest.skip(f"{url} answered {response.status_code} {response.reason}")
assert response.ok, f"{url} answered {response.status_code} {response.reason}"
return response
[docs]
def walk_commands(
command: click.Command,
ctx: click.Context | None = None,
path: tuple[str, ...] = (),
) -> Iterator[tuple[tuple[str, ...], click.Command]]:
"""Yield every `(path, command)` pair under `command`, itself included.
`path` holds the subcommand names leading to the command, the root's own
name excluded, so joining it names the invocation a user would type.
:param command: the command to walk, a group or a leaf.
:param ctx: the context `command` is looked up in. Built from the command
itself when omitted, which is what a walk starting at the root wants.
:param path: the names already walked through, for the recursion.
"""
if ctx is None:
ctx = click.Context(command, info_name=command.name)
yield path, command
if isinstance(command, click.Group):
for name in command.list_commands(ctx):
sub = command.get_command(ctx, name)
if sub is None:
continue
sub_ctx = click.Context(sub, parent=ctx, info_name=name)
yield from walk_commands(sub, sub_ctx, (*path, name))
[docs]
class TTYStringIO(io.StringIO):
"""An in-memory text buffer that claims to be an interactive terminal."""
[docs]
def isatty(self) -> bool:
return True
[docs]
class StallingTTY(TTYStringIO):
"""A terminal whose writes block once `stall` is set, until `release` is.
Stands in for a terminal paused by flow control (Ctrl+S), or a full pipe.
"""
def __init__(self) -> None:
super().__init__()
self.stall = threading.Event()
self.blocked = threading.Event()
self.release = threading.Event()
[docs]
def write(self, text: str) -> int:
if self.stall.is_set() and not self.release.is_set():
self.blocked.set()
self.release.wait()
return super().write(text)
[docs]
def wait_until(predicate: Callable[[], bool], timeout: float = 3.0) -> bool:
"""Poll ``predicate`` until it is true or ``timeout`` seconds elapse.
Lets thread-driven assertions wait for an outcome instead of sleeping a fixed
(and racy) amount.
"""
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
if predicate():
return True
time.sleep(0.01)
return predicate()