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
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
|
import os
import subprocess
import sys
import timeit
import pytest
import procrunner
def test_simple_command_invocation():
if os.name == "nt":
command = ["cmd.exe", "/c", "echo", "hello"]
else:
command = ["echo", "hello"]
result = procrunner.run(command)
assert result.returncode == 0
assert result.stdout == b"hello" + os.linesep.encode("utf-8")
assert result.stderr == b""
def test_simple_command_invocation_with_closed_stdin():
if os.name == "nt":
command = ["cmd.exe", "/c", "echo", "hello"]
else:
command = ["echo", "hello"]
result = procrunner.run(command, stdin=subprocess.DEVNULL)
assert result.returncode == 0
assert result.stdout == b"hello" + os.linesep.encode("utf-8")
assert result.stderr == b""
def test_decode_invalid_utf8_input(capsys):
test_string = b"test\xa0string\n"
if os.name == "nt":
pytest.xfail("Test requires stdin feature which does not work on Windows")
command = ["cmd.exe", "/c", "type", "CON"]
else:
command = ["cat"]
result = procrunner.run(command, stdin=test_string)
assert result.returncode == 0
assert not result.stderr
if os.name == "nt":
# Windows modifies line endings
assert result.stdout == test_string[:-1] + b"\r\n"
else:
assert result.stdout == test_string
out, err = capsys.readouterr()
assert out == "test\ufffdstring\n"
assert err == ""
def test_running_wget(tmp_path):
command = ["wget", "https://www.google.com", "-O", "-"]
try:
result = procrunner.run(command, working_directory=tmp_path)
except OSError as e:
if e.errno == 2:
pytest.skip("wget not available")
raise
assert result.returncode == 0
assert b"http" in result.stderr
assert b"google" in result.stdout
def test_path_object_resolution(tmp_path):
sentinel_value = b"sentinel"
tmp_path.joinpath("tempfile").write_bytes(sentinel_value)
tmp_path.joinpath("reader.py").write_text(
"with open('tempfile') as fh:\n print(fh.read())"
)
assert "LEAK_DETECTOR" not in os.environ
result = procrunner.run(
[sys.executable, tmp_path / "reader.py"],
environment_override={"PYTHONHASHSEED": "random", "LEAK_DETECTOR": "1"},
working_directory=tmp_path,
)
assert result.returncode == 0
assert not result.stderr
assert sentinel_value == result.stdout.strip()
assert (
"LEAK_DETECTOR" not in os.environ
), "overridden environment variable leaked into parent process"
def test_timeout_behaviour_legacy(tmp_path):
start = timeit.default_timer()
try:
with pytest.warns(DeprecationWarning, match="timeout"):
result = procrunner.run(
[sys.executable, "-c", "import time; time.sleep(5)"],
timeout=0.1,
working_directory=tmp_path,
raise_timeout_exception=False,
)
except RuntimeError:
# This test sometimes fails with a RuntimeError.
runtime = timeit.default_timer() - start
assert runtime < 3
return
runtime = timeit.default_timer() - start
with pytest.warns(DeprecationWarning, match="\\.timeout"):
assert result.timeout
assert runtime < 3
assert not result.stdout
assert not result.stderr
assert result.returncode
def test_timeout_behaviour(tmp_path):
command = (sys.executable, "-c", "import time; time.sleep(5)")
start = timeit.default_timer()
try:
with pytest.raises(subprocess.TimeoutExpired) as te:
procrunner.run(
command,
timeout=0.1,
working_directory=tmp_path,
raise_timeout_exception=True,
)
except RuntimeError:
# This test sometimes fails with a RuntimeError.
runtime = timeit.default_timer() - start
assert runtime < 3
return
runtime = timeit.default_timer() - start
assert runtime < 3
assert te.value.stdout == b""
assert te.value.stderr == b""
assert te.value.timeout == 0.1
assert te.value.cmd == command
def test_argument_deprecation(tmp_path):
with pytest.warns(DeprecationWarning, match="keyword arguments"):
result = procrunner.run(
[sys.executable, "-V"],
None,
working_directory=tmp_path,
)
assert not result.returncode
assert result.stderr or result.stdout
|
