How to use the execnet.XSpec function in execnet

To help you get started, we’ve selected a few execnet examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github pytest-dev / pytest-xdist / testing / test_plugin.py View on Github external
def test_getrsyncignore(self, testdir):
        config = testdir.parseconfigure("--rsyncignore=fo*")
        nm = NodeManager(config, specs=[execnet.XSpec("popen//chdir=qwe")])
        assert "fo*" in nm.rsyncoptions["ignores"]
github pytest-dev / execnet / execnet / multi.py View on Github external
Valid types: ``popen``, ``ssh=hostname``, ``socket=host:port``.
        Valid configuration::

            id=     specifies the gateway id
            python=
github pytest-dev / pytest-xdist / xdist / gwmanage.py View on Github external
def __init__(self, specs, hook, defaultchdir="pyexecnetcache"):
        self.specs = []
        self.hook = hook
        self.group = execnet.Group()
        for spec in specs:
            if not isinstance(spec, execnet.XSpec):
                spec = execnet.XSpec(spec)
            if not spec.chdir and not spec.popen:
                spec.chdir = defaultchdir
            self.specs.append(spec)
github pytest-dev / pytest-xdist / testing / test_txnode.py View on Github external
def makenode(self, config=None, xspec="popen"):
        if config is None:
            testdir = self.request.getfuncargvalue("testdir")
            config = testdir.reparseconfig([])
        self.config = config
        self.queue = Queue()
        self.xspec = execnet.XSpec(xspec)
        self.gateway = execnet.makegateway(self.xspec)
        self.id += 1
        self.gateway.id = str(self.id)
        self.nodemanager = None
        self.node = TXNode(self.nodemanager, self.gateway, self.config, putevent=self.queue.put)
        assert not self.node.channel.isclosed()
        return self.node
github pytest-dev / pytest-xdist / src / xdist / workermanage.py View on Github external
def _getxspecs(self):
        return [execnet.XSpec(x) for x in parse_spec_config(self.config)]
github pytest-dev / pytest-xdist / testing / test_txnode.py View on Github external
def test_send_one_with_env(self, testdir, mysetup, monkeypatch):
        if execnet.XSpec("popen").env is None:
            py.test.skip("requires execnet 1.0.7 or above")
        monkeypatch.delenv('ENV1', raising=False)
        monkeypatch.delenv('ENV2', raising=False)
        monkeypatch.setenv('ENV3', 'var3')

        item = testdir.getitem("""
            def test_func():
                import os
                # ENV1, ENV2 set by xspec; ENV3 inherited from parent process
                assert os.getenv('ENV2') == 'var2'
                assert os.getenv('ENV1') == 'var1'
                assert os.getenv('ENV3') == 'var3'
        """)
        node = mysetup.makenode(item.config,
                xspec="popen//env:ENV1=var1//env:ENV2=var2")
        node.send(item)
github pytest-dev / execnet / testing / test_gateway.py View on Github external
def test_popen_args(spec, expected_args):
    expected_args = expected_args + ["-u", "-c", gateway_io.popen_bootstrapline]
    args = gateway_io.popen_args(execnet.XSpec(spec))
    assert args == expected_args
github pytest-dev / pytest-xdist / src / xdist / workermanage.py View on Github external
def __init__(self, config, specs=None, defaultchdir="pyexecnetcache"):
        self.config = config
        self.trace = self.config.trace.get("nodemanager")
        self.testrunuid = self.config.getoption("testrunuid")
        if self.testrunuid is None:
            self.testrunuid = uuid.uuid4().hex
        self.group = execnet.Group()
        if specs is None:
            specs = self._getxspecs()
        self.specs = []
        for spec in specs:
            if not isinstance(spec, execnet.XSpec):
                spec = execnet.XSpec(spec)
            if not spec.chdir and not spec.popen:
                spec.chdir = defaultchdir
            self.group.allocate_id(spec)
            self.specs.append(spec)
        self.roots = self._getrsyncdirs()
        self.rsyncoptions = self._getrsyncoptions()
        self._rsynced_specs = set()
github pytest-dev / pytest / testing / pytest / dist / test_txnode.py View on Github external
def makenode(self, config=None):
        if config is None:
            testdir = self.request.getfuncargvalue("testdir")
            config = testdir.reparseconfig([])
        self.config = config
        self.queue = Queue()
        self.xspec = execnet.XSpec("popen")
        self.gateway = execnet.makegateway(self.xspec)
        self.id += 1
        self.gateway.id = str(self.id)
        self.node = TXNode(self.gateway, self.config, putevent=self.queue.put)
        assert not self.node.channel.isclosed()
        return self.node
github alfredodeza / remoto / remoto / lib / execnet / multi.py View on Github external
Valid types: ``popen``, ``ssh=hostname``, ``socket=host:port``.
        Valid configuration::

            id=     specifies the gateway id
            python=