IQ.Lvbs/pytest-xdist
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Improve typing

Browse Source 
Fix #1057
This commit is contained in:
Ran Benita
2023-04-18 23:06:34 +03:00
parent 5dfc590a8c
commit b78cf1c0ce
24 changed files with 858 additions and 477 deletions
Download Patch File Download Diff File Expand all files Collapse all files

180
src/xdist/dsession.py
View File

@@ -5,11 +5,15 @@ from enum import Enum
from queue import Empty
from queue import Queue
import sys
from typing import Any
from typing import Sequence
import warnings
import execnet
import pytest
from xdist.remote import Producer
from xdist.remote import WorkerInfo
from xdist.scheduler import EachScheduling
from xdist.scheduler import LoadFileScheduling
from xdist.scheduler import LoadGroupScheduling
@@ -18,6 +22,7 @@ from xdist.scheduler import LoadScopeScheduling
from xdist.scheduler import Scheduling
from xdist.scheduler import WorkStealingScheduling
from xdist.workermanage import NodeManager
from xdist.workermanage import WorkerController
class Interrupted(KeyboardInterrupt):
@@ -38,29 +43,31 @@ class DSession:
it will wait for instructions.
"""
def __init__(self, config):
shouldstop: bool | str
def __init__(self, config: pytest.Config) -> None:
self.config = config
self.log = Producer("dsession", enabled=config.option.debug)
self.nodemanager = None
self.sched = None
self.nodemanager: NodeManager | None = None
self.sched: Scheduling | None = None
self.shuttingdown = False
self.countfailures = 0
self.maxfail = config.getvalue("maxfail")
self.queue = Queue()
self._session = None
self._failed_collection_errors = {}
self._active_nodes = set()
self.maxfail: int = config.getvalue("maxfail")
self.queue: Queue[tuple[str, dict[str, Any]]] = Queue()
self._session: pytest.Session | None = None
self._failed_collection_errors: dict[object, bool] = {}
self._active_nodes: set[WorkerController] = set()
self._failed_nodes_count = 0
self._max_worker_restart = get_default_max_worker_restart(self.config)
# summary message to print at the end of the session
self._summary_report = None
self._summary_report: str | None = None
self.terminal = config.pluginmanager.getplugin("terminalreporter")
if self.terminal:
self.trdist = TerminalDistReporter(config)
config.pluginmanager.register(self.trdist, "terminaldistreporter")
@property
def session_finished(self):
def session_finished(self) -> bool:
"""Return True if the distributed session has finished.
This means all nodes have executed all test items. This is
@@ -68,12 +75,12 @@ class DSession:
"""
return bool(self.shuttingdown and not self._active_nodes)
def report_line(self, line):
def report_line(self, line: str) -> None:
if self.terminal and self.config.option.verbose >= 0:
self.terminal.write_line(line)
@pytest.hookimpl(trylast=True)
def pytest_sessionstart(self, session):
def pytest_sessionstart(self, session: pytest.Session) -> None:
"""Creates and starts the nodes.
The nodes are setup to put their events onto self.queue. As
@@ -85,7 +92,7 @@ class DSession:
self._session = session
@pytest.hookimpl
def pytest_sessionfinish(self, session):
def pytest_sessionfinish(self) -> None:
"""Shutdown all nodes."""
nm = getattr(self, "nodemanager", None) # if not fully initialized
if nm is not None:
@@ -93,12 +100,16 @@ class DSession:
self._session = None
@pytest.hookimpl
def pytest_collection(self):
def pytest_collection(self) -> bool:
# prohibit collection of test items in controller process
return True
@pytest.hookimpl(trylast=True)
def pytest_xdist_make_scheduler(self, config, log) -> Scheduling | None:
def pytest_xdist_make_scheduler(
self,
config: pytest.Config,
log: Producer,
) -> Scheduling | None:
dist = config.getvalue("dist")
if dist == "each":
return EachScheduling(config, log)
@@ -115,7 +126,7 @@ class DSession:
return None
@pytest.hookimpl
def pytest_runtestloop(self):
def pytest_runtestloop(self) -> bool:
self.sched = self.config.hook.pytest_xdist_make_scheduler(
config=self.config, log=self.log
)
@@ -132,7 +143,7 @@ class DSession:
raise pending_exception
return True
def loop_once(self):
def loop_once(self) -> None:
"""Process one callback from one of the workers."""
while 1:
if not self._active_nodes:
@@ -150,6 +161,7 @@ class DSession:
call = getattr(self, method)
self.log("calling method", method, kwargs)
call(**kwargs)
assert self.sched is not None
if self.sched.tests_finished:
self.triggershutdown()
@@ -157,7 +169,11 @@ class DSession:
# callbacks for processing events from workers
#
def worker_workerready(self, node, workerinfo):
def worker_workerready(
self,
node: WorkerController,
workerinfo: WorkerInfo,
) -> None:
"""Emitted when a node first starts up.
This adds the node to the scheduler, nodes continue with
@@ -171,9 +187,10 @@ class DSession:
if self.shuttingdown:
node.shutdown()
else:
assert self.sched is not None
self.sched.add_node(node)
def worker_workerfinished(self, node):
def worker_workerfinished(self, node: WorkerController) -> None:
"""Emitted when node executes its pytest_sessionfinish hook.
Removes the node from the scheduler.
@@ -194,12 +211,15 @@ class DSession:
self.shouldstop = shouldx
break
else:
assert self.sched is not None
if node in self.sched.nodes:
crashitem = self.sched.remove_node(node)
assert not crashitem, (crashitem, node)
self._active_nodes.remove(node)
def worker_internal_error(self, node, formatted_error):
def worker_internal_error(
self, node: WorkerController, formatted_error: str
) -> None:
"""
pytest_internalerror() was called on the worker.
@@ -215,9 +235,10 @@ class DSession:
excrepr = excinfo.getrepr()
self.config.hook.pytest_internalerror(excrepr=excrepr, excinfo=excinfo)
def worker_errordown(self, node, error):
def worker_errordown(self, node: WorkerController, error: object | None) -> None:
"""Emitted by the WorkerController when a node dies."""
self.config.hook.pytest_testnodedown(node=node, error=error)
assert self.sched is not None
try:
crashitem = self.sched.remove_node(node)
except KeyError:
@@ -235,7 +256,7 @@ class DSession:
if self._max_worker_restart == 0:
msg = f"worker {node.gateway.id} crashed and worker restarting disabled"
else:
msg = "maximum crashed workers reached: %d" % self._max_worker_restart
msg = f"maximum crashed workers reached: {self._max_worker_restart}"
self._summary_report = msg
self.report_line("\n" + msg)
self.triggershutdown()
@@ -246,11 +267,13 @@ class DSession:
self._active_nodes.remove(node)
@pytest.hookimpl
def pytest_terminal_summary(self, terminalreporter):
def pytest_terminal_summary(self, terminalreporter: Any) -> None:
if self.config.option.verbose >= 0 and self._summary_report:
terminalreporter.write_sep("=", f"xdist: {self._summary_report}")
def worker_collectionfinish(self, node, ids):
def worker_collectionfinish(
self, node: WorkerController, ids: Sequence[str]
) -> None:
"""Worker has finished test collection.
This adds the collection for this node to the scheduler. If
@@ -264,7 +287,9 @@ class DSession:
self.config.hook.pytest_xdist_node_collection_finished(node=node, ids=ids)
# tell session which items were effectively collected otherwise
# the controller node will finish the session with EXIT_NOTESTSCOLLECTED
assert self._session is not None
self._session.testscollected = len(ids)
assert self.sched is not None
self.sched.add_node_collection(node, ids)
if self.terminal:
self.trdist.setstatus(
@@ -280,29 +305,44 @@ class DSession:
)
self.sched.schedule()
def worker_logstart(self, node, nodeid, location):
def worker_logstart(
self,
node: WorkerController,
nodeid: str,
location: tuple[str, int | None, str],
) -> None:
"""Emitted when a node calls the pytest_runtest_logstart hook."""
self.config.hook.pytest_runtest_logstart(nodeid=nodeid, location=location)
def worker_logfinish(self, node, nodeid, location):
def worker_logfinish(
self,
node: WorkerController,
nodeid: str,
location: tuple[str, int | None, str],
) -> None:
"""Emitted when a node calls the pytest_runtest_logfinish hook."""
self.config.hook.pytest_runtest_logfinish(nodeid=nodeid, location=location)
def worker_testreport(self, node, rep):
def worker_testreport(self, node: WorkerController, rep: pytest.TestReport) -> None:
"""Emitted when a node calls the pytest_runtest_logreport hook."""
rep.node = node
rep.node = node # type: ignore[attr-defined]
self.config.hook.pytest_runtest_logreport(report=rep)
self._handlefailures(rep)
def worker_runtest_protocol_complete(self, node, item_index, duration):
def worker_runtest_protocol_complete(
self, node: WorkerController, item_index: int, duration: float
) -> None:
"""
Emitted when a node fires the 'runtest_protocol_complete' event,
signalling that a test has completed the runtestprotocol and should be
removed from the pending list in the scheduler.
"""
assert self.sched is not None
self.sched.mark_test_complete(node, item_index, duration)
def worker_unscheduled(self, node, indices):
def worker_unscheduled(
self, node: WorkerController, indices: Sequence[int]
) -> None:
"""
Emitted when a node fires the 'unscheduled' event, signalling that
some tests have been removed from the worker's queue and should be
@@ -311,9 +351,14 @@ class DSession:
This should happen only in response to 'steal' command, so schedulers
not using 'steal' command don't have to implement it.
"""
assert self.sched is not None
self.sched.remove_pending_tests_from_node(node, indices)
def worker_collectreport(self, node, rep):
def worker_collectreport(
self,
node: WorkerController,
rep: pytest.CollectReport | pytest.TestReport,
) -> None:
"""Emitted when a node calls the pytest_collectreport hook.
Because we only need the report when there's a failure/skip, as optimization
@@ -322,14 +367,20 @@ class DSession:
assert not rep.passed
self._failed_worker_collectreport(node, rep)
def worker_warning_recorded(self, warning_message, when, nodeid, location):
def worker_warning_recorded(
self,
warning_message: warnings.WarningMessage,
when: str,
nodeid: str,
location: tuple[str, int, str] | None,
) -> None:
"""Emitted when a node calls the pytest_warning_recorded hook."""
kwargs = dict(
warning_message=warning_message, when=when, nodeid=nodeid, location=location
)
self.config.hook.pytest_warning_recorded.call_historic(kwargs=kwargs)
def _clone_node(self, node):
def _clone_node(self, node: WorkerController) -> WorkerController:
"""Return new node based on an existing one.
This is normally for when a node dies, this will copy the spec
@@ -339,12 +390,17 @@ class DSession:
"""
spec = node.gateway.spec
spec.id = None
assert self.nodemanager is not None
self.nodemanager.group.allocate_id(spec)
node = self.nodemanager.setup_node(spec, self.queue.put)
self._active_nodes.add(node)
return node
clone = self.nodemanager.setup_node(spec, self.queue.put)
self._active_nodes.add(clone)
return clone
def _failed_worker_collectreport(self, node, rep):
def _failed_worker_collectreport(
self,
node: WorkerController,
rep: pytest.CollectReport | pytest.TestReport,
) -> None:
# Check we haven't already seen this report (from
# another worker).
if rep.longrepr not in self._failed_collection_errors:
@@ -352,7 +408,10 @@ class DSession:
self.config.hook.pytest_collectreport(report=rep)
self._handlefailures(rep)
def _handlefailures(self, rep):
def _handlefailures(
self,
rep: pytest.CollectReport | pytest.TestReport,
) -> None:
if rep.failed:
self.countfailures += 1
if (
@@ -362,22 +421,28 @@ class DSession:
):
self.shouldstop = f"stopping after {self.countfailures} failures"
def triggershutdown(self):
def triggershutdown(self) -> None:
if not self.shuttingdown:
self.log("triggering shutdown")
self.shuttingdown = True
assert self.sched is not None
for node in self.sched.nodes:
node.shutdown()
def handle_crashitem(self, nodeid, worker):
def handle_crashitem(self, nodeid: str, worker: WorkerController) -> None:
# XXX get more reporting info by recording pytest_runtest_logstart?
# XXX count no of failures and retry N times
fspath = nodeid.split("::")[0]
msg = f"worker {worker.gateway.id!r} crashed while running {nodeid!r}"
rep = pytest.TestReport(
nodeid, (fspath, None, fspath), (), "failed", msg, "???"
nodeid=nodeid,
location=(fspath, None, fspath),
keywords={},
outcome="failed",
longrepr=msg,
when="???", # type: ignore[arg-type]
)
rep.node = worker
rep.node = worker # type: ignore[attr-defined]
self.config.hook.pytest_handlecrashitem(
crashitem=nodeid,
@@ -404,10 +469,10 @@ class WorkerStatus(Enum):
class TerminalDistReporter:
def __init__(self, config) -> None:
def __init__(self, config: pytest.Config) -> None:
self.config = config
self.tr = config.pluginmanager.getplugin("terminalreporter")
self._status: dict[str, tuple[WorkerStatus, int]] = {}
self._status: dict[object, tuple[WorkerStatus, int]] = {}
self._lastlen = 0
self._isatty = getattr(self.tr, "isatty", self.tr.hasmarkup)
@@ -419,7 +484,12 @@ class TerminalDistReporter:
self.write_line(self.getstatus())
def setstatus(
self, spec, status: WorkerStatus, *, tests_collected: int, show: bool = True
self,
spec: execnet.XSpec,
status: WorkerStatus,
*,
tests_collected: int,
show: bool = True,
) -> None:
self._status[spec.id] = (status, tests_collected)
if show and self._isatty:
@@ -433,7 +503,7 @@ class TerminalDistReporter:
return "bringing up nodes..."
def rewrite(self, line, newline=False):
def rewrite(self, line: str, newline: bool = False) -> None:
pline = line + " " * max(self._lastlen - len(line), 0)
if newline:
self._lastlen = 0
@@ -443,7 +513,7 @@ class TerminalDistReporter:
self.tr.rewrite(pline, bold=True)
@pytest.hookimpl
def pytest_xdist_setupnodes(self, specs) -> None:
def pytest_xdist_setupnodes(self, specs: Sequence[execnet.XSpec]) -> None:
self._specs = specs
for spec in specs:
self.setstatus(spec, WorkerStatus.Created, tests_collected=0, show=False)
@@ -451,7 +521,7 @@ class TerminalDistReporter:
self.ensure_show_status()
@pytest.hookimpl
def pytest_xdist_newgateway(self, gateway) -> None:
def pytest_xdist_newgateway(self, gateway: execnet.Gateway) -> None:
if self.config.option.verbose > 0:
rinfo = gateway._rinfo()
different_interpreter = rinfo.executable != sys.executable
@@ -464,7 +534,7 @@ class TerminalDistReporter:
self.setstatus(gateway.spec, WorkerStatus.Initialized, tests_collected=0)
@pytest.hookimpl
def pytest_testnodeready(self, node) -> None:
def pytest_testnodeready(self, node: WorkerController) -> None:
if self.config.option.verbose > 0:
d = node.workerinfo
different_interpreter = d.get("executable") != sys.executable
@@ -476,23 +546,25 @@ class TerminalDistReporter:
)
@pytest.hookimpl
def pytest_testnodedown(self, node, error) -> None:
def pytest_testnodedown(self, node: WorkerController, error: object) -> None:
if not error:
return
self.write_line(f"[{node.gateway.id}] node down: {error}")
def get_default_max_worker_restart(config):
def get_default_max_worker_restart(config: pytest.Config) -> int | None:
"""Gets the default value of --max-worker-restart option if it is not provided.
Use a reasonable default to avoid workers from restarting endlessly due to crashing collections (#226).
"""
result = config.option.maxworkerrestart
if result is not None:
result = int(result)
result_str: str | None = config.option.maxworkerrestart
if result_str is not None:
result = int(result_str)
elif config.option.numprocesses:
# if --max-worker-restart was not provided, use a reasonable default (#226)
result = config.option.numprocesses * 4
else:
result = None
return result

63
src/xdist/looponfail.py
View File

@@ -13,6 +13,7 @@ import os
from pathlib import Path
import sys
import time
from typing import Any
from typing import Sequence
from _pytest._io import TerminalWriter
@@ -23,7 +24,7 @@ from xdist._path import visit_path
@pytest.hookimpl
def pytest_addoption(parser):
def pytest_addoption(parser: pytest.Parser) -> None:
group = parser.getgroup("xdist", "distributed and subprocess testing")
group._addoption(
"-f",
@@ -37,13 +38,14 @@ def pytest_addoption(parser):
@pytest.hookimpl
def pytest_cmdline_main(config):
def pytest_cmdline_main(config: pytest.Config) -> int | None:
if config.getoption("looponfail"):
usepdb = config.getoption("usepdb", False) # a core option
if usepdb:
raise pytest.UsageError("--pdb is incompatible with --looponfail.")
looponfail_main(config)
return 2 # looponfail only can get stop with ctrl-C anyway
return None
def looponfail_main(config: pytest.Config) -> None:
@@ -68,19 +70,21 @@ def looponfail_main(config: pytest.Config) -> None:
class RemoteControl:
def __init__(self, config):
self.config = config
self.failures = []
gateway: execnet.Gateway
def trace(self, *args):
def __init__(self, config: pytest.Config) -> None:
self.config = config
self.failures: list[str] = []
def trace(self, *args: object) -> None:
if self.config.option.debug:
msg = " ".join(str(x) for x in args)
print("RemoteControl:", msg)
def initgateway(self):
def initgateway(self) -> execnet.Gateway:
return execnet.makegateway("popen")
def setup(self):
def setup(self) -> None:
if hasattr(self, "gateway"):
raise ValueError("already have gateway %r" % self.gateway)
self.trace("setting up worker session")
@@ -90,17 +94,17 @@ class RemoteControl:
args=self.config.args,
option_dict=vars(self.config.option),
)
remote_outchannel = channel.receive()
remote_outchannel: execnet.Channel = channel.receive()
out = TerminalWriter()
def write(s):
def write(s: str) -> None:
out._file.write(s)
out._file.flush()
remote_outchannel.setcallback(write)
def ensure_teardown(self):
def ensure_teardown(self) -> None:
if hasattr(self, "channel"):
if not self.channel.isclosed():
self.trace("closing", self.channel)
@@ -111,12 +115,12 @@ class RemoteControl:
self.gateway.exit()
del self.gateway
def runsession(self):
def runsession(self) -> tuple[list[str], list[str], bool]:
try:
self.trace("sending", self.failures)
self.channel.send(self.failures)
try:
return self.channel.receive()
return self.channel.receive() # type: ignore[no-any-return]
except self.channel.RemoteError:
e = sys.exc_info()[1]
self.trace("ERROR", e)
@@ -124,7 +128,7 @@ class RemoteControl:
finally:
self.ensure_teardown()
def loop_once(self):
def loop_once(self) -> None:
self.setup()
self.wasfailing = self.failures and len(self.failures)
result = self.runsession()
@@ -139,7 +143,9 @@ class RemoteControl:
self.failures = uniq_failures
def repr_pytest_looponfailinfo(failreports, rootdirs):
def repr_pytest_looponfailinfo(
failreports: Sequence[str], rootdirs: Sequence[Path]
) -> None:
tr = TerminalWriter()
if failreports:
tr.sep("#", "LOOPONFAILING", bold=True)
@@ -151,12 +157,16 @@ def repr_pytest_looponfailinfo(failreports, rootdirs):
tr.line(f"### Watching: {rootdir}", bold=True)
def init_worker_session(channel, args, option_dict):
def init_worker_session(
channel: "execnet.Channel", # noqa: UP037
args: list[str],
option_dict: dict[str, "Any"], # noqa: UP037
) -> None:
import os
import sys
outchannel = channel.gateway.newchannel()
sys.stdout = sys.stderr = outchannel.makefile("w")
sys.stdout = sys.stderr = outchannel.makefile("w") # type: ignore[assignment]
channel.send(outchannel)
# prune sys.path to not contain relative paths
newpaths = []
@@ -179,21 +189,21 @@ def init_worker_session(channel, args, option_dict):
class WorkerFailSession:
def __init__(self, config, channel):
def __init__(self, config: pytest.Config, channel: execnet.Channel) -> None:
self.config = config
self.channel = channel
self.recorded_failures = []
self.recorded_failures: list[pytest.CollectReport | pytest.TestReport] = []
self.collection_failed = False
config.pluginmanager.register(self)
config.option.looponfail = False
config.option.usepdb = False
def DEBUG(self, *args):
def DEBUG(self, *args: object) -> None:
if self.config.option.debug:
print(" ".join(map(str, args)))
@pytest.hookimpl
def pytest_collection(self, session):
def pytest_collection(self, session: pytest.Session) -> bool:
self.session = session
self.trails = self.current_command
hook = self.session.ihook
@@ -208,17 +218,17 @@ class WorkerFailSession:
return True
@pytest.hookimpl
def pytest_runtest_logreport(self, report):
def pytest_runtest_logreport(self, report: pytest.TestReport) -> None:
if report.failed:
self.recorded_failures.append(report)
@pytest.hookimpl
def pytest_collectreport(self, report):
def pytest_collectreport(self, report: pytest.CollectReport) -> None:
if report.failed:
self.recorded_failures.append(report)
self.collection_failed = True
def main(self):
def main(self) -> None:
self.DEBUG("WORKER: received configuration, waiting for command trails")
try:
command = self.channel.receive()
@@ -233,7 +243,8 @@ class WorkerFailSession:
loc = rep.longrepr
loc = str(getattr(loc, "reprcrash", loc))
failreports.append(loc)
self.channel.send((trails, failreports, self.collection_failed))
result = (trails, failreports, self.collection_failed)
self.channel.send(result)
class StatRecorder:
@@ -248,7 +259,7 @@ class StatRecorder:
def rec(self, p: Path) -> bool:
return not p.name.startswith(".") and p.exists()
def waitonchange(self, checkinterval=1.0):
def waitonchange(self, checkinterval: float = 1.0) -> None:
while 1:
changed = self.check()
if changed:

53
src/xdist/newhooks.py
View File

@@ -12,16 +12,32 @@ must be taken in plugins in case ``xdist`` is not installed. Please see:
http://pytest.org/en/latest/writing_plugins.html#optionally-using-hooks-from-3rd-party-plugins
"""
from __future__ import annotations
import os
from typing import Any
from typing import Sequence
from typing import TYPE_CHECKING
import execnet
import pytest
if TYPE_CHECKING:
from xdist.remote import Producer
from xdist.scheduler.protocol import Scheduling
from xdist.workermanage import WorkerController
@pytest.hookspec()
def pytest_xdist_setupnodes(config, specs):
def pytest_xdist_setupnodes(
config: pytest.Config, specs: Sequence[execnet.XSpec]
) -> None:
"""Called before any remote node is set up."""
@pytest.hookspec()
def pytest_xdist_newgateway(gateway):
def pytest_xdist_newgateway(gateway: execnet.Gateway) -> None:
"""Called on new raw gateway creation."""
@@ -30,7 +46,10 @@ def pytest_xdist_newgateway(gateway):
"rsync feature is deprecated and will be removed in pytest-xdist 4.0"
)
)
def pytest_xdist_rsyncstart(source, gateways):
def pytest_xdist_rsyncstart(
source: str | os.PathLike[str],
gateways: Sequence[execnet.Gateway],
) -> None:
"""Called before rsyncing a directory to remote gateways takes place."""
@@ -39,52 +58,62 @@ def pytest_xdist_rsyncstart(source, gateways):
"rsync feature is deprecated and will be removed in pytest-xdist 4.0"
)
)
def pytest_xdist_rsyncfinish(source, gateways):
def pytest_xdist_rsyncfinish(
source: str | os.PathLike[str],
gateways: Sequence[execnet.Gateway],
) -> None:
"""Called after rsyncing a directory to remote gateways takes place."""
@pytest.hookspec(firstresult=True)
def pytest_xdist_getremotemodule():
def pytest_xdist_getremotemodule() -> Any:
"""Called when creating remote node."""
@pytest.hookspec()
def pytest_configure_node(node):
def pytest_configure_node(node: WorkerController) -> None:
"""Configure node information before it gets instantiated."""
@pytest.hookspec()
def pytest_testnodeready(node):
def pytest_testnodeready(node: WorkerController) -> None:
"""Test Node is ready to operate."""
@pytest.hookspec()
def pytest_testnodedown(node, error):
def pytest_testnodedown(node: WorkerController, error: object | None) -> None:
"""Test Node is down."""
@pytest.hookspec()
def pytest_xdist_node_collection_finished(node, ids):
def pytest_xdist_node_collection_finished(
node: WorkerController, ids: Sequence[str]
) -> None:
"""Called by the controller node when a worker node finishes collecting."""
@pytest.hookspec(firstresult=True)
def pytest_xdist_make_scheduler(config, log):
def pytest_xdist_make_scheduler(
config: pytest.Config, log: Producer
) -> Scheduling | None:
"""Return a node scheduler implementation."""
@pytest.hookspec(firstresult=True)
def pytest_xdist_auto_num_workers(config):
def pytest_xdist_auto_num_workers(config: pytest.Config) -> int:
"""
Return the number of workers to spawn when ``--numprocesses=auto`` is given in the
command-line.
.. versionadded:: 2.1
"""
raise NotImplementedError()
@pytest.hookspec(firstresult=True)
def pytest_handlecrashitem(crashitem, report, sched):
def pytest_handlecrashitem(
crashitem: str, report: pytest.TestReport, sched: Scheduling
) -> None:
"""
Handle a crashitem, modifying the report if necessary.

54
src/xdist/plugin.py
View File

@@ -1,5 +1,8 @@
from __future__ import annotations
import os
import sys
from typing import Literal
import uuid
import warnings
@@ -10,7 +13,7 @@ _sys_path = list(sys.path) # freeze a copy of sys.path at interpreter startup
@pytest.hookimpl
def pytest_xdist_auto_num_workers(config):
def pytest_xdist_auto_num_workers(config: pytest.Config) -> int:
env_var = os.environ.get("PYTEST_XDIST_AUTO_NUM_WORKERS")
if env_var:
try:
@@ -25,14 +28,14 @@ def pytest_xdist_auto_num_workers(config):
except ImportError:
pass
else:
use_logical = config.option.numprocesses == "logical"
use_logical: bool = config.option.numprocesses == "logical"
count = psutil.cpu_count(logical=use_logical) or psutil.cpu_count()
if count:
return count
try:
from os import sched_getaffinity
def cpu_count():
def cpu_count() -> int:
return len(sched_getaffinity(0))
except ImportError:
@@ -40,7 +43,7 @@ def pytest_xdist_auto_num_workers(config):
# workaround https://bitbucket.org/pypy/pypy/issues/2375
return 2
try:
from os import cpu_count
from os import cpu_count # type: ignore[assignment]
except ImportError:
from multiprocessing import cpu_count
try:
@@ -50,15 +53,15 @@ def pytest_xdist_auto_num_workers(config):
return n if n else 1
def parse_numprocesses(s):
def parse_numprocesses(s: str) -> int | Literal["auto", "logical"]:
if s in ("auto", "logical"):
return s
return s # type: ignore[return-value]
elif s is not None:
return int(s)
@pytest.hookimpl
def pytest_addoption(parser):
def pytest_addoption(parser: pytest.Parser) -> None:
# 'Help' formatting (same rules as pytest's):
# Start with capitalized letters.
# If a single phrase, do not end with period. If more than one phrase, all phrases end with periods.
@@ -206,7 +209,7 @@ def pytest_addoption(parser):
@pytest.hookimpl
def pytest_addhooks(pluginmanager):
def pytest_addhooks(pluginmanager: pytest.PytestPluginManager) -> None:
from xdist import newhooks
pluginmanager.add_hookspecs(newhooks)
@@ -218,7 +221,7 @@ def pytest_addhooks(pluginmanager):
@pytest.hookimpl(trylast=True)
def pytest_configure(config):
def pytest_configure(config: pytest.Config) -> None:
config_line = (
"xdist_group: specify group for tests should run in same session."
"in relation to one another. Provided by pytest-xdist."
@@ -256,16 +259,13 @@ def pytest_configure(config):
config.issue_config_time_warning(warning, 2)
def _is_distribution_mode(config):
"""Return `True` if distribution mode is on, `False` otherwise.
:param config: the `pytest` `config` object
"""
return config.getoption("dist") != "no" and config.getoption("tx")
def _is_distribution_mode(config: pytest.Config) -> bool:
"""Whether distribution mode is on."""
return config.getoption("dist") != "no" and bool(config.getoption("tx"))
@pytest.hookimpl(tryfirst=True)
def pytest_cmdline_main(config):
def pytest_cmdline_main(config: pytest.Config) -> None:
if config.option.distload:
config.option.dist = "load"
@@ -302,7 +302,9 @@ def pytest_cmdline_main(config):
# -------------------------------------------------------------------------
def is_xdist_worker(request_or_session) -> bool:
def is_xdist_worker(
request_or_session: pytest.FixtureRequest | pytest.Session,
) -> bool:
"""Return `True` if this is an xdist worker, `False` otherwise.
:param request_or_session: the `pytest` `request` or `session` object
@@ -310,7 +312,9 @@ def is_xdist_worker(request_or_session) -> bool:
return hasattr(request_or_session.config, "workerinput")
def is_xdist_controller(request_or_session) -> bool:
def is_xdist_controller(
request_or_session: pytest.FixtureRequest | pytest.Session,
) -> bool:
"""Return `True` if this is the xdist controller, `False` otherwise.
Note: this method also returns `False` when distribution has not been
@@ -328,7 +332,9 @@ def is_xdist_controller(request_or_session) -> bool:
is_xdist_master = is_xdist_controller
def get_xdist_worker_id(request_or_session):
def get_xdist_worker_id(
request_or_session: pytest.FixtureRequest | pytest.Session,
) -> str:
"""Return the id of the current worker ('gw0', 'gw1', etc) or 'master'
if running on the controller node.
@@ -338,14 +344,15 @@ def get_xdist_worker_id(request_or_session):
:param request_or_session: the `pytest` `request` or `session` object
"""
if hasattr(request_or_session.config, "workerinput"):
return request_or_session.config.workerinput["workerid"]
workerid: str = request_or_session.config.workerinput["workerid"]
return workerid
else:
# TODO: remove "master", ideally for a None
return "master"
@pytest.fixture(scope="session")
def worker_id(request):
def worker_id(request: pytest.FixtureRequest) -> str:
"""Return the id of the current worker ('gw0', 'gw1', etc) or 'master'
if running on the master node.
"""
@@ -354,9 +361,10 @@ def worker_id(request):
@pytest.fixture(scope="session")
def testrun_uid(request):
def testrun_uid(request: pytest.FixtureRequest) -> str:
"""Return the unique id of the current test."""
if hasattr(request.config, "workerinput"):
return request.config.workerinput["testrunuid"]
testrunid: str = request.config.workerinput["testrunuid"]
return testrunid
else:
return uuid.uuid4().hex

138
src/xdist/remote.py
View File

@@ -6,16 +6,22 @@ on the rest of the xdist code. This means that the xdist-plugin
needs not to be installed in remote environments.
"""
from __future__ import annotations
import contextlib
import enum
import os
import sys
import time
from typing import Any
from typing import Generator
from typing import Literal
from typing import Sequence
from typing import TypedDict
import warnings
from _pytest.config import _prepareconfig
from execnet.gateway_base import DumpError
from execnet.gateway_base import dumps
import execnet
import pytest
@@ -23,7 +29,7 @@ try:
from setproctitle import setproctitle
except ImportError:
def setproctitle(title):
def setproctitle(title: str) -> None:
pass
@@ -35,7 +41,7 @@ class Producer:
to have the other way around.
"""
def __init__(self, name: str, *, enabled: bool = True):
def __init__(self, name: str, *, enabled: bool = True) -> None:
self.name = name
self.enabled = enabled
@@ -46,11 +52,11 @@ class Producer:
if self.enabled:
print(f"[{self.name}]", *a, **k, file=sys.stderr)
def __getattr__(self, name: str) -> "Producer":
def __getattr__(self, name: str) -> Producer:
return type(self)(name, enabled=self.enabled)
def worker_title(title):
def worker_title(title: str) -> None:
try:
setproctitle(title)
except Exception:
@@ -64,59 +70,63 @@ class Marker(enum.Enum):
class WorkerInteractor:
def __init__(self, config, channel):
def __init__(self, config: pytest.Config, channel: execnet.Channel) -> None:
self.config = config
self.workerid = config.workerinput.get("workerid", "?")
self.testrunuid = config.workerinput["testrunuid"]
workerinput: dict[str, Any] = config.workerinput # type: ignore[attr-defined]
self.workerid = workerinput.get("workerid", "?")
self.testrunuid = workerinput["testrunuid"]
self.log = Producer(f"worker-{self.workerid}", enabled=config.option.debug)
self.channel = channel
self.torun = self._make_queue()
self.nextitem_index = None
self.nextitem_index: int | None | Literal[Marker.SHUTDOWN] = None
config.pluginmanager.register(self)
def _make_queue(self):
def _make_queue(self) -> Any:
return self.channel.gateway.execmodel.queue.Queue()
def _get_next_item_index(self):
def _get_next_item_index(self) -> int | Literal[Marker.SHUTDOWN]:
"""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 Marker.QUEUE_REPLACED:
result = self.torun.get()
return result
return result # type: ignore[no-any-return]
def sendevent(self, name, **kwargs):
def sendevent(self, name: str, **kwargs: object) -> None:
self.log("sending", name, kwargs)
self.channel.send((name, kwargs))
@pytest.hookimpl
def pytest_internalerror(self, excrepr):
def pytest_internalerror(self, excrepr: object) -> None:
formatted_error = str(excrepr)
for line in formatted_error.split("\n"):
self.log("IERROR>", line)
interactor.sendevent("internal_error", formatted_error=formatted_error)
@pytest.hookimpl
def pytest_sessionstart(self, session):
def pytest_sessionstart(self, session: pytest.Session) -> None:
self.session = session
workerinfo = getinfodict()
self.sendevent("workerready", workerinfo=workerinfo)
@pytest.hookimpl(hookwrapper=True)
def pytest_sessionfinish(self, exitstatus):
def pytest_sessionfinish(self, exitstatus: int) -> Generator[None, object, None]:
workeroutput: dict[str, Any] = self.config.workeroutput # type: ignore[attr-defined]
# in pytest 5.0+, exitstatus is an IntEnum object
self.config.workeroutput["exitstatus"] = int(exitstatus)
self.config.workeroutput["shouldfail"] = self.session.shouldfail
self.config.workeroutput["shouldstop"] = self.session.shouldstop
workeroutput["exitstatus"] = int(exitstatus)
workeroutput["shouldfail"] = self.session.shouldfail
workeroutput["shouldstop"] = self.session.shouldstop
yield
self.sendevent("workerfinished", workeroutput=self.config.workeroutput)
self.sendevent("workerfinished", workeroutput=workeroutput)
@pytest.hookimpl
def pytest_collection(self, session):
def pytest_collection(self) -> None:
self.sendevent("collectionstart")
def handle_command(self, command):
def handle_command(
self, command: tuple[str, dict[str, Any]] | Literal[Marker.SHUTDOWN]
) -> None:
if command is Marker.SHUTDOWN:
self.torun.put(Marker.SHUTDOWN)
return
@@ -135,18 +145,19 @@ class WorkerInteractor:
elif name == "steal":
self.steal(kwargs["indices"])
def steal(self, indices):
indices = set(indices)
def steal(self, indices: Sequence[int]) -> None:
indices_set = set(indices)
stolen = []
old_queue, self.torun = self.torun, self._make_queue()
def old_queue_get_nowait_noraise():
def old_queue_get_nowait_noraise() -> int | None:
with contextlib.suppress(self.channel.gateway.execmodel.queue.Empty):
return old_queue.get_nowait()
return old_queue.get_nowait() # type: ignore[no-any-return]
return None
for i in iter(old_queue_get_nowait_noraise, None):
if i in indices:
if i in indices_set:
stolen.append(i)
else:
self.torun.put(i)
@@ -155,7 +166,7 @@ class WorkerInteractor:
old_queue.put(Marker.QUEUE_REPLACED)
@pytest.hookimpl
def pytest_runtestloop(self, session):
def pytest_runtestloop(self, session: pytest.Session) -> bool:
self.log("entering main loop")
self.channel.setcallback(self.handle_command, endmarker=Marker.SHUTDOWN)
self.nextitem_index = self._get_next_item_index()
@@ -165,7 +176,8 @@ class WorkerInteractor:
break
return True
def run_one_test(self):
def run_one_test(self) -> None:
assert isinstance(self.nextitem_index, int)
self.item_index = self.nextitem_index
self.nextitem_index = self._get_next_item_index()
@@ -174,6 +186,7 @@ class WorkerInteractor:
if self.nextitem_index is Marker.SHUTDOWN:
nextitem = None
else:
assert self.nextitem_index is not None
nextitem = items[self.nextitem_index]
worker_title("[pytest-xdist running] %s" % item.nodeid)
@@ -188,7 +201,11 @@ class WorkerInteractor:
"runtest_protocol_complete", item_index=self.item_index, duration=duration
)
def pytest_collection_modifyitems(self, session, config, items):
def pytest_collection_modifyitems(
self,
config: pytest.Config,
items: list[pytest.Item],
) -> None:
# add the group name to nodeid as suffix if --dist=loadgroup
if config.getvalue("loadgroup"):
for item in items:
@@ -203,7 +220,7 @@ class WorkerInteractor:
item._nodeid = f"{item.nodeid}@{gname}"
@pytest.hookimpl
def pytest_collection_finish(self, session):
def pytest_collection_finish(self, session: pytest.Session) -> None:
self.sendevent(
"collectionfinish",
topdir=str(self.config.rootpath),
@@ -211,15 +228,23 @@ class WorkerInteractor:
)
@pytest.hookimpl
def pytest_runtest_logstart(self, nodeid, location):
def pytest_runtest_logstart(
self,
nodeid: str,
location: tuple[str, int | None, str],
) -> None:
self.sendevent("logstart", nodeid=nodeid, location=location)
@pytest.hookimpl
def pytest_runtest_logfinish(self, nodeid, location):
def pytest_runtest_logfinish(
self,
nodeid: str,
location: tuple[str, int | None, str],
) -> None:
self.sendevent("logfinish", nodeid=nodeid, location=location)
@pytest.hookimpl
def pytest_runtest_logreport(self, report):
def pytest_runtest_logreport(self, report: pytest.TestReport) -> None:
data = self.config.hook.pytest_report_to_serializable(
config=self.config, report=report
)
@@ -230,7 +255,7 @@ class WorkerInteractor:
self.sendevent("testreport", data=data)
@pytest.hookimpl
def pytest_collectreport(self, report):
def pytest_collectreport(self, report: pytest.CollectReport) -> None:
# send only reports that have not passed to controller as optimization (#330)
if not report.passed:
data = self.config.hook.pytest_report_to_serializable(
@@ -239,7 +264,13 @@ class WorkerInteractor:
self.sendevent("collectreport", data=data)
@pytest.hookimpl
def pytest_warning_recorded(self, warning_message, when, nodeid, location):
def pytest_warning_recorded(
self,
warning_message: warnings.WarningMessage,
when: str,
nodeid: str,
location: tuple[str, int, str] | None,
) -> None:
self.sendevent(
"warning_recorded",
warning_message_data=serialize_warning_message(warning_message),
@@ -249,7 +280,9 @@ class WorkerInteractor:
)
def serialize_warning_message(warning_message):
def serialize_warning_message(
warning_message: warnings.WarningMessage,
) -> dict[str, Any]:
if isinstance(warning_message.message, Warning):
message_module = type(warning_message.message).__module__
message_class_name = type(warning_message.message).__name__
@@ -257,8 +290,8 @@ def serialize_warning_message(warning_message):
# check now if we can serialize the warning arguments (#349)
# if not, we will just use the exception message on the controller node
try:
dumps(warning_message.message.args)
except DumpError:
execnet.dumps(warning_message.message.args)
except execnet.DumpError:
message_args = None
else:
message_args = warning_message.message.args
@@ -283,27 +316,38 @@ def serialize_warning_message(warning_message):
"category_class_name": category_class_name,
}
# access private _WARNING_DETAILS because the attributes vary between Python versions
for attr_name in warning_message._WARNING_DETAILS:
for attr_name in warning_message._WARNING_DETAILS: # type: ignore[attr-defined]
if attr_name in ("message", "category"):
continue
attr = getattr(warning_message, attr_name)
# Check if we can serialize the warning detail, marking `None` otherwise
# Note that we need to define the attr (even as `None`) to allow deserializing
try:
dumps(attr)
except DumpError:
execnet.dumps(attr)
except execnet.DumpError:
result[attr_name] = repr(attr)
else:
result[attr_name] = attr
return result
def getinfodict():
class WorkerInfo(TypedDict):
version: str
version_info: tuple[int, int, int, str, int]
sysplatform: str
platform: str
executable: str
cwd: str
id: str
spec: execnet.XSpec
def getinfodict() -> WorkerInfo:
import platform
return dict(
version=sys.version,
version_info=tuple(sys.version_info),
version_info=tuple(sys.version_info), # type: ignore[typeddict-item]
sysplatform=sys.platform,
platform=platform.platform(),
executable=sys.executable,
@@ -311,7 +355,7 @@ def getinfodict():
)
def setup_config(config, basetemp):
def setup_config(config: pytest.Config, basetemp: str | None) -> None:
config.option.loadgroup = config.getvalue("dist") == "loadgroup"
config.option.looponfail = False
config.option.usepdb = False
@@ -323,7 +367,7 @@ def setup_config(config, basetemp):
if __name__ == "__channelexec__":
channel = channel # type: ignore[name-defined] # noqa: F821, PLW0127
channel: execnet.Channel = channel # type: ignore[name-defined] # noqa: F821, PLW0127
workerinput, args, option_dict, change_sys_path = channel.receive() # type: ignore[name-defined]
if change_sys_path is None:

10
src/xdist/report.py
View File

@@ -1,7 +1,15 @@
from __future__ import annotations
from difflib import unified_diff
from typing import Sequence
def report_collection_diff(from_collection, to_collection, from_id, to_id):
def report_collection_diff(
from_collection: Sequence[str],
to_collection: Sequence[str],
from_id: str,
to_id: str,
) -> str | None:
"""Report the collected test difference between two nodes.
:returns: detailed message describing the difference between the given

47
src/xdist/scheduler/each.py
View File

@@ -1,6 +1,13 @@
from __future__ import annotations
from typing import Sequence
import pytest
from xdist.remote import Producer
from xdist.report import report_collection_diff
from xdist.workermanage import parse_spec_config
from xdist.workermanage import WorkerController
class EachScheduling:
@@ -17,13 +24,13 @@ class EachScheduling:
assigned the remaining items from the removed node.
"""
def __init__(self, config, log=None):
def __init__(self, config: pytest.Config, log: Producer | None = None) -> None:
self.config = config
self.numnodes = len(parse_spec_config(config))
self.node2collection = {}
self.node2pending = {}
self._started = []
self._removed2pending = {}
self.node2collection: dict[WorkerController, list[str]] = {}
self.node2pending: dict[WorkerController, list[int]] = {}
self._started: list[WorkerController] = []
self._removed2pending: dict[WorkerController, list[int]] = {}
if log is None:
self.log = Producer("eachsched")
else:
@@ -31,12 +38,12 @@ class EachScheduling:
self.collection_is_completed = False
@property
def nodes(self):
def nodes(self) -> list[WorkerController]:
"""A list of all nodes in the scheduler."""
return list(self.node2pending.keys())
@property
def tests_finished(self):
def tests_finished(self) -> bool:
if not self.collection_is_completed:
return False
if self._removed2pending:
@@ -47,7 +54,7 @@ class EachScheduling:
return True
@property
def has_pending(self):
def has_pending(self) -> bool:
"""Return True if there are pending test items.
This indicates that collection has finished and nodes are
@@ -59,11 +66,13 @@ class EachScheduling:
return True
return False
def add_node(self, node):
def add_node(self, node: WorkerController) -> None:
assert node not in self.node2pending
self.node2pending[node] = []
def add_node_collection(self, node, collection):
def add_node_collection(
self, node: WorkerController, collection: Sequence[str]
) -> None:
"""Add the collected test items from a node.
Collection is complete once all nodes have submitted their
@@ -97,26 +106,32 @@ class EachScheduling:
self.node2pending[node] = pending
break
def mark_test_complete(self, node, item_index, duration=0):
def mark_test_complete(
self, node: WorkerController, item_index: int, duration: float = 0
) -> None:
self.node2pending[node].remove(item_index)
def mark_test_pending(self, item):
def mark_test_pending(self, item: str) -> None:
raise NotImplementedError()
def remove_pending_tests_from_node(self, node, indices):
def remove_pending_tests_from_node(
self,
node: WorkerController,
indices: Sequence[int],
) -> None:
raise NotImplementedError()
def remove_node(self, node):
def remove_node(self, node: WorkerController) -> str | None:
# KeyError if we didn't get an add_node() yet
pending = self.node2pending.pop(node)
if not pending:
return
return None
crashitem = self.node2collection[node][pending.pop(0)]
if pending:
self._removed2pending[node] = pending
return crashitem
def schedule(self):
def schedule(self) -> None:
"""Schedule the test items on the nodes.
If the node's pending list is empty it is a new node which

54
src/xdist/scheduler/load.py
View File

@@ -1,10 +1,14 @@
from __future__ import annotations
from itertools import cycle
from typing import Sequence
import pytest
from xdist.remote import Producer
from xdist.report import report_collection_diff
from xdist.workermanage import parse_spec_config
from xdist.workermanage import WorkerController
class LoadScheduling:
@@ -53,12 +57,12 @@ class LoadScheduling:
:config: Config object, used for handling hooks.
"""
def __init__(self, config, log=None):
def __init__(self, config: pytest.Config, log: Producer | None = None) -> None:
self.numnodes = len(parse_spec_config(config))
self.node2collection = {}
self.node2pending = {}
self.pending = []
self.collection = None
self.node2collection: dict[WorkerController, list[str]] = {}
self.node2pending: dict[WorkerController, list[int]] = {}
self.pending: list[int] = []
self.collection: list[str] | None = None
if log is None:
self.log = Producer("loadsched")
else:
@@ -67,12 +71,12 @@ class LoadScheduling:
self.maxschedchunk = self.config.getoption("maxschedchunk")
@property
def nodes(self):
def nodes(self) -> list[WorkerController]:
"""A list of all nodes in the scheduler."""
return list(self.node2pending.keys())
@property
def collection_is_completed(self):
def collection_is_completed(self) -> bool:
"""Boolean indication initial test collection is complete.
This is a boolean indicating all initial participating nodes
@@ -82,7 +86,7 @@ class LoadScheduling:
return len(self.node2collection) >= self.numnodes
@property
def tests_finished(self):
def tests_finished(self) -> bool:
"""Return True if all tests have been executed by the nodes."""
if not self.collection_is_completed:
return False
@@ -94,7 +98,7 @@ class LoadScheduling:
return True
@property
def has_pending(self):
def has_pending(self) -> bool:
"""Return True if there are pending test items.
This indicates that collection has finished and nodes are
@@ -108,7 +112,7 @@ class LoadScheduling:
return True
return False
def add_node(self, node):
def add_node(self, node: WorkerController) -> None:
"""Add a new node to the scheduler.
From now on the node will be allocated chunks of tests to
@@ -120,7 +124,9 @@ class LoadScheduling:
assert node not in self.node2pending
self.node2pending[node] = []
def add_node_collection(self, node, collection):
def add_node_collection(
self, node: WorkerController, collection: Sequence[str]
) -> None:
"""Add the collected test items from a node.
The collection is stored in the ``.node2collection`` map.
@@ -141,7 +147,9 @@ class LoadScheduling:
return
self.node2collection[node] = list(collection)
def mark_test_complete(self, node, item_index, duration=0):
def mark_test_complete(
self, node: WorkerController, item_index: int, duration: float = 0
) -> None:
"""Mark test item as completed by node.
The duration it took to execute the item is used as a hint to
@@ -152,7 +160,8 @@ class LoadScheduling:
self.node2pending[node].remove(item_index)
self.check_schedule(node, duration=duration)
def mark_test_pending(self, item):
def mark_test_pending(self, item: str) -> None:
assert self.collection is not None
self.pending.insert(
0,
self.collection.index(item),
@@ -160,10 +169,14 @@ class LoadScheduling:
for node in self.node2pending:
self.check_schedule(node)
def remove_pending_tests_from_node(self, node, indices):
def remove_pending_tests_from_node(
self,
node: WorkerController,
indices: Sequence[int],
) -> None:
raise NotImplementedError()
def check_schedule(self, node, duration=0):
def check_schedule(self, node: WorkerController, duration: float = 0) -> None:
"""Maybe schedule new items on the node.
If there are any globally pending nodes left then this will
@@ -197,7 +210,7 @@ class LoadScheduling:
self.log("num items waiting for node:", len(self.pending))
def remove_node(self, node):
def remove_node(self, node: WorkerController) -> str | None:
"""Remove a node from the scheduler.
This should be called either when the node crashed or at
@@ -212,16 +225,17 @@ class LoadScheduling:
"""
pending = self.node2pending.pop(node)
if not pending:
return
return None
# The node crashed, reassing pending items
assert self.collection is not None
crashitem = self.collection[pending.pop(0)]
self.pending.extend(pending)
for node in self.node2pending:
self.check_schedule(node)
return crashitem
def schedule(self):
def schedule(self) -> None:
"""Initiate distribution of the test collection.
Initiate scheduling of the items across the nodes. If this
@@ -285,14 +299,14 @@ class LoadScheduling:
for node in self.nodes:
node.shutdown()
def _send_tests(self, node, num):
def _send_tests(self, node: WorkerController, num: int) -> None:
tests_per_node = self.pending[:num]
if tests_per_node:
del self.pending[:num]
self.node2pending[node].extend(tests_per_node)
node.send_runtest_some(tests_per_node)
def _check_nodes_have_same_collection(self):
def _check_nodes_have_same_collection(self) -> bool:
"""Return True if all nodes have collected the same items.
If collections differ, this method returns False while logging

8
src/xdist/scheduler/loadfile.py
View File

@@ -1,3 +1,7 @@
from __future__ import annotations
import pytest
from xdist.remote import Producer
from .loadscope import LoadScopeScheduling
@@ -21,14 +25,14 @@ class LoadFileScheduling(LoadScopeScheduling):
This class behaves very much like LoadScopeScheduling, but with a file-level scope.
"""
def __init__(self, config, log=None):
def __init__(self, config: pytest.Config, log: Producer | None = None) -> None:
super().__init__(config, log)
if log is None:
self.log = Producer("loadfilesched")
else:
self.log = log.loadfilesched
def _split_scope(self, nodeid):
def _split_scope(self, nodeid: str) -> str:
"""Determine the scope (grouping) of a nodeid.
There are usually 3 cases for a nodeid::

8
src/xdist/scheduler/loadgroup.py
View File

@@ -1,3 +1,7 @@
from __future__ import annotations
import pytest
from xdist.remote import Producer
from .loadscope import LoadScopeScheduling
@@ -10,14 +14,14 @@ class LoadGroupScheduling(LoadScopeScheduling):
instead of the module or class to which they belong to.
"""
def __init__(self, config, log=None):
def __init__(self, config: pytest.Config, log: Producer | None = None) -> None:
super().__init__(config, log)
if log is None:
self.log = Producer("loadgroupsched")
else:
self.log = log.loadgroupsched
def _split_scope(self, nodeid):
def _split_scope(self, nodeid: str) -> str:
"""Determine the scope (grouping) of a nodeid.
There are usually 3 cases for a nodeid::

57
src/xdist/scheduler/loadscope.py
View File

@@ -1,10 +1,15 @@
from __future__ import annotations
from collections import OrderedDict
from typing import NoReturn
from typing import Sequence
import pytest
from xdist.remote import Producer
from xdist.report import report_collection_diff
from xdist.workermanage import parse_spec_config
from xdist.workermanage import WorkerController
class LoadScopeScheduling:
@@ -85,13 +90,13 @@ class LoadScopeScheduling:
:config: Config object, used for handling hooks.
"""
def __init__(self, config, log=None):
def __init__(self, config: pytest.Config, log: Producer | None = None) -> None:
self.numnodes = len(parse_spec_config(config))
self.collection = None
self.collection: list[str] | None = None
self.workqueue = OrderedDict()
self.assigned_work = {}
self.registered_collections = {}
self.workqueue: OrderedDict[str, dict[str, bool]] = OrderedDict()
self.assigned_work: dict[WorkerController, dict[str, dict[str, bool]]] = {}
self.registered_collections: dict[WorkerController, list[str]] = {}
if log is None:
self.log = Producer("loadscopesched")
@@ -101,12 +106,12 @@ class LoadScopeScheduling:
self.config = config
@property
def nodes(self):
def nodes(self) -> list[WorkerController]:
"""A list of all active nodes in the scheduler."""
return list(self.assigned_work.keys())
@property
def collection_is_completed(self):
def collection_is_completed(self) -> bool:
"""Boolean indication initial test collection is complete.
This is a boolean indicating all initial participating nodes have
@@ -116,7 +121,7 @@ class LoadScopeScheduling:
return len(self.registered_collections) >= self.numnodes
@property
def tests_finished(self):
def tests_finished(self) -> bool:
"""Return True if all tests have been executed by the nodes."""
if not self.collection_is_completed:
return False
@@ -131,7 +136,7 @@ class LoadScopeScheduling:
return True
@property
def has_pending(self):
def has_pending(self) -> bool:
"""Return True if there are pending test items.
This indicates that collection has finished and nodes are still
@@ -147,7 +152,7 @@ class LoadScopeScheduling:
return False
def add_node(self, node):
def add_node(self, node: WorkerController) -> None:
"""Add a new node to the scheduler.
From now on the node will be assigned work units to be executed.
@@ -158,7 +163,7 @@ class LoadScopeScheduling:
assert node not in self.assigned_work
self.assigned_work[node] = {}
def remove_node(self, node):
def remove_node(self, node: WorkerController) -> str | None:
"""Remove a node from the scheduler.
This should be called either when the node crashed or at shutdown time.
@@ -199,7 +204,9 @@ class LoadScopeScheduling:
return crashitem
def add_node_collection(self, node, collection):
def add_node_collection(
self, node: WorkerController, collection: Sequence[str]
) -> None:
"""Add the collected test items from a node.
The collection is stored in the ``.registered_collections`` dictionary.
@@ -228,7 +235,9 @@ class LoadScopeScheduling:
self.registered_collections[node] = list(collection)
def mark_test_complete(self, node, item_index, duration=0):
def mark_test_complete(
self, node: WorkerController, item_index: int, duration: float = 0
) -> None:
"""Mark test item as completed by node.
Called by the hook:
@@ -241,13 +250,17 @@ class LoadScopeScheduling:
self.assigned_work[node][scope][nodeid] = True
self._reschedule(node)
def mark_test_pending(self, item):
def mark_test_pending(self, item: str) -> NoReturn:
raise NotImplementedError()
def remove_pending_tests_from_node(self, node, indices):
def remove_pending_tests_from_node(
self,
node: WorkerController,
indices: Sequence[int],
) -> None:
raise NotImplementedError()
def _assign_work_unit(self, node):
def _assign_work_unit(self, node: WorkerController) -> None:
"""Assign a work unit to a node."""
assert self.workqueue
@@ -268,7 +281,7 @@ class LoadScopeScheduling:
node.send_runtest_some(nodeids_indexes)
def _split_scope(self, nodeid):
def _split_scope(self, nodeid: str) -> str:
"""Determine the scope (grouping) of a nodeid.
There are usually 3 cases for a nodeid::
@@ -292,12 +305,12 @@ class LoadScopeScheduling:
"""
return nodeid.rsplit("::", 1)[0]
def _pending_of(self, workload):
def _pending_of(self, workload: dict[str, dict[str, bool]]) -> int:
"""Return the number of pending tests in a workload."""
pending = sum(list(scope.values()).count(False) for scope in workload.values())
return pending
def _reschedule(self, node):
def _reschedule(self, node: WorkerController) -> None:
"""Maybe schedule new items on the node.
If there are any globally pending work units left then this will check
@@ -322,7 +335,7 @@ class LoadScopeScheduling:
# Pop one unit of work and assign it
self._assign_work_unit(node)
def schedule(self):
def schedule(self) -> None:
"""Initiate distribution of the test collection.
Initiate scheduling of the items across the nodes. If this gets called
@@ -352,7 +365,7 @@ class LoadScopeScheduling:
return
# Determine chunks of work (scopes)
unsorted_workqueue = {}
unsorted_workqueue: dict[str, dict[str, bool]] = {}
for nodeid in self.collection:
scope = self._split_scope(nodeid)
work_unit = unsorted_workqueue.setdefault(scope, {})
@@ -389,7 +402,7 @@ class LoadScopeScheduling:
for node in self.nodes:
node.shutdown()
def _check_nodes_have_same_collection(self):
def _check_nodes_have_same_collection(self) -> bool:
"""Return True if all nodes have collected the same items.
If collections differ, this method returns False while logging

59
src/xdist/scheduler/worksteal.py
View File

@@ -1,17 +1,18 @@
from __future__ import annotations
from typing import Any
from typing import NamedTuple
from typing import Sequence
import pytest
from xdist.remote import Producer
from xdist.report import report_collection_diff
from xdist.workermanage import parse_spec_config
from xdist.workermanage import WorkerController
class NodePending(NamedTuple):
node: Any
node: WorkerController
pending: list[int]
@@ -63,26 +64,26 @@ class WorkStealingScheduling:
simultaneous requests.
"""
def __init__(self, config, log=None):
def __init__(self, config: pytest.Config, log: Producer | None = None) -> None:
self.numnodes = len(parse_spec_config(config))
self.node2collection = {}
self.node2pending = {}
self.pending = []
self.collection = None
self.node2collection: dict[WorkerController, list[str]] = {}
self.node2pending: dict[WorkerController, list[int]] = {}
self.pending: list[int] = []
self.collection: list[str] | None = None
if log is None:
self.log = Producer("workstealsched")
else:
self.log = log.workstealsched
self.config = config
self.steal_requested_from_node = None
self.steal_requested_from_node: WorkerController | None = None
@property
def nodes(self):
def nodes(self) -> list[WorkerController]:
"""A list of all nodes in the scheduler."""
return list(self.node2pending.keys())
@property
def collection_is_completed(self):
def collection_is_completed(self) -> bool:
"""Boolean indication initial test collection is complete.
This is a boolean indicating all initial participating nodes
@@ -92,7 +93,7 @@ class WorkStealingScheduling:
return len(self.node2collection) >= self.numnodes
@property
def tests_finished(self):
def tests_finished(self) -> bool:
"""Return True if all tests have been executed by the nodes."""
if not self.collection_is_completed:
return False
@@ -106,7 +107,7 @@ class WorkStealingScheduling:
return True
@property
def has_pending(self):
def has_pending(self) -> bool:
"""Return True if there are pending test items.
This indicates that collection has finished and nodes are
@@ -120,7 +121,7 @@ class WorkStealingScheduling:
return True
return False
def add_node(self, node):
def add_node(self, node: WorkerController) -> None:
"""Add a new node to the scheduler.
From now on the node will be allocated chunks of tests to
@@ -132,7 +133,9 @@ class WorkStealingScheduling:
assert node not in self.node2pending
self.node2pending[node] = []
def add_node_collection(self, node, collection):
def add_node_collection(
self, node: WorkerController, collection: Sequence[str]
) -> None:
"""Add the collected test items from a node.
The collection is stored in the ``.node2collection`` map.
@@ -153,7 +156,9 @@ class WorkStealingScheduling:
return
self.node2collection[node] = list(collection)
def mark_test_complete(self, node, item_index, duration=None):
def mark_test_complete(
self, node: WorkerController, item_index: int, duration: float | None = None
) -> None:
"""Mark test item as completed by node.
This is called by the ``DSession.worker_testreport`` hook.
@@ -161,14 +166,19 @@ class WorkStealingScheduling:
self.node2pending[node].remove(item_index)
self.check_schedule()
def mark_test_pending(self, item):
def mark_test_pending(self, item: str) -> None:
assert self.collection is not None
self.pending.insert(
0,
self.collection.index(item),
)
self.check_schedule()
def remove_pending_tests_from_node(self, node, indices):
def remove_pending_tests_from_node(
self,
node: WorkerController,
indices: Sequence[int],
) -> None:
"""Node returned some test indices back in response to 'steal' command.
This is called by ``DSession.worker_unscheduled``.
@@ -183,7 +193,7 @@ class WorkStealingScheduling:
self.pending.extend(indices)
self.check_schedule()
def check_schedule(self):
def check_schedule(self) -> None:
"""Reschedule tests/perform load balancing."""
nodes_up = [
NodePending(node, pending)
@@ -191,7 +201,7 @@ class WorkStealingScheduling:
if not node.shutting_down
]
def get_idle_nodes():
def get_idle_nodes() -> list[WorkerController]:
return [node for node, pending in nodes_up if len(pending) < MIN_PENDING]
idle_nodes = get_idle_nodes()
@@ -235,10 +245,11 @@ class WorkStealingScheduling:
node.shutdown()
return
assert steal_from is not None
steal_from.node.send_steal(steal_from.pending[-num_steal:])
self.steal_requested_from_node = steal_from.node
def remove_node(self, node):
def remove_node(self, node: WorkerController) -> str | None:
"""Remove a node from the scheduler.
This should be called either when the node crashed or at
@@ -249,12 +260,12 @@ class WorkStealingScheduling:
Return the item which was being executing while the node
crashed or None if the node has no more pending items.
"""
pending = self.node2pending.pop(node)
# If node was removed without completing its assigned tests - it crashed
if pending:
assert self.collection is not None
crashitem = self.collection[pending.pop(0)]
else:
crashitem = None
@@ -268,7 +279,7 @@ class WorkStealingScheduling:
self.check_schedule()
return crashitem
def schedule(self):
def schedule(self) -> None:
"""Initiate distribution of the test collection.
Initiate scheduling of the items across the nodes. If this
@@ -298,14 +309,14 @@ class WorkStealingScheduling:
self.check_schedule()
def _send_tests(self, node, num):
def _send_tests(self, node: WorkerController, num: int) -> None:
tests_per_node = self.pending[:num]
if tests_per_node:
del self.pending[:num]
self.node2pending[node].extend(tests_per_node)
node.send_runtest_some(tests_per_node)
def _check_nodes_have_same_collection(self):
def _check_nodes_have_same_collection(self) -> bool:
"""Return True if all nodes have collected the same items.
If collections differ, this method returns False while logging

122
src/xdist/workermanage.py
View File

@@ -7,9 +7,12 @@ from pathlib import Path
import re
import sys
from typing import Any
from typing import Callable
from typing import Literal
from typing import Sequence
from typing import Union
import uuid
import warnings
import execnet
import pytest
@@ -17,11 +20,13 @@ import pytest
from xdist.plugin import _sys_path
import xdist.remote
from xdist.remote import Producer
from xdist.remote import WorkerInfo
def parse_spec_config(config):
def parse_spec_config(config: pytest.Config) -> list[str]:
xspeclist = []
for xspec in config.getvalue("tx"):
tx: list[str] = config.getvalue("tx")
for xspec in tx:
i = xspec.find("*")
try:
num = int(xspec[:i])
@@ -40,7 +45,12 @@ class NodeManager:
EXIT_TIMEOUT = 10
DEFAULT_IGNORES = [".*", "*.pyc", "*.pyo", "*~"]
def __init__(self, config, specs=None, defaultchdir="pyexecnetcache") -> None:
def __init__(
self,
config: pytest.Config,
specs: Sequence[execnet.XSpec | str] | None = None,
defaultchdir: str = "pyexecnetcache",
) -> None:
self.config = config
self.trace = self.config.trace.get("nodemanager")
self.testrunuid = self.config.getoption("testrunuid")
@@ -49,7 +59,7 @@ class NodeManager:
self.group = execnet.Group()
if specs is None:
specs = self._getxspecs()
self.specs = []
self.specs: list[execnet.XSpec] = []
for spec in specs:
if not isinstance(spec, execnet.XSpec):
spec = execnet.XSpec(spec)
@@ -61,31 +71,39 @@ class NodeManager:
self.rsyncoptions = self._getrsyncoptions()
self._rsynced_specs: set[tuple[Any, Any]] = set()
def rsync_roots(self, gateway):
def rsync_roots(self, gateway: execnet.Gateway) -> None:
"""Rsync the set of roots to the node's gateway cwd."""
if self.roots:
for root in self.roots:
self.rsync(gateway, root, **self.rsyncoptions)
def setup_nodes(self, putevent):
def setup_nodes(
self,
putevent: Callable[[tuple[str, dict[str, Any]]], None],
) -> list[WorkerController]:
self.config.hook.pytest_xdist_setupnodes(config=self.config, specs=self.specs)
self.trace("setting up nodes")
return [self.setup_node(spec, putevent) for spec in self.specs]
def setup_node(self, spec, putevent):
def setup_node(
self,
spec: execnet.XSpec,
putevent: Callable[[tuple[str, dict[str, Any]]], None],
) -> WorkerController:
gw = self.group.makegateway(spec)
self.config.hook.pytest_xdist_newgateway(gateway=gw)
self.rsync_roots(gw)
node = WorkerController(self, gw, self.config, putevent)
gw.node = node # keep the node alive
# Keep the node alive.
gw.node = node # type: ignore[attr-defined]
node.setup()
self.trace("started node %r" % node)
return node
def teardown_nodes(self):
def teardown_nodes(self) -> None:
self.group.terminate(self.EXIT_TIMEOUT)
def _getxspecs(self):
def _getxspecs(self) -> list[execnet.XSpec]:
return [execnet.XSpec(x) for x in parse_spec_config(self.config)]
def _getrsyncdirs(self) -> list[Path]:
@@ -97,7 +115,7 @@ class NodeManager:
import _pytest
import pytest
def get_dir(p):
def get_dir(p: str) -> str:
"""Return the directory path if p is a package or the path to the .py file otherwise."""
stripped = p.rstrip("co")
if os.path.basename(stripped) == "__init__.py":
@@ -115,14 +133,14 @@ class NodeManager:
candidates.extend(rsyncroots)
roots = []
for root in candidates:
root = Path(root).resolve()
if not root.exists():
root_path = Path(root).resolve()
if not root_path.exists():
raise pytest.UsageError(f"rsyncdir doesn't exist: {root!r}")
if root not in roots:
roots.append(root)
if root_path not in roots:
roots.append(root_path)
return roots
def _getrsyncoptions(self):
def _getrsyncoptions(self) -> dict[str, Any]:
"""Get options to be passed for rsync."""
ignores = list(self.DEFAULT_IGNORES)
ignores += [str(path) for path in self.config.option.rsyncignore]
@@ -133,7 +151,16 @@ class NodeManager:
"verbose": getattr(self.config.option, "verbose", 0),
}
def rsync(self, gateway, source, notify=None, verbose=False, ignores=None):
def rsync(
self,
gateway: execnet.Gateway,
source: str | os.PathLike[str],
notify: (
Callable[[str, execnet.XSpec, str | os.PathLike[str]], Any] | None
) = None,
verbose: int = False,
ignores: Sequence[str] | None = None,
) -> None:
"""Perform rsync to remote hosts for node."""
# XXX This changes the calling behaviour of
# pytest_xdist_rsyncstart and pytest_xdist_rsyncfinish to
@@ -153,7 +180,7 @@ class NodeManager:
if (spec, source) in self._rsynced_specs:
return
def finished():
def finished() -> None:
if notify:
notify("rsyncrootready", spec, source)
@@ -189,11 +216,19 @@ class HostRSync(execnet.RSync):
else:
return True
def add_target_host(self, gateway, finished=None):
def add_target_host(
self,
gateway: execnet.Gateway,
finished: Callable[[], None] | None = None,
) -> None:
remotepath = os.path.basename(self._sourcedir)
super().add_target(gateway, remotepath, finishedcallback=finished, delete=True)
def _report_send_file(self, gateway, modified_rel_path):
def _report_send_file(
self,
gateway: execnet.Gateway, # type: ignore[override]
modified_rel_path: str,
) -> None:
if self._verbose > 0:
path = os.path.basename(self._sourcedir) + "/" + modified_rel_path
remotepath = gateway.spec.chdir
@@ -234,12 +269,21 @@ class Marker(enum.Enum):
class WorkerController:
# Set when the worker is ready.
workerinfo: WorkerInfo
class RemoteHook:
@pytest.hookimpl(trylast=True)
def pytest_xdist_getremotemodule(self):
def pytest_xdist_getremotemodule(self) -> Any:
return xdist.remote
def __init__(self, nodemanager, gateway, config, putevent):
def __init__(
self,
nodemanager: NodeManager,
gateway: execnet.Gateway,
config: pytest.Config,
putevent: Callable[[tuple[str, dict[str, Any]]], None],
) -> None:
config.pluginmanager.register(self.RemoteHook())
self.nodemanager = nodemanager
self.putevent = putevent
@@ -255,14 +299,14 @@ class WorkerController:
self._shutdown_sent = False
self.log = Producer(f"workerctl-{gateway.id}", enabled=config.option.debug)
def __repr__(self):
def __repr__(self) -> str:
return f"<{self.__class__.__name__} {self.gateway.id}>"
@property
def shutting_down(self):
def shutting_down(self) -> bool:
return self._down or self._shutdown_sent
def setup(self):
def setup(self) -> None:
self.log("setting up worker session")
spec = self.gateway.spec
args = [str(x) for x in self.config.invocation_params.args or ()]
@@ -283,10 +327,11 @@ class WorkerController:
change_sys_path = _sys_path if self.gateway.spec.popen else None
self.channel.send((self.workerinput, args, option_dict, change_sys_path))
if self.putevent:
# putevent is only None in a test.
if self.putevent: # type: ignore[truthy-function]
self.channel.setcallback(self.process_from_remote, endmarker=Marker.END)
def ensure_teardown(self):
def ensure_teardown(self) -> None:
if hasattr(self, "channel"):
if not self.channel.isclosed():
self.log("closing", self.channel)
@@ -297,16 +342,16 @@ class WorkerController:
self.gateway.exit()
# del self.gateway
def send_runtest_some(self, indices):
def send_runtest_some(self, indices: Sequence[int]) -> None:
self.sendcommand("runtests", indices=indices)
def send_runtest_all(self):
def send_runtest_all(self) -> None:
self.sendcommand("runtests_all")
def send_steal(self, indices):
def send_steal(self, indices: Sequence[int]) -> None:
self.sendcommand("steal", indices=indices)
def shutdown(self):
def shutdown(self) -> None:
if not self._down:
try:
self.sendcommand("shutdown")
@@ -314,16 +359,18 @@ class WorkerController:
pass
self._shutdown_sent = True
def sendcommand(self, name, **kwargs):
def sendcommand(self, name: str, **kwargs: object) -> None:
"""Send a named parametrized command to the other side."""
self.log(f"sending command {name}(**{kwargs})")
self.channel.send((name, kwargs))
def notify_inproc(self, eventname, **kwargs):
def notify_inproc(self, eventname: str, **kwargs: object) -> None:
self.log(f"queuing {eventname}(**{kwargs})")
self.putevent((eventname, kwargs))
def process_from_remote(self, eventcall):
def process_from_remote(
self, eventcall: tuple[str, dict[str, Any]] | Literal[Marker.END]
) -> None:
"""This gets called for each object we receive from
the other side and if the channel closes.
@@ -333,7 +380,7 @@ class WorkerController:
"""
try:
if eventcall is Marker.END:
err = self.channel._getremoteerror()
err: object | None = self.channel._getremoteerror() # type: ignore[no-untyped-call]
if not self._down:
if not err or isinstance(err, EOFError):
err = "Not properly terminated" # lost connection?
@@ -399,9 +446,8 @@ class WorkerController:
self.notify_inproc("errordown", node=self, error=excinfo)
def unserialize_warning_message(data):
def unserialize_warning_message(data: dict[str, Any]) -> warnings.WarningMessage:
import importlib
import warnings
if data["message_module"]:
mod = importlib.import_module(data["message_module"])
@@ -438,4 +484,4 @@ def unserialize_warning_message(data):
continue
kwargs[attr_name] = data[attr_name]
return warnings.WarningMessage(**kwargs) # type: ignore[arg-type]
return warnings.WarningMessage(**kwargs)