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
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
|
"""Common functionality shared across multiple type of runs"""
from __future__ import annotations
import logging
import os
import random
import sys
import time
from argparse import Action, ArgumentError, ArgumentParser, Namespace
from concurrent.futures import CancelledError, Future, ThreadPoolExecutor, as_completed
from pathlib import Path
from signal import SIGINT, Handlers, signal
from threading import Event, Thread
from typing import Any, Iterator, Optional, Sequence, cast
from colorama import Fore
from tox.config.types import EnvList
from tox.execute import Outcome
from tox.journal import write_journal
from tox.session.cmd.run.single import ToxEnvRunResult, run_one
from tox.session.state import State
from tox.tox_env.api import ToxEnv
from tox.tox_env.runner import RunToxEnv
from tox.util.graph import stable_topological_sort
from tox.util.spinner import MISS_DURATION, Spinner
class SkipMissingInterpreterAction(Action):
def __call__(
self,
parser: ArgumentParser, # noqa: U100
namespace: Namespace,
values: str | Sequence[Any] | None,
option_string: str | None = None, # noqa: U100
) -> None:
value = "true" if values is None else values
if value not in ("config", "true", "false"):
raise ArgumentError(self, f"value must be 'config', 'true', or 'false' (got {value!r})")
setattr(namespace, self.dest, value)
class InstallPackageAction(Action):
def __call__(
self,
parser: ArgumentParser, # noqa: U100
namespace: Namespace,
values: str | Sequence[Any] | None,
option_string: str | None = None, # noqa: U100
) -> None:
if not values:
raise ArgumentError(self, "cannot be empty")
path = Path(cast(str, values)).absolute()
if not path.exists():
raise ArgumentError(self, f"{path} does not exist")
if not path.is_file():
raise ArgumentError(self, f"{path} is not a file")
setattr(namespace, self.dest, path)
def env_run_create_flags(parser: ArgumentParser, mode: str) -> None:
# mode can be one of: run, run-parallel, legacy, devenv, config
if mode not in ("config", "depends"):
parser.add_argument(
"--result-json",
dest="result_json",
metavar="path",
of_type=Path,
default=None,
help="write a JSON file with detailed information about all commands and results involved",
)
if mode not in ("devenv", "depends"):
parser.add_argument(
"-s",
"--skip-missing-interpreters",
default="config",
metavar="v",
nargs="?",
action=SkipMissingInterpreterAction,
help="don't fail tests for missing interpreters: {config,true,false} choice",
)
if mode not in ("devenv", "config", "depends"):
parser.add_argument(
"-n",
"--notest",
dest="no_test",
help="do not run the test commands",
action="store_true",
)
parser.add_argument(
"-b",
"--pkg-only",
"--sdistonly",
action="store_true",
help="only perform the packaging activity",
dest="package_only",
)
parser.add_argument(
"--installpkg",
help="use specified package for installation into venv, instead of packaging the project",
default=None,
of_type=Optional[Path],
action=InstallPackageAction,
dest="install_pkg",
)
if mode not in ("devenv", "depends"):
parser.add_argument(
"--develop",
action="store_true",
help="install package in development mode",
dest="develop",
)
if mode not in ("depends",):
class SeedAction(Action):
def __call__(
self,
parser: ArgumentParser, # noqa: U100
namespace: Namespace,
values: str | Sequence[Any] | None,
option_string: str | None = None, # noqa: U100
) -> None:
if values == "notset":
result = None
else:
try:
result = int(cast(str, values))
if result <= 0:
raise ValueError("must be greater than zero")
except ValueError as exc:
raise ArgumentError(self, str(exc))
setattr(namespace, self.dest, result)
parser.add_argument(
"--hashseed",
metavar="SEED",
help="set PYTHONHASHSEED to SEED before running commands. Defaults to a random integer in the range "
"[1, 4294967295] ([1, 1024] on Windows). Passing 'noset' suppresses this behavior.",
action=SeedAction,
of_type=Optional[int],
default=random.randint(1, 1024 if sys.platform == "win32" else 4294967295),
dest="hash_seed",
)
parser.add_argument(
"--discover",
dest="discover",
nargs="+",
metavar="path",
help="for Python discovery first try the Python executables under these paths",
default=[],
)
if mode not in ("depends",):
parser.add_argument(
"--no-recreate-pkg",
dest="no_recreate_pkg",
help="if recreate is set do not recreate packaging tox environment(s)",
action="store_true",
)
if mode not in ("devenv", "config", "depends"):
parser.add_argument(
"--skip-pkg-install",
dest="skip_pkg_install",
help="skip package installation for this run",
action="store_true",
)
def report(start: float, runs: list[ToxEnvRunResult], is_colored: bool, verbosity: int) -> int:
def _print(color_: int, message: str) -> None:
if verbosity:
print(f"{color_ if is_colored else ''}{message}{Fore.RESET if is_colored else ''}")
successful, skipped = [], []
for run in runs:
successful.append(run.code == Outcome.OK or run.ignore_outcome)
skipped.append(run.skipped)
duration_individual = [o.elapsed for o in run.outcomes] if verbosity >= 2 else []
extra = f"+cmd[{','.join(f'{i:.2f}' for i in duration_individual)}]" if duration_individual else ""
setup = run.duration - sum(duration_individual)
msg, color = _get_outcome_message(run)
out = f" {run.name}: {msg} ({run.duration:.2f}{f'=setup[{setup:.2f}]{extra}' if extra else ''} seconds)"
_print(color, out)
duration = time.monotonic() - start
all_good = all(successful) and not all(skipped)
if all_good:
_print(Fore.GREEN, f" congratulations :) ({duration:.2f} seconds)")
return Outcome.OK
_print(Fore.RED, f" evaluation failed :( ({duration:.2f} seconds)")
if len(runs) == 1:
return runs[0].code if not runs[0].skipped else -1
return -1
def _get_outcome_message(run: ToxEnvRunResult) -> tuple[str, int]:
if run.skipped:
msg, color = "SKIP", Fore.YELLOW
elif run.code == Outcome.OK:
msg, color = "OK", Fore.GREEN
else:
if run.ignore_outcome:
msg, color = f"IGNORED FAIL code {run.code}", Fore.YELLOW
else:
msg, color = f"FAIL code {run.code}", Fore.RED
return msg, color
logger = logging.getLogger(__name__)
def execute(state: State, max_workers: int | None, has_spinner: bool, live: bool) -> int:
interrupt, done = Event(), Event()
results: list[ToxEnvRunResult] = []
future_to_env: dict[Future[ToxEnvRunResult], ToxEnv] = {}
state.envs.ensure_only_run_env_is_active()
to_run_list: list[str] = list(state.envs.iter())
for name in to_run_list:
cast(RunToxEnv, state.envs[name]).mark_active()
previous, has_previous = None, False
try:
spinner = ToxSpinner(has_spinner, state, len(to_run_list))
thread = Thread(
target=_queue_and_wait,
name="tox-interrupt",
args=(state, to_run_list, results, future_to_env, interrupt, done, max_workers, spinner, live),
)
thread.start()
try:
thread.join()
except KeyboardInterrupt:
previous, has_previous = signal(SIGINT, Handlers.SIG_IGN), True
spinner.print_report = False # no need to print reports at this point, final report coming up
logger.error(f"[{os.getpid()}] KeyboardInterrupt - teardown started")
interrupt.set()
# cancel in reverse order to not allow submitting new jobs as we cancel running ones
for future, tox_env in reversed(list(future_to_env.items())):
cancelled = future.cancel()
# if cannot be cancelled and not done -> still runs
if cancelled is False and not future.done(): # pragma: no branch
tox_env.interrupt()
done.wait()
# workaround for https://bugs.python.org/issue45274
lock = getattr(thread, "_tstate_lock", None)
if lock is not None and lock.locked(): # pragma: no branch
lock.release() # pragma: no cover
thread._stop() # type: ignore # pragma: no cover # calling private method to fix thread state
thread.join()
finally:
ordered_results: list[ToxEnvRunResult] = []
name_to_run = {r.name: r for r in results}
for env in to_run_list:
ordered_results.append(name_to_run[env])
# write the journal
write_journal(getattr(state.conf.options, "result_json", None), state._journal)
# report the outcome
exit_code = report(
state.conf.options.start,
ordered_results,
state.conf.options.is_colored,
state.conf.options.verbosity,
)
if has_previous:
signal(SIGINT, previous)
return exit_code
class ToxSpinner(Spinner):
def __init__(self, enabled: bool, state: State, total: int) -> None:
super().__init__(
enabled=enabled,
colored=state.conf.options.is_colored,
stream=state._options.log_handler.stdout,
total=total,
)
def update_spinner(self, result: ToxEnvRunResult, success: bool) -> None:
if success:
done = self.skip if result.skipped else self.succeed
else:
done = self.fail
done(result.name)
def _queue_and_wait(
state: State,
to_run_list: list[str],
results: list[ToxEnvRunResult],
future_to_env: dict[Future[ToxEnvRunResult], ToxEnv],
interrupt: Event,
done: Event,
max_workers: int | None,
spinner: ToxSpinner,
live: bool,
) -> None:
try:
options = state._options
with spinner:
max_workers = len(to_run_list) if max_workers is None else max_workers
completed: set[str] = set()
envs_to_run_generator = ready_to_run_envs(state, to_run_list, completed)
def _run(tox_env: RunToxEnv) -> ToxEnvRunResult:
spinner.add(tox_env.conf.name)
return run_one(
tox_env,
options.parsed.no_test or options.parsed.package_only,
suspend_display=live is False,
)
try:
executor = ThreadPoolExecutor(max_workers=max_workers, thread_name_prefix="tox-driver")
env_list: list[str] = []
while True:
for env in env_list: # queue all available
tox_env_to_run = cast(RunToxEnv, state.envs[env])
if interrupt.is_set(): # queue the rest as failed upfront
tox_env_to_run.teardown()
future: Future[ToxEnvRunResult] = Future()
res = ToxEnvRunResult(name=env, skipped=False, code=-2, outcomes=[], duration=MISS_DURATION)
future.set_result(res)
else:
future = executor.submit(_run, tox_env_to_run)
future_to_env[future] = tox_env_to_run
if not future_to_env:
result: ToxEnvRunResult | None = None
else: # if we have queued wait for completed
future = next(as_completed(future_to_env))
tox_env_done = future_to_env.pop(future)
try:
result = future.result()
except CancelledError:
tox_env_done.teardown()
name = tox_env_done.conf.name
result = ToxEnvRunResult(
name=name,
skipped=False,
code=-3,
outcomes=[],
duration=MISS_DURATION,
)
results.append(result)
completed.add(result.name)
env_list = next(envs_to_run_generator, [])
# if nothing running and nothing more to run we're done
final_run = not env_list and not future_to_env
if final_run: # disable report on final env
spinner.print_report = False
if result is not None:
_handle_one_run_done(result, spinner, state, live)
if final_run:
break
except BaseException: # pragma: no cover
logging.exception("Internal Error") # pragma: no cover
raise # pragma: no cover
finally:
executor.shutdown(wait=True)
finally:
try:
# call teardown - configuration only environments for example could not be finished
for name in to_run_list:
state.envs[name].teardown()
finally:
done.set()
def _handle_one_run_done(result: ToxEnvRunResult, spinner: ToxSpinner, state: State, live: bool) -> None:
success = result.code == Outcome.OK
spinner.update_spinner(result, success)
tox_env = cast(RunToxEnv, state.envs[result.name])
if tox_env.journal: # add overall journal entry
tox_env.journal["result"] = {
"success": success,
"exit_code": result.code,
"duration": result.duration,
}
if live is False and state.conf.options.parallel_live is False: # teardown background run
out_err = tox_env.close_and_read_out_err() # sync writes from buffer to stdout/stderr
pkg_out_err_list = []
for package_env in tox_env.package_envs:
pkg_out_err = package_env.close_and_read_out_err()
if pkg_out_err is not None: # pragma: no branch
pkg_out_err_list.append(pkg_out_err)
if not success or tox_env.conf["parallel_show_output"]:
for pkg_out_err in pkg_out_err_list:
state._options.log_handler.write_out_err(pkg_out_err) # pragma: no cover
if out_err is not None: # pragma: no branch # first show package build
state._options.log_handler.write_out_err(out_err)
def ready_to_run_envs(state: State, to_run: list[str], completed: set[str]) -> Iterator[list[str]]:
"""Generate tox environments ready to run"""
order, todo = run_order(state, to_run)
while order:
ready_to_run: list[str] = []
new_order: list[str] = []
for env in order: # collect next batch of ready to run
if todo[env] - completed:
new_order.append(env)
else:
ready_to_run.append(env)
order = new_order
yield ready_to_run
def run_order(state: State, to_run: list[str]) -> tuple[list[str], dict[str, set[str]]]:
to_run_set = set(to_run)
todo: dict[str, set[str]] = {}
for env in to_run:
run_env = cast(RunToxEnv, state.envs[env])
depends = set(cast(EnvList, run_env.conf["depends"]).envs)
todo[env] = to_run_set & depends
order = stable_topological_sort(todo)
return order, todo
|