diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index af89906..0595436 100644 --- a/.pre-commit-config.yaml +++ b/.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 diff --git a/pyproject.toml b/pyproject.toml index 7ae28b1..71d429d 100644 --- a/pyproject.toml +++ b/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 diff --git a/src/xdist/dsession.py b/src/xdist/dsession.py index 56b332f..6590707 100644 --- a/src/xdist/dsession.py +++ b/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 diff --git a/src/xdist/looponfail.py b/src/xdist/looponfail.py index 05b38f0..8c2a60a 100644 --- a/src/xdist/looponfail.py +++ b/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): - if out is None: - out = TerminalWriter() + def setup(self): 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) diff --git a/src/xdist/remote.py b/src/xdist/remote.py index fc3a3ff..ac1bf1c 100644 --- a/src/xdist/remote.py +++ b/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: - SHUTDOWN_MARK = object() - QUEUE_REPLACED_MARK = object() +class Marker(enum.Enum): + SHUTDOWN = 0 + 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: - self.torun.put(self.SHUTDOWN_MARK) + if command is Marker.SHUTDOWN: + 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]") diff --git a/src/xdist/scheduler/each.py b/src/xdist/scheduler/each.py index 91084bf..dab0ff8 100644 --- a/src/xdist/scheduler/each.py +++ b/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( - 0, - self.collection.index(item), - ) - for node in self.node2pending: - self.check_schedule(node) + raise NotImplementedError() def remove_node(self, node): # KeyError if we didn't get an add_node() yet diff --git a/src/xdist/scheduler/loadscope.py b/src/xdist/scheduler/loadscope.py index e8addbb..076840c 100644 --- a/src/xdist/scheduler/loadscope.py +++ b/src/xdist/scheduler/loadscope.py @@ -90,8 +90,8 @@ class LoadScopeScheduling: self.collection = None self.workqueue = OrderedDict() - self.assigned_work = OrderedDict() - self.registered_collections = OrderedDict() + self.assigned_work = {} + 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", diff --git a/src/xdist/workermanage.py b/src/xdist/workermanage.py index d2f790f..c3793ef 100644 --- a/src/xdist/workermanage.py +++ b/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: - ENDMARK = -1 +class Marker(enum.Enum): + 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"] diff --git a/testing/acceptance_test.py b/testing/acceptance_test.py index e63443a..d17ddf0 100644 --- a/testing/acceptance_test.py +++ b/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" -) -> Dict[str, int]: - result: Dict[str, int] = {} + prefix: str, lines: list[str], expected_status: str = "PASSED" +) -> 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 diff --git a/testing/conftest.py b/testing/conftest.py index 195fb87..70bdfdf 100644 --- a/testing/conftest.py +++ b/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")] diff --git a/testing/test_dsession.py b/testing/test_dsession.py index 3ce205c..2a32a46 100644 --- a/testing/test_dsession.py +++ b/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") diff --git a/testing/test_looponfail.py b/testing/test_looponfail.py index 348aa3e..e2fa02d 100644 --- a/testing/test_looponfail.py +++ b/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")