Gecko [ betsfootball.net ]

Name	Size	Permission	Date
__pycache__	[ DIR ]	drwxr-xr-x	2025-04-08 17:00
externals	[ DIR ]	drwxr-xr-x	2025-04-08 17:00
test	[ DIR ]	drwxr-xr-x	2025-04-08 17:00
__init__.py	5.01 KB	-rw-r--r--	2025-04-08 17:00
_cloudpickle_wrapper.py	417 B	-rw-r--r--	2025-04-08 17:00
_dask.py	13 KB	-rw-r--r--	2025-04-08 17:00
_memmapping_reducer.py	27.43 KB	-rw-r--r--	2025-04-08 17:00
_multiprocessing_helpers.py	1.88 KB	-rw-r--r--	2025-04-08 17:00
_parallel_backends.py	24.89 KB	-rw-r--r--	2025-04-08 17:00
_store_backends.py	16.29 KB	-rw-r--r--	2025-04-08 17:00
_utils.py	2.03 KB	-rw-r--r--	2025-04-08 17:00
backports.py	5.24 KB	-rw-r--r--	2025-04-08 17:00
compressor.py	19.3 KB	-rw-r--r--	2025-04-08 17:00
disk.py	4.29 KB	-rw-r--r--	2025-04-08 17:00
executor.py	5.02 KB	-rw-r--r--	2025-04-08 17:00
func_inspect.py	13.87 KB	-rw-r--r--	2025-04-08 17:00
hashing.py	10.29 KB	-rw-r--r--	2025-04-08 17:00
logger.py	5.33 KB	-rw-r--r--	2025-04-08 17:00
memory.py	45.45 KB	-rw-r--r--	2025-04-08 17:00
numpy_pickle.py	26.26 KB	-rw-r--r--	2025-04-08 17:00
numpy_pickle_compat.py	8.35 KB	-rw-r--r--	2025-04-08 17:00
numpy_pickle_utils.py	8.52 KB	-rw-r--r--	2025-04-08 17:00
parallel.py	82.6 KB	-rw-r--r--	2025-04-08 17:00
pool.py	14.08 KB	-rw-r--r--	2025-04-08 17:00
testing.py	3.02 KB	-rw-r--r--	2025-04-08 17:00

Rename

"""
Helper for testing.
"""

import sys
import warnings
import os.path
import re
import subprocess
import threading

import pytest
import _pytest

raises = pytest.raises
warns = pytest.warns
SkipTest = _pytest.runner.Skipped
skipif = pytest.mark.skipif
fixture = pytest.fixture
parametrize = pytest.mark.parametrize
timeout = pytest.mark.timeout
xfail = pytest.mark.xfail
param = pytest.param

def warnings_to_stdout():
    """ Redirect all warnings to stdout.
    """
    showwarning_orig = warnings.showwarning

def showwarning(msg, cat, fname, lno, file=None, line=0):
        showwarning_orig(msg, cat, os.path.basename(fname), line, sys.stdout)

warnings.showwarning = showwarning
    # warnings.simplefilter('always')

def check_subprocess_call(cmd, timeout=5, stdout_regex=None,
                          stderr_regex=None):
    """Runs a command in a subprocess with timeout in seconds.

A SIGTERM is sent after `timeout` and if it does not terminate, a
    SIGKILL is sent after `2 * timeout`.

Also checks returncode is zero, stdout if stdout_regex is set, and
    stderr if stderr_regex is set.
    """
    proc = subprocess.Popen(cmd, stdout=subprocess.PIPE,
                            stderr=subprocess.PIPE)

def terminate_process():  # pragma: no cover
        """
        Attempt to terminate a leftover process spawned during test execution:
        ideally this should not be needed but can help avoid clogging the CI
        workers in case of deadlocks.
        """
        warnings.warn(f"Timeout running {cmd}")
        proc.terminate()

def kill_process():  # pragma: no cover
        """
        Kill a leftover process spawned during test execution: ideally this
        should not be needed but can help avoid clogging the CI workers in
        case of deadlocks.
        """
        warnings.warn(f"Timeout running {cmd}")
        proc.kill()

try:
        if timeout is not None:
            terminate_timer = threading.Timer(timeout, terminate_process)
            terminate_timer.start()
            kill_timer = threading.Timer(2 * timeout, kill_process)
            kill_timer.start()
        stdout, stderr = proc.communicate()
        stdout, stderr = stdout.decode(), stderr.decode()
        if proc.returncode != 0:
            message = (
                'Non-zero return code: {}.\nStdout:\n{}\n'
                'Stderr:\n{}').format(
                    proc.returncode, stdout, stderr)
            raise ValueError(message)

if (stdout_regex is not None and
                not re.search(stdout_regex, stdout)):
            raise ValueError(
                "Unexpected stdout: {!r} does not match:\n{!r}".format(
                    stdout_regex, stdout))
        if (stderr_regex is not None and
                not re.search(stderr_regex, stderr)):
            raise ValueError(
                "Unexpected stderr: {!r} does not match:\n{!r}".format(
                    stderr_regex, stderr))

finally:
        if timeout is not None:
            terminate_timer.cancel()
            kill_timer.cancel()