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

fix test

Browse Source
This commit is contained in:
Shane Smiskol
2025-07-25 19:35:11 -07:00
parent 9d7dcae9d2
commit 28c0a72e3a
2 changed files with 17 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/xdist/workermanage.py
View File

@@ -97,17 +97,19 @@ class NodeManager:
self.trace("setting up nodes") self.trace("setting up nodes")
with ThreadPoolExecutor(max_workers=len(self.specs)) as executor: with ThreadPoolExecutor(max_workers=len(self.specs)) as executor:
futs = [ futs = [
executor.submit(self.setup_node, spec, putevent) for spec in self.specs executor.submit(self.setup_node, idx, spec, putevent) for idx, spec in enumerate(self.specs)
] ]
return [f.result() for f in futs] return [f.result() for f in futs]
def setup_node( def setup_node(
self, self,
idx: int,
spec: execnet.XSpec, spec: execnet.XSpec,
putevent: Callable[[tuple[str, dict[str, Any]]], None], putevent: Callable[[tuple[str, dict[str, Any]]], None],
) -> WorkerController: ) -> WorkerController:
if getattr(spec, "execmodel", None) != "main_thread_only": if getattr(spec, "execmodel", None) != "main_thread_only":
spec = execnet.XSpec(f"execmodel=main_thread_only//{spec}") spec = execnet.XSpec(f"execmodel=main_thread_only//{spec}")
spec = execnet.XSpec(f"{spec}//id=gw{idx}")
gw = self.group.makegateway(spec) gw = self.group.makegateway(spec)
self.config.hook.pytest_xdist_newgateway(gateway=gw) self.config.hook.pytest_xdist_newgateway(gateway=gw)
self.rsync_roots(gw) self.rsync_roots(gw)

19
testing/test_workermanage.py
View File

@@ -82,11 +82,20 @@ class TestNodeManagerPopen:
call = hookrecorder.popcall("pytest_xdist_setupnodes") call = hookrecorder.popcall("pytest_xdist_setupnodes")
assert len(call.specs) == 2 assert len(call.specs) == 2
call = hookrecorder.popcall("pytest_xdist_newgateway") # check expected gateways
assert call.gateway.spec == execnet.XSpec("execmodel=main_thread_only//popen") gw_calls = [
assert call.gateway.id == "gw0" hookrecorder.popcall("pytest_xdist_newgateway"),
call = hookrecorder.popcall("pytest_xdist_newgateway") hookrecorder.popcall("pytest_xdist_newgateway"),
assert call.gateway.id == "gw1" ]
ids = {c.gateway.id for c in gw_calls}
assert ids == {"gw0", "gw1"}
for c in gw_calls:
expected_spec = execnet.XSpec(
f"execmodel=main_thread_only//popen//id={c.gateway.id}"
)
assert c.gateway.spec == expected_spec
assert len(hm.group) == 2 assert len(hm.group) == 2
hm.teardown_nodes() hm.teardown_nodes()
assert not len(hm.group) assert not len(hm.group)