Aluode/PerceptionLabPortable

1"""2Helper for testing.3"""4 5import os.path6import re7import subprocess8import sys9import threading10import warnings11 12import _pytest13import pytest14 15raises = pytest.raises16warns = pytest.warns17SkipTest = _pytest.runner.Skipped18skipif = pytest.mark.skipif19fixture = pytest.fixture20parametrize = pytest.mark.parametrize21timeout = pytest.mark.timeout22xfail = pytest.mark.xfail23param = pytest.param24 25 26def warnings_to_stdout():27    """Redirect all warnings to stdout."""28    showwarning_orig = warnings.showwarning29 30    def showwarning(msg, cat, fname, lno, file=None, line=0):31        showwarning_orig(msg, cat, os.path.basename(fname), line, sys.stdout)32 33    warnings.showwarning = showwarning34    # warnings.simplefilter('always')35 36 37def check_subprocess_call(cmd, timeout=5, stdout_regex=None, stderr_regex=None):38    """Runs a command in a subprocess with timeout in seconds.39 40    A SIGTERM is sent after `timeout` and if it does not terminate, a41    SIGKILL is sent after `2 * timeout`.42 43    Also checks returncode is zero, stdout if stdout_regex is set, and44    stderr if stderr_regex is set.45    """46    proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)47 48    def terminate_process():  # pragma: no cover49        """50        Attempt to terminate a leftover process spawned during test execution:51        ideally this should not be needed but can help avoid clogging the CI52        workers in case of deadlocks.53        """54        warnings.warn(f"Timeout running {cmd}")55        proc.terminate()56 57    def kill_process():  # pragma: no cover58        """59        Kill a leftover process spawned during test execution: ideally this60        should not be needed but can help avoid clogging the CI workers in61        case of deadlocks.62        """63        warnings.warn(f"Timeout running {cmd}")64        proc.kill()65 66    try:67        if timeout is not None:68            terminate_timer = threading.Timer(timeout, terminate_process)69            terminate_timer.start()70            kill_timer = threading.Timer(2 * timeout, kill_process)71            kill_timer.start()72        stdout, stderr = proc.communicate()73        stdout, stderr = stdout.decode(), stderr.decode()74        if proc.returncode != 0:75            message = ("Non-zero return code: {}.\nStdout:\n{}\nStderr:\n{}").format(76                proc.returncode, stdout, stderr77            )78            raise ValueError(message)79 80        if stdout_regex is not None and not re.search(stdout_regex, stdout):81            raise ValueError(82                "Unexpected stdout: {!r} does not match:\n{!r}".format(83                    stdout_regex, stdout84                )85            )86        if stderr_regex is not None and not re.search(stderr_regex, stderr):87            raise ValueError(88                "Unexpected stderr: {!r} does not match:\n{!r}".format(89                    stderr_regex, stderr90                )91            )92 93    finally:94        if timeout is not None:95            terminate_timer.cancel()96            kill_timer.cancel()97