def test_getrsyncdirs(self, testdir):
     config = testdir.parseconfigure('--rsyncdir=' + str(testdir.tmpdir))
     nm = NodeManager(config, specs=[execnet.XSpec("popen")])
     assert not nm._getrsyncdirs()
     nm = NodeManager(config, specs=[execnet.XSpec("popen//chdir=qwe")])
     assert nm.roots
     assert testdir.tmpdir in nm.roots

 def test_getxspecs(self, testdir):
     config = testdir.parseconfigure("--tx=popen", "--tx", "ssh=xyz")
     nodemanager = NodeManager(config)
     xspecs = nodemanager._getxspecs()
     assert len(xspecs) == 2
     print(xspecs)
     assert xspecs[0].popen
     assert xspecs[1].ssh == "xyz"

 def test_rsync_same_popen_twice(self, config, mysetup, hookrecorder):
     source, dest = mysetup.source, mysetup.dest
     hm = NodeManager(config, ["popen//chdir=%s" %dest] * 2)
     hm.makegateways()
     source.ensure("dir1", "dir2", "hello")
     hm.rsync(source)
     call = hookrecorder.popcall("pytest_xdist_rsyncstart")
     assert call.source == source
     assert len(call.gateways) == 1
     assert call.gateways[0] in hm.group
     call = hookrecorder.popcall("pytest_xdist_rsyncfinish")

 def test_optimise_popen(self, testdir, mysetup, slavecontroller):
     source = mysetup.source
     specs = ["popen"] * 3
     source.join("conftest.py").write("rsyncdirs = ['a']")
     source.ensure('a', dir=1)
     config = testdir.parseconfig(source)
     nodemanager = NodeManager(config, specs)
     nodemanager.setup_nodes(None)  # calls .rysnc_roots()
     for gwspec in nodemanager.specs:
         assert gwspec._samefilesystem()
         assert not gwspec.chdir

 def test_optimise_popen(self, testdir, mysetup):
     source, dest = mysetup.source, mysetup.dest
     specs = ["popen"] * 3
     source.join("conftest.py").write("rsyncdirs = ['a']")
     source.ensure('a', dir=1)
     config = testdir.parseconfig(source)
     nodemanager = NodeManager(config, specs)
     nodemanager.makegateways()
     nodemanager.rsync_roots()
     for gwspec in nodemanager.specs:
         assert gwspec._samefilesystem()
         assert not gwspec.chdir

 def test_rsync_roots_no_roots(self, testdir, mysetup):
     mysetup.source.ensure("dir1", "file1").write("hello")
     config = testdir.parseconfig(mysetup.source)
     nodemanager = NodeManager(config, ["popen//chdir=%s" % mysetup.dest])
     # assert nodemanager.config.topdir == source == config.topdir
     nodemanager.makegateways()
     nodemanager.rsync_roots()
     p, = nodemanager.gwmanager.multi_exec(
         "import os ; channel.send(os.getcwd())").receive_each()
     p = py.path.local(p)
     py.builtin.print_("remote curdir", p)
     assert p == mysetup.dest.join(config.topdir.basename)
     assert p.join("dir1").check()
     assert p.join("dir1", "file1").check()

    def test_popen_makegateway_events(self, config, hookrecorder, _pytest):
        hm = NodeManager(config, ["popen"] * 2)
        hm.makegateways()
        call = hookrecorder.popcall("pytest_xdist_setupnodes")
        assert len(call.specs) == 2

        call = hookrecorder.popcall("pytest_xdist_newgateway")
        assert call.gateway.spec == execnet.XSpec("popen")
        assert call.gateway.id == "gw0"
        call = hookrecorder.popcall("pytest_xdist_newgateway")
        assert call.gateway.id == "gw1"
        assert len(hm.group) == 2
        hm.teardown_nodes()
        assert not len(hm.group)

 def test_init_rsync_roots(self, testdir, mysetup, slavecontroller):
     source, dest = mysetup.source, mysetup.dest
     dir2 = source.ensure("dir1", "dir2", dir=1)
     source.ensure("dir1", "somefile", dir=1)
     dir2.ensure("hello")
     source.ensure("bogusdir", "file")
     source.join("tox.ini").write(py.std.textwrap.dedent("""
         [pytest]
         rsyncdirs=dir1/dir2
     """))
     config = testdir.parseconfig(source)
     nodemanager = NodeManager(config, ["popen//chdir=%s" % dest])
     nodemanager.setup_nodes(None)  # calls .rsync_roots()
     assert dest.join("dir2").check()
     assert not dest.join("dir1").check()
     assert not dest.join("bogus").check()

 def test_getrsyncdirs_with_conftest(self, testdir):
     p = py.path.local()
     for bn in 'x y z'.split():
         p.mkdir(bn)
     testdir.makeini("""
         [pytest]
         rsyncdirs= x
     """)
     config = testdir.parseconfigure(
         testdir.tmpdir, '--rsyncdir=y', '--rsyncdir=z')
     nm = NodeManager(config, specs=[execnet.XSpec("popen//chdir=xyz")])
     roots = nm._getrsyncdirs()
     # assert len(roots) == 3 + 1 # pylib
     assert py.path.local('y') in roots
     assert py.path.local('z') in roots
     assert testdir.tmpdir.join('x') in roots

 def test_popen_rsync_subdir(self, testdir, mysetup, slavecontroller):
     source, dest = mysetup.source, mysetup.dest
     dir1 = mysetup.source.mkdir("dir1")
     dir2 = dir1.mkdir("dir2")
     dir2.ensure("hello")
     for rsyncroot in (dir1, source):
         dest.remove()
         nodemanager = NodeManager(testdir.parseconfig(
             "--tx", "popen//chdir=%s" % dest, "--rsyncdir", rsyncroot,
             source, ))
         nodemanager.setup_nodes(None)  # calls .rsync_roots()
         if rsyncroot == source:
             dest = dest.join("source")
         assert dest.join("dir1").check()
         assert dest.join("dir1", "dir2").check()
         assert dest.join("dir1", "dir2", 'hello').check()
         nodemanager.teardown_nodes()

    def pytest_sessionstart(self, session):
        """Creates and starts the nodes.

        The nodes are setup to put their events onto self.queue.  As
        soon as nodes start they will emit the slave_slaveready event.
        """
        self.nodemanager = NodeManager(self.config)
        nodes = self.nodemanager.setup_nodes(putevent=self.queue.put)
        self._active_nodes.update(nodes)

 def test_rsyncignore(self, testdir, mysetup):
     source, dest = mysetup.source, mysetup.dest
     dir2 = source.ensure("dir1", "dir2", dir=1)
     dir5 = source.ensure("dir5", "dir6", "bogus")
     dirf = source.ensure("dir5", "file")
     dir2.ensure("hello")
     source.join("tox.ini").write(py.std.textwrap.dedent("""
         [pytest]
         rsyncdirs = dir1 dir5
         rsyncignore = dir1/dir2 dir5/dir6
     """))
     config = testdir.parseconfig(source)
     nodemanager = NodeManager(config, ["popen//chdir=%s" % dest])
     nodemanager.makegateways()
     nodemanager.rsync_roots()
     assert dest.join("dir1").check()
     assert not dest.join("dir1", "dir2").check()
     assert dest.join("dir5","file").check()
     assert not dest.join("dir6").check()

    def pytest_sessionstart(self, session):
        """Creates and starts the nodes.

        The nodes are setup to put their events onto self.queue.  As
        soon as nodes start they will emit the slave_slaveready event.
        """
        self.nodemanager = NodeManager(self.config)
        nodes = self.nodemanager.setup_nodes(putevent=self.queue.put)
        for node in nodes:
            node.channel.send(("RUNNING_MODE", self.dist))

        self._active_nodes.update(nodes)
        self._session = session

 def test_rsyncignore(self, testdir, mysetup, slavecontroller):
     source, dest = mysetup.source, mysetup.dest
     dir2 = source.ensure("dir1", "dir2", dir=1)
     source.ensure("dir5", "dir6", "bogus")
     source.ensure("dir5", "file")
     dir2.ensure("hello")
     source.ensure("foo", "bar")
     source.ensure("bar", "foo")
     source.join("tox.ini").write(py.std.textwrap.dedent("""
         [pytest]
         rsyncdirs = dir1 dir5
         rsyncignore = dir1/dir2 dir5/dir6 foo*
     """))
     config = testdir.parseconfig(source)
     config.option.rsyncignore = ['bar']
     nodemanager = NodeManager(config, ["popen//chdir=%s" % dest])
     nodemanager.setup_nodes(None)  # calls .rsync_roots()
     assert dest.join("dir1").check()
     assert not dest.join("dir1", "dir2").check()
     assert dest.join("dir5", "file").check()
     assert not dest.join("dir6").check()
     assert not dest.join('foo').check()
     assert not dest.join('bar').check()

 def test_rsync_popen_with_path(self, config, mysetup):
     source, dest = mysetup.source, mysetup.dest
     hm = NodeManager(config, ["popen//chdir=%s" %dest] * 1)
     hm.makegateways()
     source.ensure("dir1", "dir2", "hello")
     l = []
     hm.rsync(source, notify=lambda *args: l.append(args))
     assert len(l) == 1
     assert l[0] == ("rsyncrootready", hm.group['gw0'].spec, source)
     hm.teardown_nodes()
     dest = dest.join(source.basename)
     assert dest.join("dir1").check()
     assert dest.join("dir1", "dir2").check()
     assert dest.join("dir1", "dir2", 'hello').check()

 def test_rsync_same_popen_twice(self, config, mysetup, hookrecorder,
                                 slavecontroller):
     source, dest = mysetup.source, mysetup.dest
     hm = NodeManager(config, ["popen//chdir=%s" % dest] * 2)
     hm.roots = []
     hm.setup_nodes(None)
     source.ensure("dir1", "dir2", "hello")
     gw = hm.group[0]
     hm.rsync(gw, source)
     call = hookrecorder.popcall("pytest_xdist_rsyncstart")
     assert call.source == source
     assert len(call.gateways) == 1
     assert call.gateways[0] in hm.group
     call = hookrecorder.popcall("pytest_xdist_rsyncfinish")

 def test_popens_rsync(self, config, mysetup):
     source = mysetup.source
     hm = NodeManager(config, ["popen"] * 2)
     hm.makegateways()
     assert len(hm.group) == 2
     for gw in hm.group:
         class pseudoexec:
             args = []
             def __init__(self, *args):
                 self.args.extend(args)
             def waitclose(self):
                 pass
         gw.remote_exec = pseudoexec
     l = []
     hm.rsync(source, notify=lambda *args: l.append(args))
     assert not l
     hm.teardown_nodes()
     assert not len(hm.group)
     assert "sys.path.insert" in gw.remote_exec.args[0]

 def pytest_sessionstart(self, session):
     self.nodemanager = NodeManager(self.config)
     self.nodemanager.setup_nodes(putevent=self.queue.put)

class DSession:
    def __init__(self, config):
        self.config = config
        self.log = py.log.Producer("dsession")
        if not config.option.debug:
            py.log.setconsumer(self.log._keywords, None)
        self.shuttingdown = False
        self.countfailures = 0
        self.maxfail = config.getvalue("maxfail")
        self.queue = queue.Queue()
        try:
            self.terminal = config.pluginmanager.getplugin("terminalreporter")
        except KeyError:
            self.terminal = None
        else:
            self.trdist = TerminalDistReporter(config)
            config.pluginmanager.register(self.trdist, "terminaldistreporter")

    def report_line(self, line):
        if self.terminal and self.config.option.verbose >= 0:
            self.terminal.write_line(line)

    @pytest.mark.trylast
    def pytest_sessionstart(self, session):
        self.nodemanager = NodeManager(self.config)
        self.nodemanager.setup_nodes(putevent=self.queue.put)

    def pytest_sessionfinish(self, session):
        """ teardown any resources after a test run. """
        nm = getattr(self, "nodemanager", None)  # if not fully initialized
        if nm is not None:
            nm.teardown_nodes()

    def pytest_collection(self):
        # prohibit collection of test items in master process
        return True

    def pytest_runtestloop(self):
        numnodes = len(self.nodemanager.specs)
        dist = self.config.getvalue("dist")
        if dist == "load":
            self.sched = LoadScheduling(numnodes, log=self.log)
        elif dist == "each":
            self.sched = EachScheduling(numnodes, log=self.log)
        else:
            assert 0, dist
        self.shouldstop = False
        self.session_finished = False
        while not self.session_finished:
            self.loop_once()
            if self.shouldstop:
                raise Interrupted(str(self.shouldstop))
        return True

    def loop_once(self):
        """ process one callback from one of the slaves. """
        while 1:
            try:
                eventcall = self.queue.get(timeout=2.0)
                break
            except queue.Empty:
                continue
        callname, kwargs = eventcall
        assert callname, kwargs
        method = "slave_" + callname
        call = getattr(self, method)
        self.log("calling method: %s(**%s)" % (method, kwargs))
        call(**kwargs)
        if self.sched.tests_finished():
            self.triggershutdown()

    #
    # callbacks for processing events from slaves
    #

    def slave_slaveready(self, node, slaveinfo):
        node.slaveinfo = slaveinfo
        node.slaveinfo["id"] = node.gateway.id
        node.slaveinfo["spec"] = node.gateway.spec
        self.config.hook.pytest_testnodeready(node=node)
        self.sched.addnode(node)
        if self.shuttingdown:
            node.shutdown()

    def slave_slavefinished(self, node):
        self.config.hook.pytest_testnodedown(node=node, error=None)
        if node.slaveoutput["exitstatus"] == 2:  # keyboard-interrupt
            self.shouldstop = "%s received keyboard-interrupt" % (node,)
            self.slave_errordown(node, "keyboard-interrupt")
            return
        crashitem = self.sched.remove_node(node)
        # assert not crashitem, (crashitem, node)
        if self.shuttingdown and not self.sched.hasnodes():
            self.session_finished = True

    def slave_errordown(self, node, error):
        self.config.hook.pytest_testnodedown(node=node, error=error)
        try:
            crashitem = self.sched.remove_node(node)
        except KeyError:
#.........这里部分代码省略.........

class DSession:
    """A py.test plugin which runs a distributed test session

    At the beginning of the test session this creates a NodeManager
    instance which creates and starts all nodes.  Nodes then emit
    events processed in the pytest_runtestloop hook using the slave_*
    methods.

    Once a node is started it will automatically start running the
    py.test mainloop with some custom hooks.  This means a node
    automatically starts collecting tests.  Once tests are collected
    it will wait for instructions.
    """
    def __init__(self, config):
        self.config = config
        self.log = py.log.Producer("dsession")
        if not config.option.debug:
            py.log.setconsumer(self.log._keywords, None)
        self.shuttingdown = False
        self.countfailures = 0
        self.maxfail = config.getvalue("maxfail")
        self.queue = queue.Queue()
        self._session = None
        self._failed_collection_errors = {}
        self._active_nodes = set()
        self._failed_nodes_count = 0
        self._max_slave_restart = self.config.getoption('max_slave_restart')
        if self._max_slave_restart is not None:
            self._max_slave_restart = int(self._max_slave_restart)
        try:
            self.terminal = config.pluginmanager.getplugin("terminalreporter")
        except KeyError:
            self.terminal = None
        else:
            self.trdist = TerminalDistReporter(config)
            config.pluginmanager.register(self.trdist, "terminaldistreporter")

    @property
    def session_finished(self):
        """Return True if the distributed session has finished

        This means all nodes have executed all test items.  This is
        used to by pytest_runtestloop to break out of it's loop.
        """
        return bool(self.shuttingdown and not self._active_nodes)

    def report_line(self, line):
        if self.terminal and self.config.option.verbose >= 0:
            self.terminal.write_line(line)

    @pytest.mark.trylast
    def pytest_sessionstart(self, session):
        """Creates and starts the nodes.

        The nodes are setup to put their events onto self.queue.  As
        soon as nodes start they will emit the slave_slaveready event.
        """
        self.nodemanager = NodeManager(self.config)
        nodes = self.nodemanager.setup_nodes(putevent=self.queue.put)
        self._active_nodes.update(nodes)
        self._session = session

    def pytest_sessionfinish(self, session):
        """Shutdown all nodes."""
        nm = getattr(self, 'nodemanager', None)  # if not fully initialized
        if nm is not None:
            nm.teardown_nodes()
        self._session = None

    def pytest_collection(self):
        # prohibit collection of test items in master process
        return True

    def pytest_runtestloop(self):
        numnodes = len(self.nodemanager.specs)
        dist = self.config.getvalue("dist")
        if dist == "load":
            self.sched = LoadScheduling(numnodes, log=self.log,
                                        config=self.config)
        elif dist == "each":
            self.sched = EachScheduling(numnodes, log=self.log)
        else:
            assert 0, dist
        self.shouldstop = False
        while not self.session_finished:
            self.loop_once()
            if self.shouldstop:
                raise Interrupted(str(self.shouldstop))
        return True

    def loop_once(self):
        """Process one callback from one of the slaves."""
        while 1:
            try:
                eventcall = self.queue.get(timeout=2.0)
                break
            except queue.Empty:
                continue
        callname, kwargs = eventcall
        assert callname, kwargs
#.........这里部分代码省略.........