Release 3.5.0

Follow up to # 632.

---------

Co-authored-by: Jörg Kohlsdorf <joerg@cornershopapp.com>

.github/workflows/deploy.yml

@@ -1,33 +1,55 @@
 name: deploy
 
 on:
-  push:
-    tags:
-      - "v*"
+  workflow_dispatch:
+    inputs:
+      version:
+        description: 'Release version'
+        required: true
+        default: '1.2.3'
 
 jobs:
-  deploy:
 
+  package:
     runs-on: ubuntu-latest
+    env:
+      SETUPTOOLS_SCM_PRETEND_VERSION: ${{ github.event.inputs.version }}
 
     steps:
     - uses: actions/checkout@v3
+
+    - name: Build and Check Package
+      uses: hynek/build-and-inspect-python-package@v1.5
+
+  deploy:
+    needs: package
+    runs-on: ubuntu-latest
+    environment: deploy
+    permissions:
+      id-token: write  # For PyPI trusted publishers.
+      contents: write  # For tag.
+
+    steps:
+    - uses: actions/checkout@v3
+
+    - name: Download Package
+      uses: actions/download-artifact@v3
       with:
-        # Needed to fetch tags, which are required by setuptools-scm.
-        fetch-depth: 0
-    - name: Set up Python
-      uses: actions/setup-python@v2
-      with:
-        python-version: "3.10"
-    - name: Install build
-      run: |
-        python -m pip install --upgrade pip
-        pip install build
-    - name: Build package
-      run: |
-        python -m build
+        name: Packages
+        path: dist
+
     - name: Publish package to PyPI
-      uses: pypa/gh-action-pypi-publish@release/v1
+      uses: pypa/gh-action-pypi-publish@v1.8.5
+
+    - name: Push tag
+      run: |
+        git config user.name "pytest bot"
+        git config user.email "pytestbot@gmail.com"
+        git tag --annotate --message=v${{ github.event.inputs.version }} v${{ github.event.inputs.version }} ${{ github.sha }}
+        git push origin v${{ github.event.inputs.version }}
+
+    - name: GitHub Release
+      uses: softprops/action-gh-release@v1
       with:
-        user: __token__
-        password: ${{ secrets.pypi_token }}
+        files: dist/*
+        tag_name: v${{ github.event.inputs.version }}

.github/workflows/test.yml

@@ -3,15 +3,31 @@ name: test
 on:
   push:
     branches:
-      - "*"
+      - master
+      - "test-me-*"
 
   pull_request:
     branches:
       - "*"
 
+# Cancel running jobs for the same workflow and branch.
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref }}
+  cancel-in-progress: true
+
 jobs:
+
+  package:
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/checkout@v3
+    - name: Build and Check Package
+      uses: hynek/build-and-inspect-python-package@v1.5
+
   test:
 
+    needs: [package]
+
     runs-on: ${{ matrix.os }}
 
     strategy:
@@ -22,9 +38,9 @@ jobs:
           - "py38-pytestlatest"
           - "py39-pytestlatest"
           - "py310-pytestlatest"
-          - "py310-pytestmain"
           - "py311-pytestlatest"
           - "py311-pytestmain"
+          - "py312-pytestlatest"
           - "py310-psutil"
           - "py310-setproctitle"
 
@@ -38,12 +54,12 @@ jobs:
             python: "3.9"
           - tox_env: "py310-pytestlatest"
             python: "3.10"
-          - tox_env: "py310-pytestmain"
-            python: "3.10"
           - tox_env: "py311-pytestlatest"
             python: "3.11"
           - tox_env: "py311-pytestmain"
             python: "3.11"
+          - tox_env: "py312-pytestlatest"
+            python: "3.12"
           - tox_env: "py310-psutil"
             python: "3.10"
           - tox_env: "py310-setproctitle"
@@ -54,14 +70,24 @@ jobs:
       with:
         # Needed to fetch tags, which are required by setuptools-scm.
         fetch-depth: 0
+
+    - name: Download Package
+      uses: actions/download-artifact@v3
+      with:
+        name: Packages
+        path: dist
+
     - name: Set up Python
       uses: actions/setup-python@v2
       with:
         python-version: ${{ matrix.python }}
+
     - name: Install tox
       run: |
         python -m pip install --upgrade pip
         pip install tox
+
     - name: Test
+      shell: bash
       run: |
-        tox -e ${{ matrix.tox_env }}
+        tox run -e ${{ matrix.tox_env }} --installpkg `find dist/*.tar.gz`

.pre-commit-config.yaml

@@ -1,32 +1,32 @@
 repos:
 -   repo: https://github.com/PyCQA/autoflake
-    rev: v2.0.1
+    rev: v2.2.1
     hooks:
     -   id: autoflake
         args: ["--in-place", "--remove-unused-variables", "--remove-all-unused-imports"]
 -   repo: https://github.com/psf/black
-    rev: 23.1.0
+    rev: 23.11.0
     hooks:
     -   id: black
         args: [--safe, --quiet, --target-version, py35]
 -   repo: https://github.com/asottile/blacken-docs
-    rev: 1.13.0
+    rev: 1.16.0
     hooks:
     -   id: blacken-docs
         additional_dependencies: [black==23.1.0]
 -   repo: https://github.com/pre-commit/pre-commit-hooks
-    rev: v4.4.0
+    rev: v4.5.0
     hooks:
     -   id: trailing-whitespace
     -   id: end-of-file-fixer
     -   id: check-yaml
     -   id: debug-statements
 -   repo: https://github.com/PyCQA/flake8
-    rev: 6.0.0
+    rev: 6.1.0
     hooks:
     -   id: flake8
 -   repo: https://github.com/asottile/pyupgrade
-    rev: v3.3.1
+    rev: v3.15.0
     hooks:
     -   id: pyupgrade
         args: [--py3-plus]
@@ -39,7 +39,7 @@ repos:
         language: python
         additional_dependencies: [pygments, restructuredtext_lint]
 -   repo: https://github.com/pre-commit/mirrors-mypy
-    rev: v0.991
+    rev: v1.7.0
     hooks:
     -   id: mypy
         files: ^(src/|testing/)

.readthedocs.yaml

@@ -0,0 +1,14 @@
+version: 2
+
+build:
+  os: ubuntu-22.04
+  tools:
+    python: "3.11"
+
+sphinx:
+  configuration: docs/conf.py
+
+python:
+  install:
+    - path: .
+    - requirements: docs/requirements.txt

CHANGELOG.rst

@@ -1,3 +1,58 @@
+pytest-xdist 3.5.0 (2023-11-21)
+===============================
+
+Features
+--------
+
+- `#632 <https://github.com/pytest-dev/pytest-xdist/issues/632>`_: ``--dist=loadscope`` now sorts scopes by number of tests to assign largest scopes early -- in many cases this should improve overall test session running time, as there is less chance of a large scope being left to be processed near the end of the session, leaving other workers idle.
+
+
+pytest-xdist 3.4.0 (2023-11-11)
+===============================
+
+Features
+--------
+
+- `#963 <https://github.com/pytest-dev/pytest-xdist/issues/963>`_: Wait for workers to finish reporting when test run stops early.
+
+  This makes sure that the results of in-progress tests are displayed.
+  Previously these reports were being discarded, losing information about the
+  test run.
+
+- `#965 <https://github.com/pytest-dev/pytest-xdist/issues/965>`_: Added support for Python 3.12.
+
+
+pytest-xdist 3.3.1 (2023-05-19)
+===============================
+
+Bug Fixes
+---------
+
+- `#907 <https://github.com/pytest-dev/pytest-xdist/issues/907>`_: Avoid remote calls during startup as ``execnet`` by default does not ensure remote affinity with the
+  main thread and might accidentally schedule the pytest worker into a non-main thread, which breaks numerous frameworks,
+  for example ``asyncio``, ``anyio``, ``PyQt/PySide``, etc.
+
+  A more safe correction will require thread affinity in ``execnet`` (`pytest-dev/execnet#96 <https://github.com/pytest-dev/execnet/issues/96>`__).
+
+
+pytest-xdist 3.3.0 (2023-05-12)
+===============================
+
+Features
+--------
+
+- `#555 <https://github.com/pytest-dev/pytest-xdist/issues/555>`_: Improved progress output when collecting nodes to be less verbose.
+
+
+pytest-xdist 3.2.1 (2023-03-12)
+===============================
+
+Bug Fixes
+---------
+
+- `#884 <https://github.com/pytest-dev/pytest-xdist/issues/884>`_: Fixed hang in ``worksteal`` scheduler.
+
+
 pytest-xdist 3.2.0 (2023-02-07)
 ===============================
 

RELEASING.rst

@@ -32,14 +32,8 @@ To publish a new release ``X.Y.Z``, the steps are as follows:
 
     $ tox -e release -- X.Y.Z
 
-#. Commit and push the branch for review.
+#. Commit and push the branch to ``upstream`` and open a PR.
 
-#. Once PR is **green** and **approved**, create and push a tag::
+#. Once the PR is **green** and **approved**, start the ``deploy`` workflow manually from the branch ``release-VERSION``, passing ``VERSION`` as parameter.
 
-    $ export VERSION=X.Y.Z
-    $ git tag v$VERSION release-$VERSION
-    $ git push git@github.com:pytest-dev/pytest-xdist.git v$VERSION
-
-That will build the package and publish it on ``PyPI`` automatically.
-
-#. Merge the release PR to `master`.
+#. Merge the release PR to ``master``.

docs/distribution.rst

@@ -50,7 +50,8 @@ The test distribution algorithm is configured with the ``--dist`` command-line o
 .. _distribution modes:
 
 * ``--dist load`` **(default)**: Sends pending tests to any worker that is
-  available, without any guaranteed order.
+  available, without any guaranteed order. Scheduling can be fine-tuned with
+  the `--maxschedchunk` option, see output of `pytest --help`.
 
 * ``--dist loadscope``: Tests are grouped by **module** for *test functions*
   and by **class** for *test methods*. Groups are distributed to available

docs/how-it-works.rst

@@ -29,8 +29,8 @@ The execution flow is:
    corresponds to the position of that test in the original collection
    list. This works because all nodes have the same collection list, and
    saves bandwidth because the **controller** can now tell one of the
-   workers to just *execute test index 3* index of passing the full test
-   id.
+   workers to just *execute test index 3* instead of passing the full
+   test id.
 
 4. If **dist-mode** is **each**: the **controller** just sends the full
    list of test indexes to each node at this moment.

docs/how-to.rst

@@ -221,7 +221,6 @@ Example:
     def pytest_configure(config):
         worker_id = os.environ.get("PYTEST_XDIST_WORKER")
         if worker_id is not None:
-            log_file = config.getini("worker_log_file")
             logging.basicConfig(
                 format=config.getini("log_file_format"),
                 filename=f"tests_{worker_id}.log",

docs/remote.rst

@@ -9,13 +9,16 @@ Sending tests to remote SSH accounts
 
 .. warning::
 
-    This feature is deprecated because the support for ``rsync`` is faulty
+    The ``rsync`` feature is deprecated because its implementation is faulty
     in terms of reproducing the development environment in the remote
     worker, and there is no clear solution moving forward.
 
-    This feature is scheduled to be removed in release 4.0, to let the team
+    For that reason, ``rsync`` is scheduled to be removed in release 4.0, to let the team
     focus on a smaller set of features.
 
+    Note that SSH and socket server are not planned for removal, as they are part
+    of the ``execnet`` feature set.
+
 Suppose you have a package ``mypkg`` which contains some
 tests that you can successfully run locally. And you
 have a ssh-reachable machine ``myhost``.  Then

docs/requirements.txt

@@ -0,0 +1,2 @@
+sphinx
+sphinx-rtd-theme

setup.cfg

@@ -2,6 +2,7 @@
 name = pytest-xdist
 description = pytest xdist plugin for distributed testing, most importantly across multiple CPUs
 long_description = file: README.rst
+long_description_content_type = text/x-rst
 license = MIT
 author = holger krekel and contributors
 author_email = pytest-dev@python.org,holger@merlinux.eu
@@ -29,6 +30,7 @@ classifiers =
     Programming Language :: Python :: 3.9
     Programming Language :: Python :: 3.10
     Programming Language :: Python :: 3.11
+    Programming Language :: Python :: 3.12
 license_file = LICENSE
 project_urls =
     Documentation=https://pytest-xdist.readthedocs.io/en/latest

src/xdist/dsession.py

@@ -1,3 +1,8 @@
+from __future__ import annotations
+import sys
+from enum import Enum, auto
+from typing import Sequence
+
 import pytest
 
 from xdist.remote import Producer
@@ -113,11 +118,14 @@ class DSession:
         assert self.sched is not None
 
         self.shouldstop = False
+        pending_exception = None
         while not self.session_finished:
             self.loop_once()
             if self.shouldstop:
                 self.triggershutdown()
-                raise Interrupted(str(self.shouldstop))
+                pending_exception = Interrupted(str(self.shouldstop))
+        if pending_exception:
+            raise pending_exception
         return True
 
     def loop_once(self):
@@ -251,14 +259,16 @@ class DSession:
         self._session.testscollected = len(ids)
         self.sched.add_node_collection(node, ids)
         if self.terminal:
-            self.trdist.setstatus(node.gateway.spec, "[%d]" % (len(ids)))
+            self.trdist.setstatus(
+                node.gateway.spec, WorkerStatus.CollectionDone, tests_collected=len(ids)
+            )
         if self.sched.collection_is_completed:
             if self.terminal and not self.sched.has_pending:
                 self.trdist.ensure_show_status()
                 self.terminal.write_line("")
                 if self.config.option.verbose > 0:
                     self.terminal.write_line(
-                        "scheduling tests via %s" % (self.sched.__class__.__name__)
+                        f"scheduling tests via {self.sched.__class__.__name__}"
                     )
             self.sched.schedule()
 
@@ -344,14 +354,19 @@ class DSession:
     def _handlefailures(self, rep):
         if rep.failed:
             self.countfailures += 1
-            if self.maxfail and self.countfailures >= self.maxfail:
-                self.shouldstop = "stopping after %d failures" % (self.countfailures)
+            if (
+                self.maxfail
+                and self.countfailures >= self.maxfail
+                and not self.shouldstop
+            ):
+                self.shouldstop = f"stopping after {self.countfailures} failures"
 
     def triggershutdown(self):
-        self.log("triggering shutdown")
-        self.shuttingdown = True
-        for node in self.sched.nodes:
-            node.shutdown()
+        if not self.shuttingdown:
+            self.log("triggering shutdown")
+            self.shuttingdown = True
+            for node in self.sched.nodes:
+                node.shutdown()
 
     def handle_crashitem(self, nodeid, worker):
         # XXX get more reporting info by recording pytest_runtest_logstart?
@@ -372,32 +387,51 @@ class DSession:
         self.config.hook.pytest_runtest_logreport(report=rep)
 
 
+class WorkerStatus(Enum):
+    """Status of each worker during creation/collection."""
+
+    # Worker spec has just been created.
+    Created = auto()
+
+    # Worker has been initialized.
+    Initialized = auto()
+
+    # Worker is now ready for collection.
+    ReadyForCollection = auto()
+
+    # Worker has finished collection.
+    CollectionDone = auto()
+
+
 class TerminalDistReporter:
-    def __init__(self, config):
+    def __init__(self, config) -> None:
         self.config = config
         self.tr = config.pluginmanager.getplugin("terminalreporter")
-        self._status = {}
+        self._status: dict[str, tuple[WorkerStatus, int]] = {}
         self._lastlen = 0
         self._isatty = getattr(self.tr, "isatty", self.tr.hasmarkup)
 
-    def write_line(self, msg):
+    def write_line(self, msg: str) -> None:
         self.tr.write_line(msg)
 
-    def ensure_show_status(self):
+    def ensure_show_status(self) -> None:
         if not self._isatty:
             self.write_line(self.getstatus())
 
-    def setstatus(self, spec, status, show=True):
-        self._status[spec.id] = status
+    def setstatus(
+        self, spec, status: WorkerStatus, *, tests_collected: int, show: bool = True
+    ) -> None:
+        self._status[spec.id] = (status, tests_collected)
         if show and self._isatty:
             self.rewrite(self.getstatus())
 
-    def getstatus(self):
+    def getstatus(self) -> str:
         if self.config.option.verbose >= 0:
-            parts = [f"{spec.id} {self._status[spec.id]}" for spec in self._specs]
-            return " / ".join(parts)
-        else:
-            return "bringing up nodes..."
+            line = get_workers_status_line(list(self._status.values()))
+            if line:
+                return line
+
+        return "bringing up nodes..."
 
     def rewrite(self, line, newline=False):
         pline = line + " " * max(self._lastlen - len(line), 0)
@@ -409,37 +443,40 @@ class TerminalDistReporter:
         self.tr.rewrite(pline, bold=True)
 
     @pytest.hookimpl
-    def pytest_xdist_setupnodes(self, specs):
+    def pytest_xdist_setupnodes(self, specs) -> None:
         self._specs = specs
         for spec in specs:
-            self.setstatus(spec, "I", show=False)
-        self.setstatus(spec, "I", show=True)
+            self.setstatus(spec, WorkerStatus.Created, tests_collected=0, show=False)
+        self.setstatus(spec, WorkerStatus.Created, tests_collected=0, show=True)
         self.ensure_show_status()
 
     @pytest.hookimpl
-    def pytest_xdist_newgateway(self, gateway):
+    def pytest_xdist_newgateway(self, gateway) -> None:
         if self.config.option.verbose > 0:
             rinfo = gateway._rinfo()
-            version = "%s.%s.%s" % rinfo.version_info[:3]
-            self.rewrite(
-                "[%s] %s Python %s cwd: %s"
-                % (gateway.id, rinfo.platform, version, rinfo.cwd),
-                newline=True,
-            )
-        self.setstatus(gateway.spec, "C")
+            different_interpreter = rinfo.executable != sys.executable
+            if different_interpreter:
+                version = "%s.%s.%s" % rinfo.version_info[:3]
+                self.rewrite(
+                    f"[{gateway.id}] {rinfo.platform} Python {version} cwd: {rinfo.cwd}",
+                    newline=True,
+                )
+        self.setstatus(gateway.spec, WorkerStatus.Initialized, tests_collected=0)
 
     @pytest.hookimpl
-    def pytest_testnodeready(self, node):
+    def pytest_testnodeready(self, node) -> None:
         if self.config.option.verbose > 0:
             d = node.workerinfo
-            infoline = "[{}] Python {}".format(
-                d["id"], d["version"].replace("\n", " -- ")
-            )
-            self.rewrite(infoline, newline=True)
-        self.setstatus(node.gateway.spec, "ok")
+            different_interpreter = d.get("executable") != sys.executable
+            if different_interpreter:
+                version = d["version"].replace("\n", " -- ")
+                self.rewrite(f"[{d['id']}] Python {version}", newline=True)
+        self.setstatus(
+            node.gateway.spec, WorkerStatus.ReadyForCollection, tests_collected=0
+        )
 
     @pytest.hookimpl
-    def pytest_testnodedown(self, node, error):
+    def pytest_testnodedown(self, node, error) -> None:
         if not error:
             return
         self.write_line(f"[{node.gateway.id}] node down: {error}")
@@ -457,3 +494,36 @@ def get_default_max_worker_restart(config):
         # if --max-worker-restart was not provided, use a reasonable default (#226)
         result = config.option.numprocesses * 4
     return result
+
+
+def get_workers_status_line(
+    status_and_items: Sequence[tuple[WorkerStatus, int]]
+) -> str:
+    """
+    Return the line to display during worker setup/collection based on the
+    status of the workers and number of tests collected for each.
+    """
+    statuses = [s for s, c in status_and_items]
+    total_workers = len(statuses)
+    workers_noun = "worker" if total_workers == 1 else "workers"
+    if status_and_items and all(s == WorkerStatus.CollectionDone for s in statuses):
+        # All workers collect the same number of items, so we grab
+        # the total number of items from the first worker.
+        first = status_and_items[0]
+        status, tests_collected = first
+        tests_noun = "item" if tests_collected == 1 else "items"
+        return f"{total_workers} {workers_noun} [{tests_collected} {tests_noun}]"
+    if WorkerStatus.CollectionDone in statuses:
+        done = sum(1 for s, c in status_and_items if c > 0)
+        return f"collecting: {done}/{total_workers} {workers_noun}"
+    if WorkerStatus.ReadyForCollection in statuses:
+        ready = statuses.count(WorkerStatus.ReadyForCollection)
+        return f"ready: {ready}/{total_workers} {workers_noun}"
+    if WorkerStatus.Initialized in statuses:
+        initialized = statuses.count(WorkerStatus.Initialized)
+        return f"initialized: {initialized}/{total_workers} {workers_noun}"
+    if WorkerStatus.Created in statuses:
+        created = statuses.count(WorkerStatus.Created)
+        return f"created: {created}/{total_workers} {workers_noun}"
+
+    return ""

src/xdist/newhooks.py

@@ -25,14 +25,18 @@ def pytest_xdist_newgateway(gateway):
 
 
 @pytest.hookspec(
-    warn_on_impl="rsync feature is deprecated and will be removed in pytest-xdist 4.0"
+    warn_on_impl=DeprecationWarning(
+        "rsync feature is deprecated and will be removed in pytest-xdist 4.0"
+    )
 )
 def pytest_xdist_rsyncstart(source, gateways):
     """called before rsyncing a directory to remote gateways takes place."""
 
 
 @pytest.hookspec(
-    warn_on_impl="rsync feature is deprecated and will be removed in pytest-xdist 4.0"
+    warn_on_impl=DeprecationWarning(
+        "rsync feature is deprecated and will be removed in pytest-xdist 4.0"
+    )
 )
 def pytest_xdist_rsyncfinish(source, gateways):
     """called after rsyncing a directory to remote gateways takes place."""

src/xdist/plugin.py

@@ -173,6 +173,8 @@ def pytest_addoption(parser):
             "one - might be useful for a small number of slow tests. "
             "Larger numbers will allow the scheduler to submit consecutive "
             "chunks of tests to workers - allows reusing fixtures. "
+            "Due to implementation reasons, at least 2 tests are scheduled per "
+            "worker at the start. Only later tests can be scheduled one by one. "
             "Unlimited if not set."
         ),
     )

src/xdist/remote.py

@@ -58,6 +58,7 @@ def worker_title(title):
 
 class WorkerInteractor:
     SHUTDOWN_MARK = object()
+    QUEUE_REPLACED_MARK = object()
 
     def __init__(self, config, channel):
         self.config = config
@@ -72,6 +73,15 @@ class WorkerInteractor:
     def _make_queue(self):
         return self.channel.gateway.execmodel.queue.Queue()
 
+    def _get_next_item_index(self):
+        """Gets the next item from test queue. Handles the case when the queue
+        is replaced concurrently in another thread.
+        """
+        result = self.torun.get()
+        while result is self.QUEUE_REPLACED_MARK:
+            result = self.torun.get()
+        return result
+
     def sendevent(self, name, **kwargs):
         self.log("sending", name, kwargs)
         self.channel.send((name, kwargs))
@@ -136,19 +146,22 @@ class WorkerInteractor:
                 self.torun.put(i)
 
         self.sendevent("unscheduled", indices=stolen)
+        old_queue.put(self.QUEUE_REPLACED_MARK)
 
     @pytest.hookimpl
     def pytest_runtestloop(self, session):
         self.log("entering main loop")
         self.channel.setcallback(self.handle_command, endmarker=self.SHUTDOWN_MARK)
-        self.nextitem_index = self.torun.get()
+        self.nextitem_index = self._get_next_item_index()
         while self.nextitem_index is not self.SHUTDOWN_MARK:
             self.run_one_test()
         return True
 
     def run_one_test(self):
+        self.item_index = self.nextitem_index
+        self.nextitem_index = self._get_next_item_index()
+
         items = self.session.items
-        self.item_index, self.nextitem_index = self.nextitem_index, self.torun.get()
         item = items[self.item_index]
         if self.nextitem_index is self.SHUTDOWN_MARK:
             nextitem = None

src/xdist/scheduler/loadscope.py

@@ -350,11 +350,18 @@ class LoadScopeScheduling:
             return
 
         # Determine chunks of work (scopes)
+        unsorted_workqueue = OrderedDict()
         for nodeid in self.collection:
             scope = self._split_scope(nodeid)
-            work_unit = self.workqueue.setdefault(scope, default=OrderedDict())
+            work_unit = unsorted_workqueue.setdefault(scope, default=OrderedDict())
             work_unit[nodeid] = False
 
+        # Insert tests scopes into work queue ordered by number of tests.
+        for scope, nodeids in sorted(
+            unsorted_workqueue.items(), key=lambda item: -len(item[1])
+        ):
+            self.workqueue[scope] = nodeids
+
         # Avoid having more workers than work
         extra_nodes = len(self.nodes) - len(self.workqueue)
 

testing/acceptance_test.py

@@ -101,21 +101,52 @@ class TestDistribution:
             """
         )
         result = pytester.runpytest(p1, "-v", "-d", "--tx=popen", "--tx=popen")
-        result.stdout.fnmatch_lines(["*1*Python*", "*2 failed, 1 passed, 1 skipped*"])
+        result.stdout.fnmatch_lines(
+            [
+                "created: 2/2 workers",
+                "*2 failed, 1 passed, 1 skipped*",
+            ]
+        )
         assert result.ret == 1
 
-    def test_n1_fail_minus_x(self, pytester: pytest.Pytester) -> None:
+    def test_exitfail_waits_for_workers_to_finish(
+        self, pytester: pytest.Pytester
+    ) -> None:
+        """The DSession waits for workers before exiting early on failure.
+
+        When -x/--exitfail is set, the DSession wait for the workers to finish
+        before raising an Interrupt exception. This prevents reports from the
+        faiing test and other tests from being discarded.
+        """
         p1 = pytester.makepyfile(
             """
+            import time
+
             def test_fail1():
+                time.sleep(0.1)
                 assert 0
             def test_fail2():
+                time.sleep(0.2)
+            def test_fail3():
+                time.sleep(0.3)
                 assert 0
+            def test_fail4():
+                time.sleep(0.3)
+            def test_fail5():
+                time.sleep(0.3)
+            def test_fail6():
+                time.sleep(0.3)
         """
         )
-        result = pytester.runpytest(p1, "-x", "-v", "-n1")
+        result = pytester.runpytest(p1, "-x", "-rA", "-v", "-n2")
         assert result.ret == 2
-        result.stdout.fnmatch_lines(["*Interrupted: stopping*1*", "*1 failed*"])
+        result.stdout.re_match_lines([".*Interrupted: stopping.*[12].*"])
+        m = re.search(r"== (\d+) failed, (\d+) passed in ", str(result.stdout))
+        assert m
+        n_failed, n_passed = (int(s) for s in m.groups())
+        assert 1 <= n_failed <= 2
+        assert 1 <= n_passed <= 3
+        assert (n_passed + n_failed) < 6
 
     def test_basetemp_in_subprocesses(self, pytester: pytest.Pytester) -> None:
         p1 = pytester.makepyfile(
@@ -151,7 +182,12 @@ class TestDistribution:
         """
         )
         result = pytester.runpytest(p1, "-d", "-v")
-        result.stdout.fnmatch_lines(["*2*Python*", "*2 failed, 1 passed, 1 skipped*"])
+        result.stdout.fnmatch_lines(
+            [
+                "created: 3/3 workers",
+                "*2 failed, 1 passed, 1 skipped*",
+            ]
+        )
         assert result.ret == 1
 
     def test_dist_tests_with_crash(self, pytester: pytest.Pytester) -> None:
@@ -237,9 +273,6 @@ class TestDistribution:
         assert result.ret == 0
         result.stdout.fnmatch_lines(
             [
-                "*0* *cwd*",
-                # "RSyncStart: [G1]",
-                # "RSyncFinished: [G1]",
                 "*1 passed*",
             ]
         )
@@ -276,7 +309,11 @@ class TestDistribution:
         p1 = pytester.makepyfile("def test_func(): pass")
         result = pytester.runpytest("-v", p1, "-d", "--tx=popen")
         result.stdout.fnmatch_lines(
-            ["*0*Python*", "*calculated result is 49*", "*1 passed*"]
+            [
+                "created: 1/1 worker",
+                "*calculated result is 49*",
+                "*1 passed*",
+            ]
         )
         assert result.ret == 0
 
@@ -393,14 +430,14 @@ class TestTerminalReporting:
         out = result.stdout.str()
         if verbosity == "-v":
             assert "scheduling tests" in out
-            assert "gw" in out
+            assert "1 worker [1 item]" in out
         elif verbosity == "-q":
             assert "scheduling tests" not in out
             assert "gw" not in out
             assert "bringing up nodes..." in out
         else:
             assert "scheduling tests" not in out
-            assert "gw" in out
+            assert "1 worker [1 item]" in out
 
     def test_pass_skip_fail(self, pytester: pytest.Pytester) -> None:
         pytester.makepyfile(
@@ -1099,8 +1136,9 @@ def test_color_yes_collection_on_non_atty(pytester, request) -> None:
     result = pytester.runpytest(*args)
     assert "test session starts" in result.stdout.str()
     assert "\x1b[1m" in result.stdout.str()
-    assert "gw0 [10] / gw1 [10]" in result.stdout.str()
-    assert "gw0 C / gw1 C" not in result.stdout.str()
+    assert "created: 2/2 workers" in result.stdout.str()
+    assert "2 workers [10 items]" in result.stdout.str()
+    assert "collecting:" not in result.stdout.str()
 
 
 def test_without_terminal_plugin(pytester, request) -> None:
@@ -1138,7 +1176,7 @@ def test_internal_error_with_maxfail(pytester: pytest.Pytester) -> None:
     """
     )
     result = pytester.runpytest_subprocess("--maxfail=1", "-n1")
-    result.stdout.fnmatch_lines(["* 1 error in *"])
+    result.stdout.re_match_lines([".* [12] errors? in .*"])
     assert "INTERNALERROR" not in result.stderr.str()
 
 
@@ -1194,6 +1232,22 @@ class TestLoadScope:
             "test_a.py::TestB", result.outlines
         ) in ({"gw0": 10}, {"gw1": 10})
 
+    def test_workqueue_ordered_by_size(self, pytester: pytest.Pytester) -> None:
+        test_file = """
+            import pytest
+            @pytest.mark.parametrize('i', range({}))
+            def test(i):
+                pass
+        """
+        pytester.makepyfile(test_a=test_file.format(10), test_b=test_file.format(20))
+        result = pytester.runpytest("-n2", "--dist=loadscope", "-v")
+        assert get_workers_and_test_count_by_prefix(
+            "test_a.py::test", result.outlines
+        ) == {"gw1": 10}
+        assert get_workers_and_test_count_by_prefix(
+            "test_b.py::test", result.outlines
+        ) == {"gw0": 20}
+
     def test_module_single_start(self, pytester: pytest.Pytester) -> None:
         """Fix test suite never finishing in case all workers start with a single test (#277)."""
         test_file1 = """
@@ -1554,8 +1608,8 @@ def test_collection_crash(testdir):
     assert result.ret == 1
     result.stdout.fnmatch_lines(
         [
-            "gw0 I",
-            "gw0 [[]0[]]",
+            "created: 1/1 worker",
+            "1 worker [[]0 items[]]",
             "*_ ERROR collecting test_collection_crash.py _*",
             "E   assert 0",
             "*= 1 error in *",

testing/test_dsession.py

@@ -1,7 +1,13 @@
-from xdist.dsession import DSession, get_default_max_worker_restart
+from __future__ import annotations
+from xdist.dsession import (
+    DSession,
+    get_default_max_worker_restart,
+    get_workers_status_line,
+    WorkerStatus,
+)
 from xdist.report import report_collection_diff
 from xdist.scheduler import EachScheduling, LoadScheduling, WorkStealingScheduling
-from typing import Optional
+from typing import Sequence
 
 import pytest
 import execnet
@@ -473,7 +479,7 @@ def test_report_collection_diff_equal() -> None:
 def test_default_max_worker_restart() -> None:
     class config:
         class option:
-            maxworkerrestart: Optional[str] = None
+            maxworkerrestart: str | None = None
             numprocesses: int = 0
 
     assert get_default_max_worker_restart(config) is None
@@ -527,3 +533,70 @@ def test_pytest_issue419(pytester: pytest.Pytester) -> None:
     reprec = pytester.inline_run("-n1")
     reprec.assertoutcome(passed=2)
     assert 0
+
+
+Created = WorkerStatus.Created
+Initialized = WorkerStatus.Initialized
+ReadyForCollection = WorkerStatus.ReadyForCollection
+CollectionDone = WorkerStatus.CollectionDone
+
+
+@pytest.mark.parametrize(
+    "status_and_items, expected",
+    [
+        (
+            [],
+            "",
+        ),
+        (
+            [(Created, 0)],
+            "created: 1/1 worker",
+        ),
+        (
+            [(Created, 0), (Created, 0)],
+            "created: 2/2 workers",
+        ),
+        (
+            [(Initialized, 0), (Created, 0)],
+            "initialized: 1/2 workers",
+        ),
+        (
+            [(Initialized, 0), (Initialized, 0)],
+            "initialized: 2/2 workers",
+        ),
+        (
+            [(ReadyForCollection, 0), (Created, 0)],
+            "ready: 1/2 workers",
+        ),
+        (
+            [(ReadyForCollection, 0), (ReadyForCollection, 0)],
+            "ready: 2/2 workers",
+        ),
+        (
+            [(CollectionDone, 12), (Created, 0)],
+            "collecting: 1/2 workers",
+        ),
+        (
+            [(CollectionDone, 12), (CollectionDone, 12)],
+            "2 workers [12 items]",
+        ),
+        (
+            [(CollectionDone, 1), (CollectionDone, 1)],
+            "2 workers [1 item]",
+        ),
+        (
+            [(CollectionDone, 1)],
+            "1 worker [1 item]",
+        ),
+        # Different number of tests collected will raise an error and should not happen in practice,
+        # but we test for it anyway.
+        (
+            [(CollectionDone, 1), (CollectionDone, 12)],
+            "2 workers [1 item]",
+        ),
+    ],
+)
+def test_get_workers_status_line(
+    status_and_items: Sequence[tuple[WorkerStatus, int]], expected: str
+) -> None:
+    assert get_workers_status_line(status_and_items) == expected

testing/test_remote.py

@@ -271,6 +271,40 @@ class TestWorkerInteractor:
         ev = worker.popevent("workerfinished")
         assert "workeroutput" in ev.kwargs
 
+    def test_steal_empty_queue(self, worker: WorkerSetup, unserialize_report) -> None:
+        worker.pytester.makepyfile(
+            """
+            def test_func(): pass
+            def test_func2(): pass
+        """
+        )
+        worker.setup()
+        ev = worker.popevent("collectionfinish")
+        ids = ev.kwargs["ids"]
+        assert len(ids) == 2
+        worker.sendcommand("runtests_all")
+
+        for when in ["setup", "call", "teardown"]:
+            ev = worker.popevent("testreport")
+            rep = unserialize_report(ev.kwargs["data"])
+            assert rep.nodeid.endswith("::test_func")
+            assert rep.when == when
+
+        worker.sendcommand("steal", indices=[0, 1])
+        ev = worker.popevent("unscheduled")
+        assert ev.kwargs["indices"] == []
+
+        worker.sendcommand("shutdown")
+
+        for when in ["setup", "call", "teardown"]:
+            ev = worker.popevent("testreport")
+            rep = unserialize_report(ev.kwargs["data"])
+            assert rep.nodeid.endswith("::test_func2")
+            assert rep.when == when
+
+        ev = worker.popevent("workerfinished")
+        assert "workeroutput" in ev.kwargs
+
 
 def test_remote_env_vars(pytester: pytest.Pytester) -> None:
     pytester.makepyfile(

testing/test_workermanage.py

@@ -374,7 +374,7 @@ def test_unserialize_warning_msg(w_cls):
         v1 = getattr(w_msg, k)
         v2 = getattr(w_msg2, k)
         if k == "message":
-            assert type(v1) == type(v2)
+            assert type(v1) is type(v2)
             assert v1.args == v2.args
         else:
             assert v1 == v2

tox.ini

@@ -1,7 +1,7 @@
 [tox]
 envlist=
   linting
-  py{37,38,39,310,311}-pytestlatest
+  py{37,38,39,310,311,312}-pytestlatest
   py310-pytestmain
   py310-psutil
   py310-setproctitle