File size: 2,520 Bytes
7885a28 |
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 |
import re
import sys
from joblib.testing import check_subprocess_call, raises
def test_check_subprocess_call():
code = "\n".join(
["result = 1 + 2 * 3", "print(result)", "my_list = [1, 2, 3]", "print(my_list)"]
)
check_subprocess_call([sys.executable, "-c", code])
# Now checking stdout with a regex
check_subprocess_call(
[sys.executable, "-c", code],
# Regex needed for platform-specific line endings
stdout_regex=r"7\s{1,2}\[1, 2, 3\]",
)
def test_check_subprocess_call_non_matching_regex():
code = "42"
non_matching_pattern = "_no_way_this_matches_anything_"
with raises(ValueError) as excinfo:
check_subprocess_call(
[sys.executable, "-c", code], stdout_regex=non_matching_pattern
)
excinfo.match("Unexpected stdout.+{}".format(non_matching_pattern))
def test_check_subprocess_call_wrong_command():
wrong_command = "_a_command_that_does_not_exist_"
with raises(OSError):
check_subprocess_call([wrong_command])
def test_check_subprocess_call_non_zero_return_code():
code_with_non_zero_exit = "\n".join(
[
"import sys",
'print("writing on stdout")',
'sys.stderr.write("writing on stderr")',
"sys.exit(123)",
]
)
pattern = re.compile(
"Non-zero return code: 123.+"
"Stdout:\nwriting on stdout.+"
"Stderr:\nwriting on stderr",
re.DOTALL,
)
with raises(ValueError) as excinfo:
check_subprocess_call([sys.executable, "-c", code_with_non_zero_exit])
excinfo.match(pattern)
def test_check_subprocess_call_timeout():
code_timing_out = "\n".join(
[
"import time",
"import sys",
'print("before sleep on stdout")',
"sys.stdout.flush()",
'sys.stderr.write("before sleep on stderr")',
"sys.stderr.flush()",
# We need to sleep for at least 2 * timeout seconds in case the SIGKILL
# is triggered.
"time.sleep(10)",
'print("process should have be killed before")',
"sys.stdout.flush()",
]
)
pattern = re.compile(
"Non-zero return code:.+"
"Stdout:\nbefore sleep on stdout\\s+"
"Stderr:\nbefore sleep on stderr",
re.DOTALL,
)
with raises(ValueError) as excinfo:
check_subprocess_call([sys.executable, "-c", code_timing_out], timeout=1)
excinfo.match(pattern)
|