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
147
148
149
150
151
152
|
"""Unittests."""
import subprocess
import sys
from typing import Dict
import pytest
from _pytest.capture import CaptureFixture
from subprocess_tee import run
def test_run_string() -> None:
"""Valida run() called with a single string command."""
cmd = "echo 111 && >&2 echo 222"
old_result = subprocess.run(
cmd,
shell=True,
universal_newlines=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
check=False,
)
result = run(cmd)
assert result.returncode == old_result.returncode
assert result.stdout == old_result.stdout
assert result.stderr == old_result.stderr
def test_run_list() -> None:
"""Validate run call with a command made of list of strings."""
# NOTICE: subprocess.run() does fail to capture any output when cmd is
# a list and you specific shell=True. Still, when not mentioning shell,
# it does work.
cmd = [sys.executable, "--version"]
old_result = subprocess.run(
cmd,
# shell=True,
universal_newlines=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
check=False,
)
result = run(cmd)
assert result.returncode == old_result.returncode
assert result.stdout == old_result.stdout
assert result.stderr == old_result.stderr
def test_run_echo(capsys: CaptureFixture[str]) -> None:
"""Validate run call with echo dumps command."""
cmd = [sys.executable, "--version"]
old_result = subprocess.run(
cmd,
# shell=True,
universal_newlines=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
check=False,
)
result = run(cmd, echo=True)
assert result.returncode == old_result.returncode
assert result.stdout == old_result.stdout
assert result.stderr == old_result.stderr
out, err = capsys.readouterr()
assert out.startswith("COMMAND:")
assert err == ""
@pytest.mark.parametrize(
"env",
[{}, {"SHELL": "/bin/sh"}, {"SHELL": "/bin/bash"}, {"SHELL": "/bin/zsh"}],
ids=["auto", "sh", "bash", "zsh"],
)
def test_run_with_env(env: Dict[str, str]) -> None:
"""Validate that passing custom env to run() works."""
env["FOO"] = "BAR"
result = run("echo $FOO", env=env, echo=True)
assert result.stdout == "BAR\n"
def test_run_shell() -> None:
"""Validate run call with multiple shell commands works."""
cmd = "echo a && echo b && false || exit 4"
# "python --version"
result = run(cmd, echo=True)
assert result.returncode == 4
assert result.stdout == "a\nb\n"
def test_run_shell_undefined() -> None:
"""Validate run call with multiple shell commands works."""
cmd = "echo a && echo b && false || exit 4"
# "python --version"
result = run(cmd, echo=True, env={})
assert result.returncode == 4
assert result.stdout == "a\nb\n"
def test_run_cwd() -> None:
"""Validate that run accepts cwd and respects it."""
cmd = "pwd"
result = run(cmd, echo=True, cwd="/")
assert result.returncode == 0
assert result.stdout == "/\n"
def test_run_with_check_raise() -> None:
"""Asure compatibility with subprocess.run when using check (return 1)."""
with pytest.raises(subprocess.CalledProcessError) as ours:
run("false", check=True)
with pytest.raises(subprocess.CalledProcessError) as original:
subprocess.run("false", check=True, universal_newlines=True)
assert ours.value.returncode == original.value.returncode
assert ours.value.cmd == original.value.cmd
assert ours.value.output == original.value.output
assert ours.value.stdout == original.value.stdout
assert ours.value.stderr == original.value.stderr
def test_run_with_check_pass() -> None:
"""Asure compatibility with subprocess.run when using check (return 0)."""
ours = run("true", check=True)
original = subprocess.run("true", check=True, universal_newlines=True)
assert ours.returncode == original.returncode
assert ours.args == original.args
assert ours.stdout == original.stdout
assert ours.stderr == original.stderr
def test_run_compat() -> None:
"""Assure compatiblity with subprocess.run()."""
cmd = ["seq", "10"]
ours = run(cmd)
original = subprocess.run(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
universal_newlines=True,
check=False,
)
assert ours.returncode == original.returncode
assert ours.stdout == original.stdout
assert ours.stderr == original.stderr
assert ours.args == original.args
def test_run_waits_for_completion(tmp_path):
"""run() should always wait for the process to complete."""
tmpfile = tmp_path / "output.txt"
run(f"sleep 0.1 && echo 42 > {str(tmpfile)}")
assert tmpfile.read_text() == "42\n"
|
