Merge pull request #1171 from Alon-L/feature/proxy-gateways

Add proxy gateways

changelog/1170.feature

@@ -0,0 +1,4 @@
+Add the `--px` arg to create proxy gateways.
+
+Proxy gateways are passed to additional gateways using the `via` keyword.
+They can serve as a way to run multiple workers on remote machines.

docs/remote.rst

@@ -66,6 +66,20 @@ new socket host with something like this::
     pytest -d --tx socket=192.168.1.102:8888 --rsyncdir mypkg
 
 
+Using proxies to run multiple workers on remote machines
+---------------------------------------
+
+In case you want to run multiple workers on a remote machine,
+you can create a proxy gateway for the machine, and run multiple
+workers using the `via` attribute.::
+
+    pytest -d --px id=my_proxy//socket=192.168.1.102:8888 --tx 5*popen//via=my_proxy
+
+Here we declare a proxy gateway using the `--px` arg, and
+create 5 workers that run on the remote server using the proxy.
+Note that the proxy gateway does not run a worker, thus only 5
+workers are created.
+
 
 Running tests on many platforms at once
 ---------------------------------------

src/xdist/plugin.py

@@ -139,6 +139,17 @@ def pytest_addoption(parser: pytest.Parser) -> None:
             "--tx ssh=user@codespeak.net//chdir=testcache"
         ),
     )
+    group.addoption(
+        "--px",
+        dest="px",
+        action="append",
+        default=[],
+        metavar="xspec",
+        help=(
+            "Add a proxy gateway to pass to test execution environments using `via`. Example:\n"
+            "--px id=my_proxy//socket=192.168.1.102:8888 --tx 5*popen//via=my_proxy"
+        ),
+    )
     group._addoption(
         "-d",
         action="store_true",

src/xdist/scheduler/each.py

@@ -6,7 +6,7 @@ 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 parse_tx_spec_config
 from xdist.workermanage import WorkerController
 
 
@@ -26,7 +26,7 @@ class EachScheduling:
 
     def __init__(self, config: pytest.Config, log: Producer | None = None) -> None:
         self.config = config
-        self.numnodes = len(parse_spec_config(config))
+        self.numnodes = len(parse_tx_spec_config(config))
         self.node2collection: dict[WorkerController, list[str]] = {}
         self.node2pending: dict[WorkerController, list[int]] = {}
         self._started: list[WorkerController] = []

src/xdist/scheduler/load.py

@@ -7,7 +7,7 @@ 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 parse_tx_spec_config
 from xdist.workermanage import WorkerController
 
 
@@ -58,7 +58,7 @@ class LoadScheduling:
     """
 
     def __init__(self, config: pytest.Config, log: Producer | None = None) -> None:
-        self.numnodes = len(parse_spec_config(config))
+        self.numnodes = len(parse_tx_spec_config(config))
         self.node2collection: dict[WorkerController, list[str]] = {}
         self.node2pending: dict[WorkerController, list[int]] = {}
         self.pending: list[int] = []

src/xdist/scheduler/loadscope.py

@@ -8,7 +8,7 @@ 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 parse_tx_spec_config
 from xdist.workermanage import WorkerController
 
 
@@ -91,7 +91,7 @@ class LoadScopeScheduling:
     """
 
     def __init__(self, config: pytest.Config, log: Producer | None = None) -> None:
-        self.numnodes = len(parse_spec_config(config))
+        self.numnodes = len(parse_tx_spec_config(config))
         self.collection: list[str] | None = None
 
         self.workqueue: OrderedDict[str, dict[str, bool]] = OrderedDict()

src/xdist/scheduler/worksteal.py

@@ -7,7 +7,7 @@ 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 parse_tx_spec_config
 from xdist.workermanage import WorkerController
 
 
@@ -65,7 +65,7 @@ class WorkStealingScheduling:
     """
 
     def __init__(self, config: pytest.Config, log: Producer | None = None) -> None:
-        self.numnodes = len(parse_spec_config(config))
+        self.numnodes = len(parse_tx_spec_config(config))
         self.node2collection: dict[WorkerController, list[str]] = {}
         self.node2pending: dict[WorkerController, list[int]] = {}
         self.pending: list[int] = []

src/xdist/workermanage.py

@@ -23,7 +23,7 @@ from xdist.remote import Producer
 from xdist.remote import WorkerInfo
 
 
-def parse_spec_config(config: pytest.Config) -> list[str]:
+def parse_tx_spec_config(config: pytest.Config) -> list[str]:
     xspeclist = []
     tx: list[str] = config.getvalue("tx")
     for xspec in tx:
@@ -57,8 +57,17 @@ class NodeManager:
         if self.testrunuid is None:
             self.testrunuid = uuid.uuid4().hex
         self.group = execnet.Group(execmodel="main_thread_only")
+        for proxy_spec in self._getpxspecs():
+            # Proxy gateways do not run workers, and are meant to be passed with the `via` attribute
+            # to additional gateways.
+            # They are useful for running multiple workers on remote machines.
+            if getattr(proxy_spec, "id", None) is None:
+                raise pytest.UsageError(
+                    f"Proxy gateway {proxy_spec} must include an id"
+                )
+            self.group.makegateway(proxy_spec)
         if specs is None:
-            specs = self._getxspecs()
+            specs = self._gettxspecs()
         self.specs: list[execnet.XSpec] = []
         for spec in specs:
             if not isinstance(spec, execnet.XSpec):
@@ -107,8 +116,11 @@ class NodeManager:
     def teardown_nodes(self) -> None:
         self.group.terminate(self.EXIT_TIMEOUT)
 
-    def _getxspecs(self) -> list[execnet.XSpec]:
+    def _gettxspecs(self) -> list[execnet.XSpec]:
-        return [execnet.XSpec(x) for x in parse_spec_config(self.config)]
+        return [execnet.XSpec(x) for x in parse_tx_spec_config(self.config)]
+
+    def _getpxspecs(self) -> list[execnet.XSpec]:
+        return [execnet.XSpec(x) for x in self.config.getoption("px")]
 
     def _getrsyncdirs(self) -> list[Path]:
         for spec in self.specs:

testing/test_plugin.py

@@ -295,7 +295,7 @@ class TestDistOptions:
     def test_getxspecs(self, pytester: pytest.Pytester) -> None:
         config = pytester.parseconfigure("--tx=popen", "--tx", "ssh=xyz")
         nodemanager = NodeManager(config)
-        xspecs = nodemanager._getxspecs()
+        xspecs = nodemanager._gettxspecs()
         assert len(xspecs) == 2
         print(xspecs)
         assert xspecs[0].popen
@@ -303,7 +303,7 @@ class TestDistOptions:
 
     def test_xspecs_multiplied(self, pytester: pytest.Pytester) -> None:
         config = pytester.parseconfigure("--tx=3*popen")
-        xspecs = NodeManager(config)._getxspecs()
+        xspecs = NodeManager(config)._gettxspecs()
         assert len(xspecs) == 3
         assert xspecs[1].popen
 

testing/test_workermanage.py

@@ -368,6 +368,34 @@ class TestNodeManager:
         (rep,) = reprec.getreports("pytest_runtest_logreport")
         assert rep.passed
 
+    def test_proxy_gateway_setup_nodes(self, pytester: pytest.Pytester) -> None:
+        nodemanager = NodeManager(
+            pytester.parseconfig(
+                "--px", "popen//id=my_proxy", "--tx", "popen//via=my_proxy"
+            )
+        )
+        nodes = nodemanager.setup_nodes(None)  # type: ignore[arg-type]
+
+        # Proxy gateways are considered as an execnet gateway
+        assert len(nodemanager.group) == 2
+
+        # Proxy gateways do not run workers
+        assert len(nodes) == 1
+
+    def test_proxy_gateway(self, pytester: pytest.Pytester) -> None:
+        pytester.makepyfile(
+            __init__="",
+            test_x="""
+            def test_one():
+                pass
+        """,
+        )
+        reprec = pytester.inline_run(
+            "-d", "--px", "popen//id=my_proxy", "--tx", "popen//via=my_proxy"
+        )
+        rep = reprec.getreports("pytest_runtest_logreport")
+        assert rep[1].passed
+
 
 class MyWarning(UserWarning):
     pass