|
|
""" |
|
|
Helper for testing. |
|
|
""" |
|
|
|
|
|
import os.path |
|
|
import re |
|
|
import subprocess |
|
|
import sys |
|
|
import threading |
|
|
import warnings |
|
|
|
|
|
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 |
|
|
|
|
|
|
|
|
|
|
|
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(): |
|
|
""" |
|
|
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(): |
|
|
""" |
|
|
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{}\nStderr:\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() |
|
|
|
