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
|
from __future__ import annotations
import sys
from argparse import ArgumentTypeError
from pathlib import Path
from signal import SIGINT
from subprocess import PIPE, Popen
from time import sleep
import pytest
from pytest_mock import MockerFixture
from tox.pytest import MonkeyPatch, ToxProjectCreator
from tox.session.cmd.run.parallel import parse_num_processes
from tox.tox_env.api import ToxEnv
from tox.tox_env.errors import Fail
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":
raise Fail("something bad happened")
return prev_setup(self)
prev_setup = ToxEnv._setup_env
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")
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("{str(marker)}"); 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():
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
|
