Implement parallel mode - without output redirect

Signed-off-by: Bernát Gábor <bgabor8@bloomberg.net>

13 files changed, 221 insertions, 230 deletions

diff --git a/src/tox/session/cmd/legacy.py b/src/tox/session/cmd/legacy.py
index 2bd7807a..b60335ea 100644
--- a/src/tox/session/cmd/legacy.py
+++ b/src/tox/session/cmd/legacy.py
@@ -3,7 +3,7 @@ from pathlib import Path
 from tox.config.cli.parser import DEFAULT_VERBOSITY, ToxParser
 from tox.plugin.impl import impl
 from tox.session.cmd.run.common import env_run_create_flags
-from tox.session.cmd.run.parallel import parallel_flags, run_parallel
+from tox.session.cmd.run.parallel import OFF_VALUE, parallel_flags, run_parallel
 from tox.session.cmd.run.sequential import run_sequential
 from tox.session.common import env_list_flag
 from tox.session.state import State
@@ -49,7 +49,7 @@ def tox_add_option(parser: ToxParser) -> None:
     our.add_argument("-c", metavar="CONFIGFILE", help="show live configuration", dest="config_file", default="")
     env_list_flag(our)
     env_run_create_flags(our)
-    parallel_flags(our)
+    parallel_flags(our, default_parallel=OFF_VALUE)
     our.add_argument(
         "--pre",
         action="store_true",
diff --git a/src/tox/session/cmd/run/common.py b/src/tox/session/cmd/run/common.py
index 59ca4526..9b332e50 100644
--- a/src/tox/session/cmd/run/common.py
+++ b/src/tox/session/cmd/run/common.py
@@ -1,7 +1,14 @@
 """Common functionality shared across multiple type of runs"""
+import time
 from argparse import Action, ArgumentParser, ArgumentTypeError, Namespace
 from pathlib import Path
-from typing import Any, Optional, Sequence, Union
+from typing import Any, Dict, Iterator, List, Optional, Sequence, Tuple, Union
+
+from colorama import Fore
+
+from tox.execute import Outcome
+from tox.journal import write_journal
+from tox.session.state import State
 
 
 class SkipMissingInterpreterAction(Action):
@@ -86,3 +93,34 @@ def env_run_create_flags(parser: ArgumentParser) -> None:
         help="for python discovery first try the python executables under these paths",
         default=[],
     )
+
+
+def run_and_report(state: State, result: Iterator[Tuple[str, Tuple[int, List[Outcome], float]]]) -> int:
+    status_codes: Dict[str, Tuple[int, float, List[float]]] = {}
+    for name, (code, outcomes, duration) in result:
+        status_codes[name] = code, duration, [o.elapsed for o in outcomes]
+    write_journal(getattr(state.options, "result_json", None), state.journal)
+    return report(state.options.start, status_codes, state.options.is_colored)
+
+
+def report(start: float, status_dict: Dict[str, Tuple[int, float, List[float]]], is_colored: bool) -> int:
+    def _print(color: int, message: str) -> None:
+        print(f"{color if is_colored else ''}{message}{Fore.RESET if is_colored else ''}")
+
+    end = time.monotonic()
+    all_ok = True
+    for name, (status, duration_one, duration_individual) in status_dict.items():
+        ok = status == Outcome.OK
+        msg = "OK " if ok else f"FAIL code {status}"
+        extra = f"+cmd[{','.join(f'{i:.2f}' for i in duration_individual)}]" if len(duration_individual) else ""
+        setup = duration_one - sum(duration_individual)
+        out = f"  {name}: {msg}({duration_one:.2f}{f'=setup[{setup:.2f}]{extra}' if extra else ''} seconds)"
+        _print(Fore.GREEN if ok else Fore.RED, out)
+        all_ok = ok and all_ok
+    duration = end - start
+    if all_ok:
+        _print(Fore.GREEN, f"  congratulations :) ({duration:.2f} seconds)")
+        return Outcome.OK
+    else:
+        _print(Fore.RED, f"  evaluation failed :( ({duration:.2f} seconds)")
+        return -1
diff --git a/src/tox/session/cmd/run/parallel.py b/src/tox/session/cmd/run/parallel.py
index 83c97e23..0f28bfae 100644
--- a/src/tox/session/cmd/run/parallel.py
+++ b/src/tox/session/cmd/run/parallel.py
@@ -2,32 +2,30 @@
 """
 Run tox environments in parallel.
 """
-import inspect
 import logging
 import os
-import sys
 from argparse import ArgumentParser, ArgumentTypeError
-from collections import OrderedDict, deque
-from pathlib import Path
-from threading import Event, Semaphore, Thread
-from typing import cast
+from concurrent.futures import Future, ThreadPoolExecutor, as_completed
+from typing import Callable, Dict, Iterator, List, Set, Tuple, cast
 
-import tox
 from tox.config.cli.parser import ToxParser
+from tox.config.types import EnvList
+from tox.execute import Outcome
 from tox.plugin.impl import impl
+from tox.session.cmd.run.single import run_one
 from tox.session.common import env_list_flag
 from tox.session.state import State
 from tox.util.cpu import auto_detect_cpus
+from tox.util.graph import stable_topological_sort
 from tox.util.spinner import Spinner
 
-from .common import env_run_create_flags
+from .common import env_run_create_flags, run_and_report
 
 logger = logging.getLogger(__name__)
 
 ENV_VAR_KEY = "TOX_PARALLEL_ENV"
 OFF_VALUE = 0
 DEFAULT_PARALLEL = OFF_VALUE
-MAIN_FILE = Path(cast(str, inspect.getsourcefile(tox))).parent / "__main__.py"
 
 
 @impl
@@ -35,7 +33,7 @@ def tox_add_option(parser: ToxParser) -> None:
     our = parser.add_command("run-parallel", ["p"], "run environments in parallel", run_parallel)
     env_list_flag(our)
     env_run_create_flags(our)
-    parallel_flags(our)
+    parallel_flags(our, default_parallel=auto_detect_cpus())
 
 
 def parse_num_processes(str_value):
@@ -43,14 +41,13 @@ def parse_num_processes(str_value):
         return None
     if str_value == "auto":
         return auto_detect_cpus()
-    else:
-        value = int(str_value)
-        if value < 0:
-            raise ArgumentTypeError("value must be positive")
-        return value
+    value = int(str_value)
+    if value < 0:
+        raise ArgumentTypeError("value must be positive")
+    return value
 
 
-def parallel_flags(our: ArgumentParser) -> None:
+def parallel_flags(our: ArgumentParser, default_parallel: int) -> None:
     our.add_argument(
         "-p",
         "--parallel",
@@ -59,7 +56,7 @@ def parallel_flags(our: ArgumentParser) -> None:
         " auto - cpu count, some positive number, zero is turn off",
         action="store",
         type=parse_num_processes,
-        default=DEFAULT_PARALLEL,
+        default=default_parallel,
         metavar="VAL",
     )
     our.add_argument(
@@ -69,125 +66,70 @@ def parallel_flags(our: ArgumentParser) -> None:
         dest="parallel_live",
         help="connect to stdout while running environments",
     )
+    our.add_argument(
+        "--parallel-no-spinner",
+        action="store_true",
+        dest="parallel_no_spinner",
+        help="do not show the spinner",
+    )
 
 
 def run_parallel(state: State) -> int:
     """here we'll just start parallel sub-processes"""
-    live_out = state.options.parallel_live
-    disable_spinner = bool(os.environ.get("TOX_PARALLEL_NO_SPINNER") == "1")
-    args = [sys.executable, MAIN_FILE] + state.args
-    try:
-        position = args.index("--")
-    except ValueError:
-        position = len(args)
-
-    max_parallel = state.options.parallel
-    if max_parallel is None:
-        max_parallel = len(state.tox_envs)
-    semaphore = Semaphore(max_parallel)
-    finished = Event()
-
-    show_progress = not disable_spinner and not live_out and state.options.verbosity > 2
-
-    with Spinner(enabled=show_progress) as spinner:
-
-        def run_in_thread(tox_env, os_env, process_dict):
-            output = None
-            env_name = tox_env.envconfig.envname
-            status = "skipped tests" if state.options.no_test else None
-            try:
-                os_env[str(ENV_VAR_KEY)] = str(env_name)
-                args_sub = list(args)
-                if hasattr(tox_env, "package"):
-                    args_sub.insert(position, str(tox_env.perform_packaging))
-                    args_sub.insert(position, "--installpkg")
-                if tox_env.get_result_json_path():
-                    result_json_index = args_sub.index("--result-json")
-                    args_sub[result_json_index + 1] = f"{tox_env.get_result_json_path()}"
-                with tox_env.new_action(f"parallel {tox_env.name}") as action:
-
-                    def collect_process(process):
-                        process_dict[tox_env] = (action, process)
-
-                    print_out = not live_out and tox_env.envconfig.parallel_show_output
-                    output = action.popen(
-                        args=args_sub,
-                        env=os_env,
-                        redirect=not live_out,
-                        capture_err=print_out,
-                        callback=collect_process,
-                        returnout=print_out,
-                    )
-
-            except Exception as err:
-                status = f"parallel child exit {err!r}"
-            finally:
-                semaphore.release()
-                finished.set()
-                tox_env.status = status
-                done.add(env_name)
-                outcome = spinner.succeed
-                if state.options.notest:
-                    outcome = spinner.skip
-                elif status is not None:
-                    outcome = spinner.fail
-                outcome(env_name)
-                if print_out and output is not None:
-                    logger.warning(output)
-
-        threads = deque()
-        processes = {}
-        todo_keys = set(state.env_list)
-        todo = OrderedDict((n, todo_keys & set(v.conf["depends"])) for n, v in state.tox_envs.items())
-        done = set()
+    return run_and_report(state, _execute_parallel(state))
+
+
+def _execute_parallel(state: State) -> Iterator[Tuple[str, Tuple[int, List[Outcome], float]]]:
+    options = state.options
+    live_out = options.parallel_live
+    show_progress = not options.parallel_no_spinner and not live_out and options.verbosity > 2
+    with Spinner(enabled=show_progress, colored=state.options.is_colored) as spinner:
+        spinner_name_done: Callable[[str], None] = spinner.skip if options.no_test else spinner.succeed
+        to_run_list = list(state.env_list())
+        max_parallel = options.parallel
+        executor = ThreadPoolExecutor(len(to_run_list) if max_parallel is None else max_parallel, "tox-parallel")
         try:
-            while todo:
-                for name, depends in list(todo.items()):
-                    if depends - done:
-                        # skip if has unfinished dependencies
-                        continue
-                    del todo[name]
-                    venv = state.tox_envs[name]
-                    semaphore.acquire()
-                    spinner.add(name)
-                    thread = Thread(target=run_in_thread, args=(venv, os.environ.copy(), processes))
-                    thread.daemon = True
-                    thread.start()
-                    threads.append(thread)
-                if todo:
-                    # wait until someone finishes and retry queuing jobs
-                    finished.wait()
-                    finished.clear()
-            while threads:
-                threads = [thread for thread in threads if not thread.join(0.1) and thread.is_alive()]
+            future_to_name: Dict[Future, str] = {}
+            completed: Set[str] = set()
+            envs_to_run_generator = ready_to_run_envs(state, to_run_list, completed)
+            while True:
+                env_list = next(envs_to_run_generator, [])
+                if not env_list and not future_to_name:
+                    break
+                for env in env_list:  # queue all available
+                    spinner.add(env)
+                    tox_env = state.tox_env(env)
+                    if live_out is False:
+                        tox_env.hide_display()
+                    future = executor.submit(run_one, tox_env, options.recreate, options.no_test)
+                    future_to_name[future] = env
+
+                future = next(as_completed(future_to_name))
+                name = future_to_name.pop(future)
+                code, outcomes, duration = future.result()
+                completed.add(name)
+                if live_out is False:
+                    state.tox_env(env).resume_display()
+                (spinner_name_done if code == Outcome.OK else spinner.fail)(name)
+
+                yield name, (code, outcomes, duration)
         except KeyboardInterrupt:
             logger.error(f"[{os.getpid()}] KeyboardInterrupt parallel - stopping children")
-            while True:
-                # do not allow interrupting until children interrupt
-                try:
-                    # putting it inside a thread to guarantee it's not interrupted
-                    stopper = Thread(target=_stop_child_processes, args=(processes, threads))
-                    stopper.start()
-                    stopper.join()
-                except KeyboardInterrupt:
-                    continue
-                raise KeyboardInterrupt
-    return 0
-
-
-def _stop_child_processes(processes, main_threads):
-    """A three level stop mechanism for children - INT (250ms) -> TERM (100ms) -> KILL"""
-
-    # first stop children
-    def shutdown(tox_env, action, process):  # noqa
-        action.handle_interrupt(process)
-
-    threads = [Thread(target=shutdown, args=(n, a, p)) for n, (a, p) in processes.items()]
-    for thread in threads:
-        thread.start()
-    for thread in threads:
-        thread.join()
-
-    # then its threads
-    for thread in main_threads:
-        thread.join()
+        executor.shutdown(wait=False)
+
+
+def ready_to_run_envs(state: State, to_run: List[str], completed: Set[str]):
+    """Generate tox environments ready to run"""
+    to_run_set = set(to_run)
+    todo: Dict[str, Set[str]] = {
+        env: (to_run_set & set(cast(EnvList, state.tox_env(env).conf["depends"]).envs)) for env in to_run
+    }
+    order, at = stable_topological_sort(todo), 0
+    while at != len(order):
+        ready_to_run = []
+        for env in order[at:]:  # collect next batch of ready to run
+            if todo[env] - completed:
+                break  # if not all dependencies completed, stop, topological order guarantees we're done
+            ready_to_run.append(env)
+            at += 1
+        yield ready_to_run
diff --git a/src/tox/session/cmd/run/sequential.py b/src/tox/session/cmd/run/sequential.py
index 7fb5c72d..133d2a3e 100644
--- a/src/tox/session/cmd/run/sequential.py
+++ b/src/tox/session/cmd/run/sequential.py
@@ -1,19 +1,15 @@
 """
 Run tox environments in sequential order.
 """
-import time
-from typing import Dict, List, Tuple
-
-from colorama import Fore
+from typing import Iterator, List, Tuple
 
 from tox.config.cli.parser import ToxParser
-from tox.execute.api import Outcome
-from tox.journal import write_journal
+from tox.execute import Outcome
 from tox.plugin.impl import impl
 from tox.session.common import env_list_flag
 from tox.session.state import State
 
-from .common import env_run_create_flags
+from .common import env_run_create_flags, run_and_report
 from .single import run_one
 
 
@@ -25,35 +21,9 @@ def tox_add_option(parser: ToxParser) -> None:
 
 
 def run_sequential(state: State) -> int:
-    status_codes: Dict[str, Tuple[int, float, List[float]]] = {}
+    return run_and_report(state, _execute_sequential(state))
+
+
+def _execute_sequential(state: State) -> Iterator[Tuple[str, Tuple[int, List[Outcome], float]]]:
     for name in state.env_list(everything=False):
-        tox_env = state.tox_env(name)
-        start_one = time.monotonic()
-        code, outcomes = run_one(tox_env, state.options.recreate, state.options.no_test)
-        duration = time.monotonic() - start_one
-        status_codes[name] = code, duration, [o.elapsed for o in outcomes]
-    write_journal(getattr(state.options, "result_json", None), state.journal)
-    return report(state.options.start, status_codes, state.options.is_colored)
-
-
-def report(start: float, status_dict: Dict[str, Tuple[int, float, List[float]]], is_colored: bool) -> int:
-    def _print(color: int, message: str) -> None:
-        print(f"{color if is_colored else ''}{message}{Fore.RESET if is_colored else ''}")
-
-    end = time.monotonic()
-    all_ok = True
-    for name, (status, duration_one, duration_individual) in status_dict.items():
-        ok = status == Outcome.OK
-        msg = "OK " if ok else f"FAIL code {status}"
-        extra = f"+cmd[{','.join(f'{i:.2f}' for i in duration_individual)}]" if len(duration_individual) else ""
-        setup = duration_one - sum(duration_individual)
-        out = f"  {name}: {msg}({duration_one:.2f}{f'=setup[{setup:.2f}]{extra}' if extra else ''} seconds)"
-        _print(Fore.GREEN if ok else Fore.RED, out)
-        all_ok = ok and all_ok
-    duration = end - start
-    if all_ok:
-        _print(Fore.GREEN, f"  congratulations :) ({duration:.2f} seconds)")
-        return Outcome.OK
-    else:
-        _print(Fore.RED, f"  evaluation failed :( ({duration:.2f} seconds)")
-        return -1
+        yield name, run_one(state.tox_env(name), state.options.recreate, state.options.no_test)
diff --git a/src/tox/session/cmd/run/single.py b/src/tox/session/cmd/run/single.py
index 5ee6ac44..8b0fe034 100644
--- a/src/tox/session/cmd/run/single.py
+++ b/src/tox/session/cmd/run/single.py
@@ -1,6 +1,7 @@
 """
 Defines how to run a single tox environment.
 """
+from datetime import datetime
 from typing import List, Tuple, cast
 
 from tox.config.types import Command
@@ -9,10 +10,14 @@ from tox.tox_env.api import ToxEnv
 from tox.tox_env.runner import RunToxEnv
 
 
-def run_one(tox_env: RunToxEnv, recreate: bool, no_test: bool) -> Tuple[int, List[Outcome]]:
-    tox_env.ensure_setup(recreate=recreate)
-    code, outcomes = run_commands(tox_env, no_test)
-    return code, outcomes
+def run_one(tox_env: RunToxEnv, recreate: bool, no_test: bool) -> Tuple[int, List[Outcome], float]:
+    start_one = datetime.now()
+    try:
+        tox_env.ensure_setup(recreate=recreate)
+        code, outcomes = run_commands(tox_env, no_test)
+    finally:
+        duration = (datetime.now() - start_one).total_seconds()
+    return code, outcomes, duration
 
 
 def run_commands(tox_env: ToxEnv, no_test: bool) -> Tuple[int, List[Outcome]]:
diff --git a/src/tox/tox_env/api.py b/src/tox/tox_env/api.py
index 07167c91..e8cc424a 100644
--- a/src/tox/tox_env/api.py
+++ b/src/tox/tox_env/api.py
@@ -218,5 +218,13 @@ class ToxEnv(ABC):
     def id() -> str:
         raise NotImplementedError
 
+    def hide_display(self) -> None:
+        """No longer show"""
+        assert self.logger.name
+
+    def resume_display(self) -> None:
+        """No longer show"""
+        assert self.logger.name
+
 
 _CWD = Path.cwd()
diff --git a/src/tox/util/graph.py b/src/tox/util/graph.py
index ea621f6b..47c6b080 100644
--- a/src/tox/util/graph.py
+++ b/src/tox/util/graph.py
@@ -1,16 +1,16 @@
 """Helper methods related to graph theory."""
 from collections import OrderedDict, defaultdict
-from typing import Dict, List, Optional, Tuple
+from typing import Dict, List, Optional, Set
 
 
-def stable_topological_sort(graph: Dict[str, Tuple[str, ...]]) -> List[str]:
+def stable_topological_sort(graph: Dict[str, Set[str]]) -> List[str]:
     to_order = set(graph.keys())  # keep a log of what  we need to order
 
     # normalize graph - fill missing nodes (assume no dependency)
     for values in list(graph.values()):
         for value in values:
             if value not in graph:
-                graph[value] = ()
+                graph[value] = set()
 
     inverse_graph = defaultdict(set)
     for key, depends in graph.items():
@@ -47,7 +47,7 @@ def stable_topological_sort(graph: Dict[str, Tuple[str, ...]]) -> List[str]:
     return result
 
 
-def identify_cycle(graph: Dict[str, Tuple[str, ...]]) -> None:
+def identify_cycle(graph: Dict[str, Set[str]]) -> None:
     path: Dict[str, None] = OrderedDict()
     visited = set()
 
diff --git a/src/tox/util/spinner.py b/src/tox/util/spinner.py
index f3d0e570..3662c17a 100644
--- a/src/tox/util/spinner.py
+++ b/src/tox/util/spinner.py
@@ -2,8 +2,8 @@
 import os
 import sys
 import threading
+import time
 from collections import OrderedDict
-from datetime import datetime, timedelta
 from types import TracebackType
 from typing import IO, Any, Dict, Optional, Sequence, Type
 
@@ -35,14 +35,15 @@ class Spinner:
     UNICODE_FRAMES = ["⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"]
     ASCII_FRAMES = ["|", "-", "+", "x", "*"]
 
-    def __init__(self, enabled: bool = True, refresh_rate: float = 0.1) -> None:
+    def __init__(self, enabled: bool = True, refresh_rate: float = 0.1, colored: bool = True) -> None:
+        self.is_colored = colored
         self.refresh_rate = refresh_rate
         self.enabled = enabled
         self.frames = (
             self.UNICODE_FRAMES if _file_support_encoding(self.UNICODE_FRAMES, sys.stdout) else self.ASCII_FRAMES
         )
         self.stream = sys.stdout
-        self._envs: Dict[str, datetime] = OrderedDict()
+        self._envs: Dict[str, float] = OrderedDict()
         self._frame_index = 0
 
     def clear(self) -> None:
@@ -99,26 +100,26 @@ class Spinner:
                 self.enable_cursor()
 
     def add(self, name: str) -> None:
-        self._envs[name] = datetime.now()
+        self._envs[name] = time.monotonic()
 
     def succeed(self, key: str) -> None:
-        self.finalize(key, "✔ OK", Fore.GREEN)
+        self.finalize(key, "OK ✔", Fore.GREEN)
 
     def fail(self, key: str) -> None:
-        self.finalize(key, "✖ FAIL", Fore.RED)
+        self.finalize(key, "FAIL ✖", Fore.RED)
 
     def skip(self, key: str) -> None:
-        self.finalize(key, "⚠ SKIP", Fore.WHITE)
+        self.finalize(key, "SKIP ⚠", Fore.WHITE)
 
     def finalize(self, key: str, status: str, color: int) -> None:
         start_at = self._envs[key]
         del self._envs[key]
         if self.enabled:
             self.clear()
-        msg = f"{color}{status} {key} in {td_human_readable(datetime.now() - start_at)}{Fore.RESET}{os.linesep}"
-        self.stream.write(msg)
-        if not self._envs:
-            self.__exit__(None, None, None)
+        base = f"{key}: {status} in {td_human_readable(time.monotonic() - start_at)}"
+        if self.is_colored:
+            base = f"{color}{base}{Fore.RESET}{os.linesep}"
+        self.stream.write(base)
 
     def disable_cursor(self) -> None:
         if self.stream.isatty():
@@ -151,15 +152,15 @@ _PERIODS = [
 ]
 
 
-def td_human_readable(delta: timedelta) -> str:
-    seconds: float = delta.total_seconds()
+def td_human_readable(seconds: float) -> str:
     texts = []
+    total_seconds = seconds
     for period_name, period_seconds in _PERIODS:
         if seconds > period_seconds or period_seconds == 1:
             period_value = int(seconds) // period_seconds
             seconds %= period_seconds
             if period_name == "second":
-                ms = period_value + delta.total_seconds() - int(delta.total_seconds())
+                ms = period_value + total_seconds - int(total_seconds)
                 period_str = f"{ms:.2f}".rstrip("0").rstrip(".")
             else:
                 period_str = str(period_value)
diff --git a/tests/config/cli/test_cli_env_var.py b/tests/config/cli/test_cli_env_var.py
index a6d989de..13054353 100644
--- a/tests/config/cli/test_cli_env_var.py
+++ b/tests/config/cli/test_cli_env_var.py
@@ -51,6 +51,7 @@ def test_verbose_no_test(monkeypatch: MonkeyPatch) -> None:
         "discover": [],
         "parallel": 0,
         "parallel_live": False,
+        "parallel_no_spinner": False,
         "pre": False,
         "index_url": [],
     }
@@ -93,6 +94,7 @@ def test_env_var_exhaustive_parallel_values(
         "package_only": False,
         "parallel": 3,
         "parallel_live": False,
+        "parallel_no_spinner": False,
         "pre": False,
         "quiet": 1,
         "recreate": True,
diff --git a/tests/config/cli/test_cli_ini.py b/tests/config/cli/test_cli_ini.py
index 5d3afde4..7ee34140 100644
--- a/tests/config/cli/test_cli_ini.py
+++ b/tests/config/cli/test_cli_ini.py
@@ -111,6 +111,7 @@ def test_ini_exhaustive_parallel_values(exhaustive_ini: Path, core_handlers: Dic
         "no_recreate_pkg": False,
         "parallel": 3,
         "parallel_live": True,
+        "parallel_no_spinner": False,
         "quiet": 1,
         "recreate": True,
         "result_json": None,
diff --git a/tests/session/cmd/test_parallel.py b/tests/session/cmd/test_parallel.py
new file mode 100644
index 00000000..04eedf7c
--- /dev/null
+++ b/tests/session/cmd/test_parallel.py
@@ -0,0 +1,24 @@
+from tox.pytest import ToxProjectCreator
+
+
+def test_parallel_run(tox_project: ToxProjectCreator) -> None:
+    ini = """
+    [tox]
+    no_package=true
+    env_list= a, b, c
+    [testenv]
+    commands=python -c 'print("{env_name}")'
+    depends = !c: c
+    """
+    project = tox_project({"tox.ini": ini})
+    outcome = project.run("p")
+    outcome.assert_success()
+
+    out = outcome.out
+    for env in "a", "b", "c":
+        env_done = f"{env}: OK ✔"
+        env_report = f"  {env}: OK ("
+
+        assert env_done in out, out
+        assert env_report in out, out
+        assert out.index(env_done) < out.index(env_report), out
diff --git a/tests/util/test_graph.py b/tests/util/test_graph.py
index 3695f8b3..b369ad17 100644
--- a/tests/util/test_graph.py
+++ b/tests/util/test_graph.py
@@ -1,5 +1,5 @@
 from collections import OrderedDict
-from typing import Dict, Tuple
+from typing import Dict, Set
 
 import pytest
 
@@ -7,65 +7,65 @@ from tox.util.graph import stable_topological_sort
 
 
 def test_topological_order_empty() -> None:
-    graph: Dict[str, Tuple[str, ...]] = OrderedDict()
+    graph: Dict[str, Set[str]] = OrderedDict()
     result = stable_topological_sort(graph)
     assert result == []
 
 
 def test_topological_order_specified_only() -> None:
-    graph: Dict[str, Tuple[str, ...]] = OrderedDict()
-    graph["A"] = "B", "C"
+    graph: Dict[str, Set[str]] = OrderedDict()
+    graph["A"] = {"B", "C"}
     result = stable_topological_sort(graph)
     assert result == ["A"]
 
 
 def test_topological_order() -> None:
-    graph: Dict[str, Tuple[str, ...]] = OrderedDict()
-    graph["A"] = "B", "C"
-    graph["B"] = ()
-    graph["C"] = ()
+    graph: Dict[str, Set[str]] = OrderedDict()
+    graph["A"] = {"B", "C"}
+    graph["B"] = set()
+    graph["C"] = set()
     result = stable_topological_sort(graph)
     assert result == ["B", "C", "A"]
 
 
 def test_topological_order_cycle() -> None:
-    graph: Dict[str, Tuple[str, ...]] = OrderedDict()
-    graph["A"] = "B", "C"
-    graph["B"] = ("A",)
+    graph: Dict[str, Set[str]] = OrderedDict()
+    graph["A"] = {"B", "C"}
+    graph["B"] = {"A"}
     with pytest.raises(ValueError, match="A | B"):
         stable_topological_sort(graph)
 
 
 def test_topological_complex() -> None:
-    graph: Dict[str, Tuple[str, ...]] = OrderedDict()
-    graph["A"] = "B", "C"
-    graph["B"] = "C", "D"
-    graph["C"] = ("D",)
-    graph["D"] = ()
+    graph: Dict[str, Set[str]] = OrderedDict()
+    graph["A"] = {"B", "C"}
+    graph["B"] = {"C", "D"}
+    graph["C"] = {"D"}
+    graph["D"] = set()
     result = stable_topological_sort(graph)
     assert result == ["D", "C", "B", "A"]
 
 
 def test_two_sub_graph() -> None:
-    graph: Dict[str, Tuple[str, ...]] = OrderedDict()
-    graph["F"] = ()
-    graph["E"] = ()
-    graph["D"] = "E", "F"
-    graph["A"] = "B", "C"
-    graph["B"] = ()
-    graph["C"] = ()
+    graph: Dict[str, Set[str]] = OrderedDict()
+    graph["F"] = set()
+    graph["E"] = set()
+    graph["D"] = {"E", "F"}
+    graph["A"] = {"B", "C"}
+    graph["B"] = set()
+    graph["C"] = set()
 
     result = stable_topological_sort(graph)
     assert result == ["F", "E", "D", "B", "C", "A"]
 
 
 def test_two_sub_graph_circle() -> None:
-    graph: Dict[str, Tuple[str, ...]] = OrderedDict()
-    graph["F"] = ()
-    graph["E"] = ()
-    graph["D"] = "E", "F"
-    graph["A"] = "B", "C"
-    graph["B"] = ("A",)
-    graph["C"] = ()
+    graph: Dict[str, Set[str]] = OrderedDict()
+    graph["F"] = set()
+    graph["E"] = set()
+    graph["D"] = {"E", "F"}
+    graph["A"] = {"B", "C"}
+    graph["B"] = {"A"}
+    graph["C"] = set()
     with pytest.raises(ValueError, match="A | B"):
         stable_topological_sort(graph)
diff --git a/tests/util/test_spinner.py b/tests/util/test_spinner.py
index 11cebb3e..8df3a2c5 100644
--- a/tests/util/test_spinner.py
+++ b/tests/util/test_spinner.py
@@ -38,7 +38,7 @@ def test_spinner_disabled(capfd: CaptureFixture, monkeypatch: MonkeyPatch) -> No
         spin.finalize("x", "done", Fore.GREEN)
         spin.clear()
     out, err = capfd.readouterr()
-    assert out == f"{Fore.GREEN}done x in 0 seconds{Fore.RESET}{os.linesep}"
+    assert out == f"{Fore.GREEN}x: done in 0 seconds{Fore.RESET}{os.linesep}", out
     assert err == ""
 
 
@@ -85,9 +85,9 @@ def test_spinner_report(capfd: CaptureFixture, monkeypatch: MonkeyPatch) -> None
     lines = out.split(os.linesep)
     del lines[0]
     expected = [
-        f"\r{spin.CLEAR_LINE}{Fore.GREEN}✔ OK ok in 0 seconds{Fore.RESET}",
-        f"\r{spin.CLEAR_LINE}{Fore.RED}✖ FAIL fail in 0 seconds{Fore.RESET}",
-        f"\r{spin.CLEAR_LINE}{Fore.WHITE}⚠ SKIP skip in 0 seconds{Fore.RESET}",
+        f"\r{spin.CLEAR_LINE}{Fore.GREEN}ok: OK ✔ in 0 seconds{Fore.RESET}",
+        f"\r{spin.CLEAR_LINE}{Fore.RED}fail: FAIL ✖ in 0 seconds{Fore.RESET}",
+        f"\r{spin.CLEAR_LINE}{Fore.WHITE}skip: SKIP ⚠ in 0 seconds{Fore.RESET}",
         f"\r{spin.CLEAR_LINE}",
     ]
     assert lines == expected
@@ -141,4 +141,4 @@ def test_spinner_stdout_not_unicode(capfd: CaptureFixture, mocker: MockerFixture
 )
 def test_td_human_readable(seconds: float, expected: str) -> None:
     dt = timedelta(seconds=seconds)
-    assert spinner.td_human_readable(dt) == expected
+    assert spinner.td_human_readable(dt.total_seconds()) == expected