Merge pull request #1060 from bluetech/pre-typing-fixes

Pre-typing fixes/improvements

.pre-commit-config.yaml

@@ -31,4 +31,4 @@ repos:
         additional_dependencies:
           - pytest>=7.0.0
           - execnet>=2.1.0
-          - py>=1.10.0
+          - types-psutil

pyproject.toml

@@ -85,6 +85,7 @@ select = [
     "W",        # pycodestyle
     "T10",      # flake8-debugger
     "PIE",      # flake8-pie
+    "FA",       # flake8-future-annotations
     "PGH",      # pygrep-hooks
     "PLE",      # pylint error
     "PLW",      # pylint warning
@@ -135,6 +136,7 @@ lines-after-imports = 2
 
 [tool.mypy]
 mypy_path = ["src"]
+files = ["src", "testing"]
 # TODO: Enable this & fix errors.
 # check_untyped_defs = true
 disallow_any_generics = true

src/xdist/dsession.py

@@ -317,13 +317,6 @@ class DSession:
         assert not rep.passed
         self._failed_worker_collectreport(node, rep)
 
-    def worker_warning_captured(self, warning_message, when, item):
-        """Emitted when a node calls the pytest_warning_captured hook (deprecated in 6.0)."""
-        # This hook as been removed in pytest 7.1, and we can remove support once we only
-        # support pytest >=7.1.
-        kwargs = dict(warning_message=warning_message, when=when, item=item)
-        self.config.hook.pytest_warning_captured.call_historic(kwargs=kwargs)
-
     def worker_warning_recorded(self, warning_message, when, nodeid, location):
         """Emitted when a node calls the pytest_warning_recorded hook."""
         kwargs = dict(
@@ -374,10 +367,9 @@ class DSession:
     def handle_crashitem(self, nodeid, worker):
         # XXX get more reporting info by recording pytest_runtest_logstart?
         # XXX count no of failures and retry N times
-        runner = self.config.pluginmanager.getplugin("runner")
         fspath = nodeid.split("::")[0]
         msg = f"worker {worker.gateway.id!r} crashed while running {nodeid!r}"
-        rep = runner.TestReport(
+        rep = pytest.TestReport(
             nodeid, (fspath, None, fspath), (), "failed", msg, "???"
         )
         rep.node = worker

src/xdist/looponfail.py

@@ -7,11 +7,12 @@ processes) otherwise changes to source code can crash
 the controlling process which should best never happen.
 """
 
+from __future__ import annotations
+
 import os
 from pathlib import Path
 import sys
 import time
-from typing import Dict
 from typing import Sequence
 
 from _pytest._io import TerminalWriter
@@ -45,7 +46,7 @@ def pytest_cmdline_main(config):
         return 2  # looponfail only can get stop with ctrl-C anyway
 
 
-def looponfail_main(config: "pytest.Config") -> None:
+def looponfail_main(config: pytest.Config) -> None:
     remotecontrol = RemoteControl(config)
     config_roots = config.getini("looponfailroots")
     if not config_roots:
@@ -79,9 +80,7 @@ class RemoteControl:
     def initgateway(self):
         return execnet.makegateway("popen")
 
-    def setup(self, out=None):
+    def setup(self):
-        if out is None:
-            out = TerminalWriter()
         if hasattr(self, "gateway"):
             raise ValueError("already have gateway %r" % self.gateway)
         self.trace("setting up worker session")
@@ -93,6 +92,8 @@ class RemoteControl:
         )
         remote_outchannel = channel.receive()
 
+        out = TerminalWriter()
+
         def write(s):
             out._file.write(s)
             out._file.flush()
@@ -238,7 +239,7 @@ class WorkerFailSession:
 class StatRecorder:
     def __init__(self, rootdirlist: Sequence[Path]) -> None:
         self.rootdirlist = rootdirlist
-        self.statcache: Dict[Path, os.stat_result] = {}
+        self.statcache: dict[Path, os.stat_result] = {}
         self.check()  # snapshot state
 
     def fil(self, p: Path) -> bool:
@@ -256,7 +257,7 @@ class StatRecorder:
 
     def check(self, removepycfiles: bool = True) -> bool:
         changed = False
-        newstat: Dict[Path, os.stat_result] = {}
+        newstat: dict[Path, os.stat_result] = {}
         for rootdir in self.rootdirlist:
             for path in visit_path(rootdir, filter=self.fil, recurse=self.rec):
                 oldstat = self.statcache.pop(path, None)

src/xdist/remote.py

@@ -7,6 +7,7 @@ needs not to be installed in remote environments.
 """
 
 import contextlib
+import enum
 import os
 import sys
 import time
@@ -57,10 +58,12 @@ def worker_title(title):
         pass
 
 
-class WorkerInteractor:
+class Marker(enum.Enum):
-    SHUTDOWN_MARK = object()
+    SHUTDOWN = 0
-    QUEUE_REPLACED_MARK = object()
+    QUEUE_REPLACED = 1
 
+
+class WorkerInteractor:
     def __init__(self, config, channel):
         self.config = config
         self.workerid = config.workerinput.get("workerid", "?")
@@ -79,7 +82,7 @@ class WorkerInteractor:
         is replaced concurrently in another thread.
         """
         result = self.torun.get()
-        while result is self.QUEUE_REPLACED_MARK:
+        while result is Marker.QUEUE_REPLACED:
             result = self.torun.get()
         return result
 
@@ -114,8 +117,8 @@ class WorkerInteractor:
         self.sendevent("collectionstart")
 
     def handle_command(self, command):
-        if command is self.SHUTDOWN_MARK:
+        if command is Marker.SHUTDOWN:
-            self.torun.put(self.SHUTDOWN_MARK)
+            self.torun.put(Marker.SHUTDOWN)
             return
 
         name, kwargs = command
@@ -128,7 +131,7 @@ class WorkerInteractor:
             for i in range(len(self.session.items)):
                 self.torun.put(i)
         elif name == "shutdown":
-            self.torun.put(self.SHUTDOWN_MARK)
+            self.torun.put(Marker.SHUTDOWN)
         elif name == "steal":
             self.steal(kwargs["indices"])
 
@@ -149,14 +152,14 @@ class WorkerInteractor:
                 self.torun.put(i)
 
         self.sendevent("unscheduled", indices=stolen)
-        old_queue.put(self.QUEUE_REPLACED_MARK)
+        old_queue.put(Marker.QUEUE_REPLACED)
 
     @pytest.hookimpl
     def pytest_runtestloop(self, session):
         self.log("entering main loop")
-        self.channel.setcallback(self.handle_command, endmarker=self.SHUTDOWN_MARK)
+        self.channel.setcallback(self.handle_command, endmarker=Marker.SHUTDOWN)
         self.nextitem_index = self._get_next_item_index()
-        while self.nextitem_index is not self.SHUTDOWN_MARK:
+        while self.nextitem_index is not Marker.SHUTDOWN:
             self.run_one_test()
             if session.shouldfail or session.shouldstop:
                 break
@@ -168,16 +171,16 @@ class WorkerInteractor:
 
         items = self.session.items
         item = items[self.item_index]
-        if self.nextitem_index is self.SHUTDOWN_MARK:
+        if self.nextitem_index is Marker.SHUTDOWN:
             nextitem = None
         else:
             nextitem = items[self.nextitem_index]
 
         worker_title("[pytest-xdist running] %s" % item.nodeid)
 
-        start = time.time()
+        start = time.perf_counter()
         self.config.hook.pytest_runtest_protocol(item=item, nextitem=nextitem)
-        duration = time.time() - start
+        duration = time.perf_counter() - start
 
         worker_title("[pytest-xdist idle]")
 

src/xdist/scheduler/each.py

@@ -101,12 +101,7 @@ class EachScheduling:
         self.node2pending[node].remove(item_index)
 
     def mark_test_pending(self, item):
-        self.pending.insert(
+        raise NotImplementedError()
-            0,
-            self.collection.index(item),
-        )
-        for node in self.node2pending:
-            self.check_schedule(node)
 
     def remove_node(self, node):
         # KeyError if we didn't get an add_node() yet

src/xdist/scheduler/loadscope.py

@@ -90,8 +90,8 @@ class LoadScopeScheduling:
         self.collection = None
 
         self.workqueue = OrderedDict()
-        self.assigned_work = OrderedDict()
+        self.assigned_work = {}
-        self.registered_collections = OrderedDict()
+        self.registered_collections = {}
 
         if log is None:
             self.log = Producer("loadscopesched")
@@ -156,7 +156,7 @@ class LoadScopeScheduling:
         bootstraps a new node.
         """
         assert node not in self.assigned_work
-        self.assigned_work[node] = OrderedDict()
+        self.assigned_work[node] = {}
 
     def remove_node(self, node):
         """Remove a node from the scheduler.
@@ -252,7 +252,7 @@ class LoadScopeScheduling:
         scope, work_unit = self.workqueue.popitem(last=False)
 
         # Keep track of the assigned work
-        assigned_to_node = self.assigned_work.setdefault(node, default=OrderedDict())
+        assigned_to_node = self.assigned_work.setdefault(node, {})
         assigned_to_node[scope] = work_unit
 
         # Ask the node to execute the workload
@@ -349,10 +349,10 @@ class LoadScopeScheduling:
             return
 
         # Determine chunks of work (scopes)
-        unsorted_workqueue = OrderedDict()
+        unsorted_workqueue = {}
         for nodeid in self.collection:
             scope = self._split_scope(nodeid)
-            work_unit = unsorted_workqueue.setdefault(scope, default=OrderedDict())
+            work_unit = unsorted_workqueue.setdefault(scope, {})
             work_unit[nodeid] = False
 
         # Insert tests scopes into work queue ordered by number of tests.
@@ -368,7 +368,7 @@ class LoadScopeScheduling:
             self.log(f"Shutting down {extra_nodes} nodes")
 
             for _ in range(extra_nodes):
-                unused_node, assigned = self.assigned_work.popitem(last=True)
+                unused_node, assigned = self.assigned_work.popitem()
 
                 self.log(f"Shutting down unused node {unused_node}")
                 unused_node.shutdown()
@@ -407,9 +407,6 @@ class LoadScopeScheduling:
             same_collection = False
             self.log(msg)
 
-            if self.config is None:
-                continue
-
             rep = pytest.CollectReport(
                 nodeid=node.gateway.id,
                 outcome="failed",

src/xdist/workermanage.py

@@ -1,14 +1,13 @@
+from __future__ import annotations
+
+import enum
 import fnmatch
 import os
 from pathlib import Path
 import re
 import sys
 from typing import Any
-from typing import List
-from typing import Optional
 from typing import Sequence
-from typing import Set
-from typing import Tuple
 from typing import Union
 import uuid
 
@@ -60,7 +59,7 @@ class NodeManager:
             self.specs.append(spec)
         self.roots = self._getrsyncdirs()
         self.rsyncoptions = self._getrsyncoptions()
-        self._rsynced_specs: Set[Tuple[Any, Any]] = set()
+        self._rsynced_specs: set[tuple[Any, Any]] = set()
 
     def rsync_roots(self, gateway):
         """Rsync the set of roots to the node's gateway cwd."""
@@ -89,7 +88,7 @@ class NodeManager:
     def _getxspecs(self):
         return [execnet.XSpec(x) for x in parse_spec_config(self.config)]
 
-    def _getrsyncdirs(self) -> List[Path]:
+    def _getrsyncdirs(self) -> list[Path]:
         for spec in self.specs:
             if not spec.popen or spec.chdir:
                 break
@@ -174,7 +173,7 @@ class HostRSync(execnet.RSync):
         self,
         sourcedir: PathLike,
         *,
-        ignores: Optional[Sequence[PathLike]] = None,
+        ignores: Sequence[PathLike] | None = None,
         verbose: bool = True,
     ) -> None:
         if ignores is None:
@@ -201,7 +200,7 @@ class HostRSync(execnet.RSync):
             print(f"{gateway.spec}:{remotepath} <= {path}")
 
 
-def make_reltoroot(roots: Sequence[Path], args: List[str]) -> List[str]:
+def make_reltoroot(roots: Sequence[Path], args: list[str]) -> list[str]:
     # XXX introduce/use public API for splitting pytest args
     splitcode = "::"
     result = []
@@ -216,7 +215,7 @@ def make_reltoroot(roots: Sequence[Path], args: List[str]) -> List[str]:
             result.append(arg)
             continue
         for root in roots:
-            x: Optional[Path]
+            x: Path | None
             try:
                 x = fspath.relative_to(root)
             except ValueError:
@@ -230,9 +229,11 @@ def make_reltoroot(roots: Sequence[Path], args: List[str]) -> List[str]:
     return result
 
 
-class WorkerController:
+class Marker(enum.Enum):
-    ENDMARK = -1
+    END = -1
 
+
+class WorkerController:
     class RemoteHook:
         @pytest.hookimpl(trylast=True)
         def pytest_xdist_getremotemodule(self):
@@ -283,7 +284,7 @@ class WorkerController:
         self.channel.send((self.workerinput, args, option_dict, change_sys_path))
 
         if self.putevent:
-            self.channel.setcallback(self.process_from_remote, endmarker=self.ENDMARK)
+            self.channel.setcallback(self.process_from_remote, endmarker=Marker.END)
 
     def ensure_teardown(self):
         if hasattr(self, "channel"):
@@ -331,7 +332,7 @@ class WorkerController:
         avoid raising exceptions or doing heavy work.
         """
         try:
-            if eventcall == self.ENDMARK:
+            if eventcall is Marker.END:
                 err = self.channel._getremoteerror()
                 if not self._down:
                     if not err or isinstance(err, EOFError):
@@ -374,16 +375,6 @@ class WorkerController:
                     nodeid=kwargs["nodeid"],
                     fslocation=kwargs["nodeid"],
                 )
-            elif eventname == "warning_captured":
-                warning_message = unserialize_warning_message(
-                    kwargs["warning_message_data"]
-                )
-                self.notify_inproc(
-                    eventname,
-                    warning_message=warning_message,
-                    when=kwargs["when"],
-                    item=kwargs["item"],
-                )
             elif eventname == "warning_recorded":
                 warning_message = unserialize_warning_message(
                     kwargs["warning_message_data"]

testing/acceptance_test.py

@@ -1,9 +1,8 @@
+from __future__ import annotations
+
 import os
 import re
 import shutil
-from typing import Dict
-from typing import List
-from typing import Tuple
 
 import pytest
 
@@ -1510,13 +1509,17 @@ class TestLocking:
 
     """ + ((_test_content * 4) % ("A", "B", "C", "D"))
 
-    @pytest.mark.parametrize("scope", ["each", "load", "loadscope", "loadfile", "no"])
+    @pytest.mark.parametrize(
+        "scope", ["each", "load", "loadscope", "loadfile", "worksteal", "no"]
+    )
     def test_single_file(self, pytester, scope) -> None:
         pytester.makepyfile(test_a=self.test_file1)
         result = pytester.runpytest("-n2", "--dist=%s" % scope, "-v")
         result.assert_outcomes(passed=(12 if scope != "each" else 12 * 2))
 
-    @pytest.mark.parametrize("scope", ["each", "load", "loadscope", "loadfile", "no"])
+    @pytest.mark.parametrize(
+        "scope", ["each", "load", "loadscope", "loadfile", "worksteal", "no"]
+    )
     def test_multi_file(self, pytester, scope) -> None:
         pytester.makepyfile(
             test_a=self.test_file1,
@@ -1528,7 +1531,7 @@ class TestLocking:
         result.assert_outcomes(passed=(48 if scope != "each" else 48 * 2))
 
 
-def parse_tests_and_workers_from_output(lines: List[str]) -> List[Tuple[str, str, str]]:
+def parse_tests_and_workers_from_output(lines: list[str]) -> list[tuple[str, str, str]]:
     result = []
     for line in lines:
         # example match: "[gw0] PASSED test_a.py::test[7]"
@@ -1550,9 +1553,9 @@ def parse_tests_and_workers_from_output(lines: List[str]) -> List[Tuple[str, str
 
 
 def get_workers_and_test_count_by_prefix(
-    prefix: str, lines: List[str], expected_status: str = "PASSED"
+    prefix: str, lines: list[str], expected_status: str = "PASSED"
-) -> Dict[str, int]:
+) -> dict[str, int]:
-    result: Dict[str, int] = {}
+    result: dict[str, int] = {}
     for worker, status, nodeid in parse_tests_and_workers_from_output(lines):
         if expected_status == status and nodeid.startswith(prefix):
             result[worker] = result.get(worker, 0) + 1

testing/conftest.py

@@ -1,5 +1,6 @@
+from __future__ import annotations
+
 import shutil
-from typing import List
 
 import execnet
 import pytest
@@ -41,7 +42,7 @@ def specssh(request) -> str:
 
 
 # configuration information for tests
-def getgspecs(config) -> List[execnet.XSpec]:
+def getgspecs(config) -> list[execnet.XSpec]:
     return [execnet.XSpec(spec) for spec in config.getvalueorskip("gspecs")]
 
 

testing/test_dsession.py

@@ -442,7 +442,7 @@ class TestDistReporter:
     @pytest.mark.xfail
     def test_rsync_printing(self, pytester: pytest.Pytester, linecomp) -> None:
         config = pytester.parseconfig()
-        from _pytest.pytest_terminal import TerminalReporter
+        from _pytest.terminal import TerminalReporter
 
         rep = TerminalReporter(config, file=linecomp.stringio)
         config.pluginmanager.register(rep, "terminalreporter")

testing/test_looponfail.py

@@ -1,9 +1,10 @@
+from __future__ import annotations
+
 import pathlib
 from pathlib import Path
 import shutil
 import tempfile
 import textwrap
-from typing import List
 import unittest.mock
 
 import pytest
@@ -75,7 +76,7 @@ class TestStatRecorder:
         # make check()'s visit() call return our just removed
         # path as if we were in a race condition
         dirname = str(tmp)
-        dirnames: List[str] = []
+        dirnames: list[str] = []
         filenames = [str(p)]
         with unittest.mock.patch(
             "os.walk", return_value=[(dirname, dirnames, filenames)], autospec=True
@@ -122,7 +123,7 @@ class TestRemoteControl:
         item = pytester.getitem("def test_func():\n assert 0\n")
         control = RemoteControl(item.config)
         control.setup()
-        failures = control.runsession()
+        failures = control.runsession()[0]
         assert failures
         control.setup()
         item.path.write_text("def test_func():\n assert 1\n")