testutils

Some utilities for testing.

This module is not intended to be imported by users of Tripper.

serverskip(server, port, timeout=2, exitcode=44)

Ping a remote server and calls sys.exit() with given exitcode if it doesn't respond within the timeout.

Source code in tripper/testutils.py

def serverskip(
    server: str, port: int, timeout: float = 2, exitcode: int = 44
) -> None:
    """Ping a remote server and calls `sys.exit()` with given exitcode if it
    doesn't respond within the timeout."""
    try:
        with socket.create_connection((server, port), timeout=timeout):
            pass
    except OSError as exc:
        print(  # noqa: T201
            f"Server {server}:{port} seems to be down: {exc}", file=sys.stderr
        )
        sys.exit(exitcode)

service_available(service, timeout=2)

Return true if the given test service can be contacted, otherwise false.

Source code in tripper/testutils.py

def service_available(service: str, timeout: float = 2) -> bool:
    """Return true if the given test service can be contacted, otherwise
    false."""
    config = service_config(service)
    try:
        with socket.create_connection(
            (config["host"], config["port"]), timeout=timeout
        ):
            pass
    except OSError:
        return False
    return True

service_config(service)

Return a dict with the configuration of the given test service.

Parameters:

Name	Type	Description	Default
service	str	Name of the service. Should be a key in TEST_SERVICES.	required

Returns:

Type	Description
dict	Dict with the default configuration in TEST_SERVICES, where each value may be overridden by the environment variable TRIPPER_TEST__.

Source code in tripper/testutils.py

def service_config(service: str) -> dict:
    """Return a dict with the configuration of the given test service.

    Arguments:
        service: Name of the service.  Should be a key in `TEST_SERVICES`.

    Returns:
        Dict with the default configuration in `TEST_SERVICES`, where
        each value may be overridden by the environment variable
        TRIPPER_TEST_<SERVICE>_<KEY>.
    """
    config = {}
    prefix = f"TRIPPER_TEST_{service.upper()}_"
    for key, default in TEST_SERVICES[service].items():
        value = os.environ.get(prefix + key.upper(), default)
        config[key] = type(default)(value)  # type: ignore [call-arg]
    return config

service_responds(url, timeout=1.0)

Return whether the service with given URL responds with status 200.

Unlike check_service_availability(), this makes a single request without retrying. It is therefore fast when the service is not running, but doesn't wait for a service that is starting up.

Parameters:

Name	Type	Description	Default
url	str	URL of the service to check.	required
timeout	float	Time in seconds to wait for a response.	1.0

Returns:

Type	Description
bool	Returns true if the service responds with code 200, otherwise false is returned. False is also returned if requests is not installed.

Source code in tripper/testutils.py

def service_responds(url: str, timeout: float = 1.0) -> bool:
    """Return whether the service with given URL responds with status 200.

    Unlike `check_service_availability()`, this makes a single request
    without retrying.  It is therefore fast when the service is not
    running, but doesn't wait for a service that is starting up.

    Arguments:
        url: URL of the service to check.
        timeout: Time in seconds to wait for a response.

    Returns:
        Returns true if the service responds with code 200, otherwise
        false is returned.  False is also returned if `requests` is not
        installed.
    """
    try:
        import requests
    except ImportError:
        return False

    try:
        return requests.get(url, timeout=timeout).status_code == 200
    except requests.exceptions.RequestException:
        return False

serviceskip(service, timeout=2, exitcode=44)

Call sys.exit() with the given exit code if the given test service cannot be contacted. Otherwise return its configuration (see service_config()).

Source code in tripper/testutils.py

def serviceskip(service: str, timeout: float = 2, exitcode: int = 44) -> dict:
    """Call `sys.exit()` with the given exit code if the given test
    service cannot be contacted.  Otherwise return its configuration
    (see `service_config()`)."""
    config = service_config(service)
    serverskip(
        config["host"], config["port"], timeout=timeout, exitcode=exitcode
    )
    return config