From c607baca2ba50c2690a722fdf975485c56c4a769 Mon Sep 17 00:00:00 2001 From: holger krekel Date: Sun, 17 Jan 2010 23:25:03 +0100 Subject: [PATCH] rework dist-testing model: now all items are collected at the start. move back some tests to pytest-core --- testing/test_dsession.py | 82 ++++++++++++--------------- testing/test_pickling.py | 116 --------------------------------------- xdist/dsession.py | 83 ++++++++++++++-------------- xdist/plugin.py | 4 +- xdist/remote.py | 16 +++++- xdist/txnode.py | 2 +- 6 files changed, 93 insertions(+), 210 deletions(-) diff --git a/testing/test_dsession.py b/testing/test_dsession.py index 49d0082..872c14d 100644 --- a/testing/test_dsession.py +++ b/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]) - name, args, kwargs = session.queue.get(block=False) - assert name == 'pytest_collectreport' - report = kwargs['report'] - assert len(report.result) == 1 + reprec = testdir.getreportrecorder(session) + items = session.collect_all_items([modcol]) + assert len(items) == 1 + calls= reprec.getcalls("pytest_collectreport") + 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)) - sent1 = node1.sent[0] - sent2 = node2.sent[0] - assert sent1 == [item] * session.MAXITEMSPERHOST - assert sent2 == [item] * session.MAXITEMSPERHOST + monkeypatch.setattr(session, 'ITEM_CHUNKSIZE', 3) + session.senditems_load([item] * (2*session.ITEM_CHUNKSIZE +1)) + sent1 = node1.sent + sent2 = node2.sent + 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*" + ]) + + diff --git a/testing/test_pickling.py b/testing/test_pickling.py index 9bf49ec..bc52fa3 100644 --- a/testing/test_pickling.py +++ b/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 diff --git a/xdist/dsession.py b/xdist/dsession.py index 5a24c77..46fc0fb 100644 --- a/xdist/dsession.py +++ b/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,8 +62,9 @@ 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() self.node2pending = {} @@ -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) - senditems = [] + # for now we don't allow sending collectors for next in colitems: - if isinstance(next, py.test.collect.Item): - senditems.append(next) - else: - self.config.hook.pytest_collectstart(collector=next) - colrep = self.config.hook.pytest_make_collect_report(collector=next) - self.queueevent("pytest_collectreport", report=colrep) + assert isinstance(next, py.test.collect.Item), next + senditems = list(colitems) 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) - sending = tosend[:room] - if sending: - 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 - if tosend: - # we have some left, give it to the main loop - self.queueevent("pytest_rescheduleitems", items=tosend) + node.sendlist(tosend) + pending.extend(tosend) + for item in tosend: + nodes = self.item2nodes.setdefault(item, []) + assert node not in nodes + nodes.append(node) + item.ihook.pytest_itemstart(item=item, node=node) + tosend[:] = [] def senditems_load(self, tosend): if not tosend: return + available = [] for node, pending in self.node2pending.items(): - room = self.MAXITEMSPERHOST - len(pending) - if room > 0: - sending = tosend[:room] - node.sendlist(sending) - for item in sending: - #assert item not in self.item2node, ( - # "sending same item %r to multiple " - # "not implemented" %(item,)) - self.item2nodes.setdefault(item, []).append(node) - item.ihook.pytest_itemstart(item=item, node=node) - pending.extend(sending) - tosend[:] = tosend[room:] # update inplace - if not tosend: + if len(pending) < self.LOAD_THRESHOLD_NEWITEMS: + available.append((node, pending)) + num_available = len(available) + max_one_round = num_available * self.ITEM_CHUNKSIZE -1 + if num_available: + for i, item in enumerate(tosend): + nodeindex = i % num_available + node, pending = available[nodeindex] + node.send(item) + self.item2nodes.setdefault(item, []).append(node) + item.ihook.pytest_itemstart(item=item, node=node) + pending.append(item) + if i >= max_one_round: 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. """ diff --git a/xdist/plugin.py b/xdist/plugin.py index 73e71c9..dc94057 100644 --- a/xdist/plugin.py +++ b/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 - waits until a file in your project changes and then re-runs only the +* 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 the previously failing tests. This is repeated until all tests pass after which again a full run is performed. diff --git a/xdist/remote.py b/xdist/remote.py index d7fe3f8..7b54f06 100644 --- a/xdist/remote.py +++ b/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 - trails = channel.receive() + 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() diff --git a/xdist/txnode.py b/xdist/txnode.py index afc843f..7a88f2f 100644 --- a/xdist/txnode.py +++ b/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