utils¶

def print_failure(string, **kwargs):
    """ Print but sad. """
    print(f"\033[91m\033[1m{string}\033[0m", **kwargs)

def print_notify(string, **kwargs):
    """ Print but louder. """
    print(f"\033[94m\033[1m{string}\033[0m", **kwargs)

def print_success(string, **kwargs):
    """ Print but happy. """
    print(f"\033[92m\033[1m{string}\033[0m", **kwargs)

def print_warning(string, **kwargs):
    """ Print but angry. """
    print(f"\033[93m{string}\033[0m", **kwargs)

def test_case(test_fn: Callable[[Any], Tuple[Any, str]]):
    """Wrapper for test case functions, which pretty-prints any errors
    depending on verbosity level, collates the number and severity of
    test failures, returns the response and summary string to the caller.
    Any additional positional or keyword arguments are passed directly
    to `test_fn`. The wrapper will intercept the named arguments
    `optional`, `multistage` and `request` and interpret them according
    to the docstring for `wrapper(...)` below.

    Parameters:
        test_fn: Any function that returns an object and a message to
            print upon success. The function should raise a `ResponseError`,
            `ValidationError` or a `ManualValidationError` if the test
            case has failed. The function can return `None` to indicate
            that the test was not appropriate and should be ignored.

    """
    from functools import wraps

    @wraps(test_fn)
    def wrapper(
        validator,
        *args,
        request: str = None,
        optional: bool = False,
        multistage: bool = False,
        **kwargs,
    ):
        """Wraps a function or validator method and handles
        success, failure and output depending on the keyword
        arguments passed.

        Arguments:
            validator: The validator object to accumulate errors/counters.
            *args: Positional arguments passed to the test function.
            request: Description of the request made by the wrapped
                function (e.g. a URL or a summary).
            optional: Whether or not to treat the test as optional.
            multistage: If `True`, no output will be printed for this test,
                and it will not increment the success counter. Errors will be
                handled in the normal way. This can be used to avoid flooding
                the output for mutli-stage tests.
            **kwargs: Extra named arguments passed to the test function.

        """
        try:
            try:
                if optional and not validator.run_optional_tests:
                    result = None
                    msg = "skipping optional"
                else:
                    result, msg = test_fn(validator, *args, **kwargs)

            except (json.JSONDecodeError, ResponseError, ValidationError) as exc:
                msg = f"{exc.__class__.__name__}: {exc}"
                raise exc
            except Exception as exc:
                msg = f"{exc.__class__.__name__}: {exc}"
                raise InternalError(msg)

        # Catch SystemExit and KeyboardInterrupt explicitly so that we can pass
        # them to the finally block, where they are immediately raised
        except (Exception, SystemExit, KeyboardInterrupt) as exc:
            result = exc
            traceback = tb.format_exc()

        finally:
            # This catches the case of the Client throwing a SystemExit if the server
            # did not respond, the case of the validator "fail-fast"'ing and throwing
            # a SystemExit below, and the case of the user interrupting the process manually
            if isinstance(result, (SystemExit, KeyboardInterrupt)):
                raise result

            display_request = None
            try:
                display_request = validator.client.last_request
            except AttributeError:
                pass
            if display_request is None:
                display_request = validator.base_url
                if request is not None:
                    display_request += "/" + request

            request = display_request

            # If the result was None, return it here and ignore statuses
            if result is None:
                return result, msg

            if not isinstance(result, Exception):
                if not multistage:
                    success_type = "optional" if optional else None
                    validator.results.add_success(f"{request} - {msg}", success_type)
            else:
                request = request.replace("\n", "")
                message = msg.split("\n")
                if validator.verbosity > 1:
                    # ValidationErrors from pydantic already include very detailed errors
                    # that get duplicated in the traceback
                    if not isinstance(result, ValidationError):
                        message += traceback.split("\n")

                message = "\n".join(message)

                if isinstance(result, InternalError):
                    summary = (
                        f"{request} - {test_fn.__name__} - failed with internal error"
                    )
                    failure_type = "internal"
                else:
                    summary = f"{request} - {test_fn.__name__} - failed with error"
                    failure_type = "optional" if optional else None

                validator.results.add_failure(
                    summary, message, failure_type=failure_type
                )

                # set failure result to None as this is expected by other functions
                result = None

                if validator.fail_fast and not optional:
                    validator.print_summary()
                    raise SystemExit

            # Reset the client request so that it can be properly
            # displayed if the next request fails
            if not multistage:
                validator.client.last_request = None

            return result, msg

    return wrapper