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
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
|
from __future__ import annotations
import sys
from argparse import ArgumentTypeError
from signal import SIGINT
from subprocess import PIPE, Popen
from time import sleep
from typing import TYPE_CHECKING
from unittest import mock
import pytest
from tox.session.cmd.run import parallel
from tox.session.cmd.run.parallel import parse_num_processes
from tox.tox_env.api import ToxEnv
from tox.tox_env.errors import Fail
from tox.util.cpu import auto_detect_cpus
if TYPE_CHECKING:
from pathlib import Path
from pytest_mock import MockerFixture
from tox.pytest import MonkeyPatch, ToxProjectCreator
def test_parse_num_processes_all() -> None:
assert parse_num_processes("all") is None
def test_parse_num_processes_auto() -> None:
auto = parse_num_processes("auto")
assert isinstance(auto, int)
assert auto > 0
def test_parse_num_processes_exact() -> None:
assert parse_num_processes("3") == 3
def test_parse_num_processes_not_number() -> None:
with pytest.raises(ArgumentTypeError, match="value must be a positive number"):
parse_num_processes("3df")
def test_parse_num_processes_minus_one() -> None:
with pytest.raises(ArgumentTypeError, match="value must be positive"):
parse_num_processes("-1")
def test_parallel_general(tox_project: ToxProjectCreator, monkeypatch: MonkeyPatch, mocker: MockerFixture) -> None:
def setup(self: ToxEnv) -> None:
if self.name == "f":
msg = "something bad happened"
raise Fail(msg)
return prev_setup(self)
prev_setup = ToxEnv._setup_env # noqa: SLF001
mocker.patch.object(ToxEnv, "_setup_env", autospec=True, side_effect=setup)
monkeypatch.setenv("PATH", "")
ini = """
[tox]
no_package=true
skip_missing_interpreters = true
env_list= a, b, c, d, e, f
[testenv]
commands=python -c 'print("run {env_name}")'
depends = !c: c
parallel_show_output = c: true
[testenv:d]
base_python = missing_skip
[testenv:e]
commands=python -c 'import sys; print("run {env_name}"); sys.exit(1)'
"""
project = tox_project({"tox.ini": ini})
outcome = project.run("p", "-p", "all")
outcome.assert_failed()
out = outcome.out
oks, skips, fails = {"a", "b", "c"}, {"d"}, {"e", "f"}
missing = set()
for env in "a", "b", "c", "d", "e", "f":
if env in {"c", "e"}:
assert "run c" in out, out
elif env == "f":
assert "f: failed with something bad happened" in out, out
else:
assert f"run {env}" not in out, out
of_type = "OK" if env in oks else ("SKIP" if env in skips else "FAIL")
of_type_icon = "✔" if env in oks else ("⚠" if env in skips else "✖")
env_done = f"{env}: {of_type} {of_type_icon}"
is_missing = env_done not in out
if is_missing:
missing.add(env_done)
env_report = f" {env}: {of_type} {'code 1 ' if env in fails else ''}("
assert env_report in out, out
if not is_missing:
assert out.index(env_done) < out.index(env_report), out
assert len(missing) == 1, out
def test_parallel_run_live_out(tox_project: ToxProjectCreator) -> None:
ini = """
[tox]
no_package=true
env_list= a, b
[testenv]
commands=python -c 'print("run {env_name}")'
"""
project = tox_project({"tox.ini": ini})
outcome = project.run("p", "-p", "2", "--parallel-live")
outcome.assert_success()
assert "python -c" in outcome.out
assert "run a" in outcome.out
assert "run b" in outcome.out
def test_parallel_show_output_with_pkg(tox_project: ToxProjectCreator, demo_pkg_inline: Path) -> None:
ini = "[testenv]\nparallel_show_output=True\ncommands=python -c 'print(\"r {env_name}\")'"
project = tox_project({"tox.ini": ini})
result = project.run("p", "--root", str(demo_pkg_inline))
assert "r py" in result.out
@pytest.mark.skipif(sys.platform == "win32", reason="You need a conhost shell for keyboard interrupt")
@pytest.mark.flaky(max_runs=3, min_passes=1)
def test_keyboard_interrupt(tox_project: ToxProjectCreator, demo_pkg_inline: Path, tmp_path: Path) -> None:
marker = tmp_path / "a"
ini = f"""
[testenv]
package=wheel
commands=python -c 'from time import sleep; from pathlib import Path; \
p = Path("{marker!s}"); p.write_text(""); sleep(100)'
[testenv:dep]
depends=py
"""
proj = tox_project(
{
"tox.ini": ini,
"pyproject.toml": (demo_pkg_inline / "pyproject.toml").read_text(),
"build.py": (demo_pkg_inline / "build.py").read_text(),
},
)
cmd = ["-c", str(proj.path / "tox.ini"), "p", "-p", "1", "-e", f"py,py{sys.version_info[0]},dep"]
process = Popen([sys.executable, "-m", "tox", *cmd], stdout=PIPE, stderr=PIPE, universal_newlines=True)
while not marker.exists() and (process.poll() is None):
sleep(0.05)
process.send_signal(SIGINT)
out, err = process.communicate()
assert process.returncode != 0
assert "KeyboardInterrupt" in err, err
assert "KeyboardInterrupt - teardown started\n" in out, out
assert "interrupt tox environment: py\n" in out, out
assert "requested interrupt of" in out, out
assert "send signal SIGINT" in out, out
assert "interrupt finished with success" in out, out
assert "interrupt tox environment: .pkg" in out, out
def test_parallels_help(tox_project: ToxProjectCreator) -> None:
outcome = tox_project({"tox.ini": ""}).run("p", "-h")
outcome.assert_success()
def test_parallel_legacy_accepts_no_arg(tox_project: ToxProjectCreator) -> None:
outcome = tox_project({"tox.ini": ""}).run("-p", "-h")
outcome.assert_success()
def test_parallel_requires_arg(tox_project: ToxProjectCreator) -> None:
outcome = tox_project({"tox.ini": ""}).run("p", "-p", "-h")
outcome.assert_failed()
assert "argument -p/--parallel: expected one argument" in outcome.err
def test_parallel_no_spinner(tox_project: ToxProjectCreator) -> None:
"""Ensure passing `--parallel-no-spinner` implies `--parallel`."""
with mock.patch.object(parallel, "execute") as mocked:
tox_project({"tox.ini": ""}).run("p", "--parallel-no-spinner")
mocked.assert_called_once_with(
mock.ANY,
max_workers=auto_detect_cpus(),
has_spinner=False,
live=False,
)
def test_parallel_no_spinner_with_parallel(tox_project: ToxProjectCreator) -> None:
"""Ensure `--parallel N` is still respected with `--parallel-no-spinner`."""
with mock.patch.object(parallel, "execute") as mocked:
tox_project({"tox.ini": ""}).run("p", "--parallel-no-spinner", "--parallel", "2")
mocked.assert_called_once_with(
mock.ANY,
max_workers=2,
has_spinner=False,
live=False,
)
def test_parallel_no_spinner_ci(
tox_project: ToxProjectCreator, mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Ensure spinner is disabled by default in CI."""
mocked = mocker.patch.object(parallel, "execute")
monkeypatch.setenv("CI", "1")
tox_project({"tox.ini": ""}).run("p")
mocked.assert_called_once_with(
mock.ANY,
max_workers=auto_detect_cpus(),
has_spinner=False,
live=False,
)
def test_parallel_no_spinner_legacy(tox_project: ToxProjectCreator) -> None:
with mock.patch.object(parallel, "execute") as mocked:
tox_project({"tox.ini": ""}).run("--parallel-no-spinner")
mocked.assert_called_once_with(
mock.ANY,
max_workers=auto_detect_cpus(),
has_spinner=False,
live=False,
)
|
