rework dist-testing model: now all items are collected at the start. move back some tests to pytest-core

testing/test_dsession.py

@@ -16,8 +16,11 @@ class MockNode:
     def __init__(self):
         self.sent = []
 
+    def send(self, item):
+        self.sent.append(item)
+
     def sendlist(self, items):
-        self.sent.append(items)
+        self.sent.extend(items)
 
     def shutdown(self):
         self._shutdown=True
@@ -79,24 +82,28 @@ class TestDSession:
                 pass
         """)
         session = DSession(modcol.config)
-        session.triggertesting([modcol])
+        reprec = testdir.getreportrecorder(session)
-        name, args, kwargs = session.queue.get(block=False)
+        items = session.collect_all_items([modcol])
-        assert name == 'pytest_collectreport'
+        assert len(items) == 1
-        report = kwargs['report']  
+        calls= reprec.getcalls("pytest_collectreport")
-        assert len(report.result) == 1
+        assert len(calls) == 1
+        call = calls[0]
+        assert len(call.report.result) == 1
 
-    def test_triggertesting_item(self, testdir):
+    def test_senditems_load(self, testdir, monkeypatch):
         item = testdir.getitem("def test_func(): pass")
         session = DSession(item.config)
         node1 = MockNode()
         node2 = MockNode()
         session.addnode(node1)
         session.addnode(node2)
-        session.triggertesting([item] * (session.MAXITEMSPERHOST*2 + 1))
+        monkeypatch.setattr(session, 'ITEM_CHUNKSIZE', 3)
-        sent1 = node1.sent[0]
+        session.senditems_load([item] * (2*session.ITEM_CHUNKSIZE +1))
-        sent2 = node2.sent[0]
+        sent1 = node1.sent
-        assert sent1 == [item] * session.MAXITEMSPERHOST
+        sent2 = node2.sent
-        assert sent2 == [item] * session.MAXITEMSPERHOST
+        chunkitems = [item] * session.ITEM_CHUNKSIZE
+        assert sent1 == chunkitems
+        assert sent2 == chunkitems
         assert session.node2pending[node1] == sent1
         assert session.node2pending[node2] == sent2
         name, args, kwargs = session.queue.get(block=False)
@@ -134,7 +141,7 @@ class TestDSession:
         session.loop_once(loopstate)
         session.queueevent(None)
         session.loop_once(loopstate)
-        assert node.sent == [[item]]
+        assert node.sent == [item]
         session.queueevent("pytest_runtest_logreport", report=run(item, node))
         session.loop_once(loopstate)
         assert loopstate.shuttingdown 
@@ -195,38 +202,6 @@ class TestDSession:
         session.loop_once(loopstate)
         assert len(session.item2nodes[item1]) == 1
 
-    def test_testnodedown_causes_reschedule_pending(self, testdir):
-        modcol = testdir.getmodulecol("""
-            def test_crash(): 
-                assert 0
-            def test_fail(): 
-                x
-        """)
-        item1, item2 = modcol.collect()
-
-        # setup a session with two nodes
-        session = DSession(item1.config)
-        node1, node2 = MockNode(), MockNode()
-        session.addnode(node1)
-        session.addnode(node2)
-      
-        # have one test pending for a node that goes down 
-        session.senditems_load([item1, item2])
-        node = session.item2nodes[item1] [0]
-        item1.config.option.dist = "load"
-        session.queueevent("pytest_testnodedown", node=node, error="xyz")
-        reprec = testdir.getreportrecorder(session)
-        print(session.item2nodes)
-        loopstate = session._initloopstate([])
-        session.loop_once(loopstate)
-
-        assert loopstate.colitems == [item2] # do not reschedule crash item
-        rep = reprec.matchreport(names="pytest_runtest_logreport")
-        assert rep.failed
-        assert rep.item == item1
-        assert str(rep.longrepr).find("crashed") != -1
-        #assert str(testrep.longrepr).find(node.gateway.spec) != -1
-
     def test_testnodeready_adds_to_available(self, testdir):
         item = testdir.getitem("def test_func(): pass")
         # setup a session with two nodes
@@ -248,7 +223,7 @@ class TestDSession:
         session.queueevent(None)
         session.loop_once(loopstate)
 
-        assert node.sent == [[item]]
+        assert node.sent == [item]
         ev = run(item, node, excinfo=excinfo) 
         session.queueevent("pytest_runtest_logreport", report=ev)
         session.loop_once(loopstate)
@@ -502,4 +477,19 @@ def test_funcarg_teardown_failure(testdir):
         "*1 passed*1 error*",
     ])
 
+def test_crashing_item(testdir):
+    p = testdir.makepyfile("""
+        import os
+        def test_crash():
+            os.kill(os.getpid(), 15)
+        def test_noncrash():
+            pass
+    """)
+    result = testdir.runpytest("-n2", p)
+    result.stdout.fnmatch_lines([
+        "*crashed*test_crash*",
+        "*1 failed*1 passed*"
+    ])
+
+
     

testing/test_pickling.py

@@ -64,122 +64,6 @@ class TestImmutablePickling:
         assert modback is modcol1
 
 
-class TestConfigPickling:
-    def test_config_getstate_setstate(self, testdir):
-        from py._test.config import Config
-        testdir.makepyfile(__init__="", conftest="x=1; y=2")
-        hello = testdir.makepyfile(hello="")
-        tmp = testdir.tmpdir
-        testdir.chdir()
-        config1 = testdir.parseconfig(hello)
-        config2 = Config()
-        config2.__setstate__(config1.__getstate__())
-        assert config2.topdir == py.path.local()
-        config2_relpaths = [py.path.local(x).relto(config2.topdir) 
-                                for x in config2.args]
-        config1_relpaths = [py.path.local(x).relto(config1.topdir) 
-                                for x in config1.args]
-
-        assert config2_relpaths == config1_relpaths
-        for name, value in config1.option.__dict__.items():
-            assert getattr(config2.option, name) == value
-        assert config2.getvalue("x") == 1
-
-    def test_config_pickling_customoption(self, testdir):
-        testdir.makeconftest("""
-            def pytest_addoption(parser):
-                group = parser.getgroup("testing group")
-                group.addoption('-G', '--glong', action="store", default=42, 
-                    type="int", dest="gdest", help="g value.")
-        """)
-        config = testdir.parseconfig("-G", "11")
-        assert config.option.gdest == 11
-        repr = config.__getstate__()
-
-        config = testdir.Config()
-        py.test.raises(AttributeError, "config.option.gdest")
-
-        config2 = testdir.Config()
-        config2.__setstate__(repr) 
-        assert config2.option.gdest == 11
-
-    def test_config_pickling_and_conftest_deprecated(self, testdir):
-        tmp = testdir.tmpdir.ensure("w1", "w2", dir=1)
-        tmp.ensure("__init__.py")
-        tmp.join("conftest.py").write(py.code.Source("""
-            def pytest_addoption(parser):
-                group = parser.getgroup("testing group")
-                group.addoption('-G', '--glong', action="store", default=42, 
-                    type="int", dest="gdest", help="g value.")
-        """))
-        config = testdir.parseconfig(tmp, "-G", "11")
-        assert config.option.gdest == 11
-        repr = config.__getstate__()
-
-        config = testdir.Config()
-        py.test.raises(AttributeError, "config.option.gdest")
-
-        config2 = testdir.Config()
-        config2.__setstate__(repr) 
-        assert config2.option.gdest == 11
-       
-        option = config2.addoptions("testing group", 
-                config2.Option('-G', '--glong', action="store", default=42,
-                       type="int", dest="gdest", help="g value."))
-        assert option.gdest == 11
-
-    def test_config_picklability(self, testdir):
-        config = testdir.parseconfig()
-        s = pickle.dumps(config)
-        newconfig = pickle.loads(s)
-        assert hasattr(newconfig, "topdir")
-        assert newconfig.topdir == py.path.local()
-
-    def test_collector_implicit_config_pickling(self, testdir):
-        tmpdir = testdir.tmpdir
-        testdir.chdir()
-        testdir.makepyfile(hello="def test_x(): pass")
-        config = testdir.parseconfig(tmpdir)
-        col = config.getnode(config.topdir)
-        io = py.io.BytesIO()
-        pickler = pickle.Pickler(io)
-        pickler.dump(col)
-        io.seek(0) 
-        unpickler = pickle.Unpickler(io)
-        col2 = unpickler.load()
-        assert col2.name == col.name 
-        assert col2.listnames() == col.listnames()
-
-    def test_config_and_collector_pickling(self, testdir):
-        tmpdir = testdir.tmpdir
-        dir1 = tmpdir.ensure("somedir", dir=1)
-        config = testdir.parseconfig()
-        col = config.getnode(config.topdir)
-        col1 = col.join(dir1.basename)
-        assert col1.parent is col 
-        io = py.io.BytesIO()
-        pickler = pickle.Pickler(io)
-        pickler.dump(col)
-        pickler.dump(col1)
-        pickler.dump(col)
-        io.seek(0) 
-        unpickler = pickle.Unpickler(io)
-        topdir = tmpdir.ensure("newtopdir", dir=1)
-        topdir.ensure("somedir", dir=1)
-        old = topdir.chdir()
-        try:
-            newcol = unpickler.load()
-            newcol2 = unpickler.load()
-            newcol3 = unpickler.load()
-            assert newcol2.config is newcol.config
-            assert newcol2.parent == newcol 
-            assert newcol2.config.topdir.realpath() == topdir.realpath()
-            assert newcol.fspath.realpath() == topdir.realpath()
-            assert newcol2.fspath.basename == dir1.basename
-            assert newcol2.fspath.relto(newcol2.config.topdir)
-        finally:
-            old.chdir() 
-
 def test_config__setstate__wired_correctly_in_childprocess(testdir):
     execnet = py.test.importorskip("execnet")
     from xdist.mypickle import PickleChannel

xdist/dsession.py

@@ -54,7 +54,7 @@ class LoopState(object):
                     self.colitems.extend(pending[1:])
 
     def pytest_rescheduleitems(self, items):
-        self.colitems.extend(items)
+        self.colitems[:] = items + self.colitems
         self.dowork = False # avoid busywait
 
 class DSession(Session):
@@ -62,7 +62,8 @@ class DSession(Session):
         Session drives the collection and running of tests
         and generates test events for reporters. 
     """ 
-    MAXITEMSPERHOST = 15
+    LOAD_THRESHOLD_NEWITEMS = 5
+    ITEM_CHUNKSIZE = 10
 
     def __init__(self, config):
         self.queue = queue.Queue()
@@ -73,11 +74,20 @@ class DSession(Session):
     def main(self, colitems):
         self.sessionstarts()
         self.setup()
-        exitstatus = self.loop(colitems)
+        allitems = self.collect_all_items(colitems)
+        self.nodemanager.wait_nodesready(5.0)
+        #for x in allitems:
+        #    print x.listnames()
+        exitstatus = self.loop(allitems)
         self.teardown()
         self.sessionfinishes(exitstatus=exitstatus) 
         return exitstatus
 
+    def collect_all_items(self, colitems):
+        allitems = list(self.collect(colitems))
+        print ("collected %d items" %(len(allitems)))
+        return allitems
+
     def loop_once(self, loopstate):
         if loopstate.shuttingdown:
             return self.loop_once_shutdown(loopstate)
@@ -180,15 +190,10 @@ class DSession(Session):
         return pending
 
     def triggertesting(self, colitems):
-        colitems = self.filteritems(colitems)
+        # for now we don't allow sending collectors 
-        senditems = []
         for next in colitems:
-            if isinstance(next, py.test.collect.Item):
+            assert isinstance(next, py.test.collect.Item), next
-                senditems.append(next)
+        senditems = list(colitems)
-            else:
-                self.config.hook.pytest_collectstart(collector=next)
-                colrep = self.config.hook.pytest_make_collect_report(collector=next)
-                self.queueevent("pytest_collectreport", report=colrep)
         if self.config.option.dist == "each":
             self.senditems_each(senditems)
         else:
@@ -201,42 +206,36 @@ class DSession(Session):
     def senditems_each(self, tosend):
         if not tosend:
             return 
-        room = self.MAXITEMSPERHOST
         for node, pending in self.node2pending.items():
-            room = min(self.MAXITEMSPERHOST - len(pending), room)
+            node.sendlist(tosend)
-        sending = tosend[:room]
+            pending.extend(tosend)
-        if sending:
+            for item in tosend:
-            for node, pending in self.node2pending.items():
-                node.sendlist(sending)
-                pending.extend(sending)
-                for item in sending:
                 nodes = self.item2nodes.setdefault(item, [])
                 assert node not in nodes
                 nodes.append(node)
                 item.ihook.pytest_itemstart(item=item, node=node)
-            tosend[:] = tosend[room:]  # update inplace
+        tosend[:] = []
-        if tosend:
-            # we have some left, give it to the main loop
-            self.queueevent("pytest_rescheduleitems", items=tosend)
 
     def senditems_load(self, tosend):
         if not tosend:
             return 
+        available = []
         for node, pending in self.node2pending.items():
-            room = self.MAXITEMSPERHOST - len(pending)
+            if len(pending) < self.LOAD_THRESHOLD_NEWITEMS:
-            if room > 0:
+                available.append((node, pending))
-                sending = tosend[:room]
+        num_available = len(available)
-                node.sendlist(sending)
+        max_one_round = num_available * self.ITEM_CHUNKSIZE -1
-                for item in sending:
+        if num_available:
-                    #assert item not in self.item2node, (
+            for i, item in enumerate(tosend):
-                    #    "sending same item %r to multiple "
+                nodeindex = i % num_available
-                    #    "not implemented" %(item,))
+                node, pending = available[nodeindex]
+                node.send(item)
                 self.item2nodes.setdefault(item, []).append(node)
                 item.ihook.pytest_itemstart(item=item, node=node)
-                pending.extend(sending)
+                pending.append(item)
-                tosend[:] = tosend[room:]  # update inplace
+                if i >= max_one_round:
-                if not tosend:
                     break
+            del tosend[:i+1]
         if tosend:
             # we have some left, give it to the main loop
             self.queueevent("pytest_rescheduleitems", items=tosend)
@@ -263,8 +262,6 @@ class DSession(Session):
         """ setup any neccessary resources ahead of the test run. """
         self.nodemanager = NodeManager(self.config)
         self.nodemanager.setup_nodes(putevent=self.queue.put)
-        if self.config.option.dist == "each":
-            self.nodemanager.wait_nodesready(5.0)
 
     def teardown(self):
         """ teardown any resources after a test run. """ 

xdist/plugin.py

@@ -3,8 +3,8 @@
 The `pytest-xdist`_ plugin extends py.test with some unique 
 test execution modes:
 
-* Looponfail: run your tests in a subprocess.  After it finishes py.test
+* Looponfail: run your tests repeatedly in a subprocess.  After each run py.test
-  waits until a file in your project changes and then re-runs only the 
+  waits until a file in your project changes and then re-runs the previously
   failing tests.  This is repeated until all tests pass after which again
   a full run is performed. 
 

xdist/remote.py

@@ -63,12 +63,21 @@ class RemoteControl(object):
         self.trace("setting up slave session")
         self.gateway = self.initgateway()
         self.channel = channel = self.gateway.remote_exec("""
-            import os
+            import os, sys
             import py
             chdir = channel.receive()
             outchannel = channel.gateway.newchannel()
             channel.send(outchannel)
+            # prune sys.path to not contain relative paths
+            newpaths = []
+            for p in sys.path:
+                if p:
+                    if not os.path.isabs(p):
+                        p = os.path.abspath(p)
+                    newpaths.append(p)
+            sys.path[:] = newpaths     
             os.chdir(chdir) # unpickling config uses cwd as topdir
+            
             config_state = channel.receive()
             fullwidth, hasmarkup = channel.receive()
             py.test.config.__setstate__(config_state)
@@ -126,7 +135,10 @@ def slave_runsession(channel, config, fullwidth, hasmarkup):
     #config.option.session = None
     config.option.looponfail = False 
     config.option.usepdb = False 
+    try:
         trails = channel.receive()
+    except KeyboardInterrupt:
+        return # in the slave we can't do much about this
     config.pluginmanager.do_configure(config)
     DEBUG("SLAVE: initsession()")
     session = config.initsession()

xdist/txnode.py

@@ -152,7 +152,7 @@ class SlaveNode(object):
             self.sendevent("slavefinished")
 
     def run_single(self, item):
-        call = self.runner.CallInfo(item._checkcollectable, when='setup')
+        call = self.runner.CallInfo(item._reraiseunpicklingproblem, when='setup')
         if call.excinfo:
             # likely it is not collectable here because of
             # platform/import-dependency induced skips