Spaces:
Running
on
Zero
Running
on
Zero
File size: 3,093 Bytes
d1ed09d |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 |
"""
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()
|