diff --git a/ISSUES.txt b/ISSUES.txt index a658427..c05c782 100644 --- a/ISSUES.txt +++ b/ISSUES.txt @@ -1,3 +1,11 @@ +rename / hooks +----------------------------------------------- +tag: bug + +node -> slave +transition for hooks? +configure_node -> configure_slave + allow to run xdist tests with xdist ----------------------------------------------- diff --git a/setup.py b/setup.py index 6543f84..2e7aba5 100644 --- a/setup.py +++ b/setup.py @@ -22,7 +22,7 @@ setup( packages = ['xdist'], entry_points = {'pytest11': ['xdist = xdist.plugin'],}, zip_safe=False, - install_requires = ['execnet>=1.0.7', 'py>1.3.1'], + install_requires = ['execnet>=1.0.7', 'py>1.3.9'], classifiers=[ 'Development Status :: 5 - Production/Stable', 'Intended Audience :: Developers', diff --git a/testing/acceptance_test.py b/testing/acceptance_test.py index 9110d59..8e9709a 100644 --- a/testing/acceptance_test.py +++ b/testing/acceptance_test.py @@ -2,6 +2,40 @@ import py import sys class TestDistribution: + def test_n1_pass(self, testdir): + p1 = testdir.makepyfile(""" + def test_ok(): + pass + """) + result = testdir.runpytest(p1, "-n1") + assert result.ret == 0 + result.stdout.fnmatch_lines([ + "*1 passed*", + ]) + + def test_n1_fail(self, testdir): + p1 = testdir.makepyfile(""" + def test_fail(): + assert 0 + """) + result = testdir.runpytest(p1, "-n1") + assert result.ret == 1 + result.stdout.fnmatch_lines([ + "*1 failed*", + ]) + + def test_n1_skip(self, testdir): + p1 = testdir.makepyfile(""" + def test_skip(): + import py + py.test.skip("myreason") + """) + result = testdir.runpytest(p1, "-n1") + assert result.ret == 0 + result.stdout.fnmatch_lines([ + "*1 skipped*", + ]) + def test_manytests_to_one_popen(self, testdir): p1 = testdir.makepyfile(""" import py @@ -23,6 +57,20 @@ class TestDistribution: ]) assert result.ret == 1 + def test_n1_fail_minus_x(self, testdir): + p1 = testdir.makepyfile(""" + def test_fail1(): + assert 0 + def test_fail2(): + assert 0 + """) + result = testdir.runpytest(p1, "-x", "-v", "-n1") + assert result.ret == 2 + result.stdout.fnmatch_lines([ + "*Interrupted: stopping*1*", + "*1 failed*", + ]) + def test_dist_conftest_specified(self, testdir): p1 = testdir.makepyfile(""" import py @@ -48,7 +96,7 @@ class TestDistribution: ]) assert result.ret == 1 - @py.test.mark.xfail("sys.platform.startswith('java')") + @py.test.mark.xfail("sys.platform.startswith('java')", run=False) def test_dist_tests_with_crash(self, testdir): if not hasattr(py.std.os, 'kill'): py.test.skip("no os.kill") @@ -70,12 +118,11 @@ class TestDistribution: os.kill(os.getpid(), 15) """ ) - result = testdir.runpytest(p1, "-v", '-d', '--tx=3*popen') + result = testdir.runpytest(p1, "-v", '-d', '-n1') result.stdout.fnmatch_lines([ "*popen*Python*", - "*popen*Python*", - "*popen*Python*", - "*node down*", + "*test_ok*PASS*", + "*node*down*", "*3 failed, 1 passed, 1 skipped*" ]) assert result.ret == 1 @@ -86,8 +133,9 @@ class TestDistribution: subdir = source.mkdir("example_pkg") subdir.ensure("__init__.py") p = subdir.join("test_one.py") - p.write("def test_5(): assert not __file__.startswith(%r)" % str(p)) - result = testdir.runpytest("-v", "-d", "--rsyncdir=%(subdir)s" % locals(), + p.write("def test_5():\n assert not __file__.startswith(%r)" % str(p)) + result = testdir.runpytest("-v", "-d", + "--rsyncdir=%(subdir)s" % locals(), "--tx=popen//chdir=%(dest)s" % locals(), p) assert result.ret == 0 result.stdout.fnmatch_lines([ diff --git a/testing/test_boxed.py b/testing/test_boxed.py index 6988373..d3c6daa 100644 --- a/testing/test_boxed.py +++ b/testing/test_boxed.py @@ -17,11 +17,9 @@ class TestOptionEffects: def test_boxed_option_default(self, testdir): tmpdir = testdir.tmpdir.ensure("subdir", dir=1) config = testdir.reparseconfig() - config.initsession() assert not config.option.boxed py.test.importorskip("execnet") config = testdir.reparseconfig(['-d', tmpdir]) - config.initsession() assert not config.option.boxed def test_is_not_boxed_by_default(self, testdir): diff --git a/testing/test_dsession.py b/testing/test_dsession.py index 37caca2..9dc18b6 100644 --- a/testing/test_dsession.py +++ b/testing/test_dsession.py @@ -1,4 +1,4 @@ -from xdist.dsession import DSession +from xdist.dsession import DSession, LoadScheduling from py._test import session as outcome import py import execnet @@ -16,8 +16,8 @@ class MockNode: def __init__(self): self.sent = [] - def send(self, item): - self.sent.append(item) + def send_runtest(self, nodeid): + self.sent.append(nodeid) def sendlist(self, items): self.sent.extend(items) @@ -29,23 +29,78 @@ def dumpqueue(queue): while queue.qsize(): print(queue.get()) -class TestDSession: - def test_add_remove_node(self, testdir): - item = testdir.getitem("def test_func(): pass") - node = MockNode() - rep = run(item, node) - session = DSession(item.config) - assert not session.node2pending - session.addnode(node) - assert len(session.node2pending) == 1 - session.senditems_load([item]) - pending = session.removenode(node) - assert pending == [item] - assert item not in session.item2nodes - l = session.removenode(node) - assert not l +class TestLoadScheduling: + def test_schedule_load_simple(self): + node1 = MockNode() + node2 = MockNode() + sched = LoadScheduling(2) + sched.addnode(node1) + sched.addnode(node2) + collection = ["a.py::test_1", "a.py::test_2"] + assert not sched.collection_is_completed() + sched.addnode_collection(node1, collection) + assert not sched.collection_is_completed() + sched.addnode_collection(node2, collection) + assert sched.collection_is_completed() + assert sched.node2collection[node1] == collection + assert sched.node2collection[node2] == collection + sched.init_distribute() + assert sched.pending + sched.triggertesting() + assert not sched.tests_finished() + assert node1.sent == collection[:1] + assert node2.sent == collection[1:] + sched.remove_item(node1, collection[0]) + sched.remove_item(node2, collection[1]) + assert sched.tests_finished() + assert not sched.pending - def test_senditems_each_and_receive_with_two_nodes(self, testdir): + def test_triggertesting_chunksize(self): + sched = LoadScheduling(2) + node1 = MockNode() + node2 = MockNode() + sched.addnode(node1) + sched.addnode(node2) + sched.ITEM_CHUNKSIZE = 2 + col = ["xyz"] * (2*sched.ITEM_CHUNKSIZE +1) + sched.addnode_collection(node1, col) + sched.addnode_collection(node2, col) + sched.init_distribute() + sched.triggertesting() + sent1 = node1.sent + sent2 = node2.sent + chunkitems = col[:sched.ITEM_CHUNKSIZE] + assert sent1 == chunkitems + assert sent2 == chunkitems + assert sched.node2pending[node1] == sent1 + assert sched.node2pending[node2] == sent2 + assert len(sched.pending) == 1 + for node in (node1, node2): + for i in range(sched.ITEM_CHUNKSIZE): + sched.remove_item(node, "xyz") + sched.triggertesting() + assert not sched.pending + + def test_add_remove_node(self): + node = MockNode() + sched = LoadScheduling(1) + sched.addnode(node) + collection = ["test_file.py::test_func"] + sched.addnode_collection(node, collection) + assert sched.collection_is_completed() + sched.init_distribute() + sched.triggertesting() + assert not sched.pending + sched.remove_node(node) + assert sched.pending == collection + + +class TestDSession: + + #def test_collection_fails(self, testdir): + # pass + + def xxx_test_senditems_each_and_receive_with_two_nodes(self, testdir): item = testdir.getitem("def test_func(): pass") node1 = MockNode() node2 = MockNode() @@ -63,54 +118,6 @@ class TestDSession: assert not session.node2pending[node1] assert not session.item2nodes - def test_senditems_load_and_receive_one_node(self, testdir): - item = testdir.getitem("def test_func(): pass") - node = MockNode() - rep = run(item, node) - session = DSession(item.config) - session.addnode(node) - session.senditems_load([item]) - assert session.node2pending[node] == [item] - assert session.item2nodes[item] == [node] - session.removeitem(item, node) - assert not session.node2pending[node] - assert not session.item2nodes - - def test_triggertesting_collect(self, testdir): - modcol = testdir.getmodulecol(""" - def test_func(): - pass - """) - session = DSession(modcol.config) - 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_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) - 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) - assert name == "pytest_rescheduleitems" - assert kwargs['items'] == [item] - - def test_keyboardinterrupt(self, testdir): item = testdir.getitem("def test_func(): pass") session = DSession(item.config) @@ -127,35 +134,6 @@ class TestDSession: exitstatus = session.loop([]) assert exitstatus == outcome.EXIT_INTERNALERROR - def test_rescheduleevent(self, testdir): - item = testdir.getitem("def test_func(): pass") - session = DSession(item.config) - node = MockNode() - session.addnode(node) - loopstate = session._initloopstate([]) - session.queueevent("pytest_rescheduleitems", items=[item]) - session.loop_once(loopstate) - # we need to do work because nothing is pending / we would not wake up - assert loopstate.dowork == True - - session.node2pending[node].append(item) - session.queueevent("pytest_rescheduleitems", items=[item]) - session.loop_once(loopstate) - # now we want to not directly trigger work again to avoid busy-wait - assert loopstate.dowork == False - - session.queueevent(None) - session.loop_once(loopstate) - session.queueevent(None) - session.loop_once(loopstate) - assert node.sent == [item, item] - session.queueevent("pytest_runtest_logreport", report=run(item, node)) - session.loop_once(loopstate) - session.queueevent("pytest_runtest_logreport", report=run(item, node)) - session.loop_once(loopstate) - assert loopstate.shuttingdown - assert not loopstate.testsfailed - def test_no_node_remaining_for_tests(self, testdir): item = testdir.getitem("def test_func(): pass") # setup a session with one node diff --git a/testing/test_mypickle.py b/testing/test_mypickle.py deleted file mode 100644 index a6c2860..0000000 --- a/testing/test_mypickle.py +++ /dev/null @@ -1,254 +0,0 @@ - -import py -import sys -import execnet - -Queue = py.builtin._tryimport('queue', 'Queue').Queue - -from xdist.mypickle import ImmutablePickler, PickleChannel -from xdist.mypickle import UnpickleError, makekey -# first let's test some basic functionality - -def pytest_generate_tests(metafunc): - if 'picklemod' in metafunc.funcargnames: - import pickle - metafunc.addcall(funcargs={'picklemod': pickle}) - try: - import cPickle - except ImportError: - pass - else: - metafunc.addcall(funcargs={'picklemod': cPickle}) - elif "obj" in metafunc.funcargnames and "proto" in metafunc.funcargnames: - a1 = A() - a2 = A() - a2.a1 = a1 - for proto in (0,1,2, -1): - for obj in {1:2}, [1,2,3], a1, a2: - metafunc.addcall(funcargs=dict(obj=obj, proto=proto)) - -def test_underlying_basic_pickling_mechanisms(picklemod): - f1 = py.io.BytesIO() - f2 = py.io.BytesIO() - - pickler1 = picklemod.Pickler(f1) - unpickler1 = picklemod.Unpickler(f2) - - pickler2 = picklemod.Pickler(f2) - unpickler2 = picklemod.Unpickler(f1) - - #pickler1.memo = unpickler1.memo = {} - #pickler2.memo = unpickler2.memo = {} - - d = {} - - pickler1.dump(d) - f1.seek(0) - d_other = unpickler2.load() - - # translate unpickler2 memo to pickler2 - pickler2.memo = dict([(id(obj), (int(x), obj)) - for x, obj in unpickler2.memo.items()]) - - pickler2.dump(d_other) - f2.seek(0) - - unpickler1.memo = dict([(makekey(x), y) - for x, y in pickler1.memo.values()]) - d_back = unpickler1.load() - assert d is d_back - - -class A: - pass - - -def test_pickle_and_back_IS_same(obj, proto): - p1 = ImmutablePickler(uneven=False, protocol=proto) - p2 = ImmutablePickler(uneven=True, protocol=proto) - s1 = p1.dumps(obj) - d2 = p2.loads(s1) - s2 = p2.dumps(d2) - obj_back = p1.loads(s2) - assert obj is obj_back - -def test_pickling_twice_before_unpickling(): - p1 = ImmutablePickler(uneven=False) - p2 = ImmutablePickler(uneven=True) - - a1 = A() - a2 = A() - a3 = A() - a3.a1 = a1 - a2.a1 = a1 - s1 = p1.dumps(a1) - a1.a3 = a3 - s2 = p1.dumps(a2) - other_a1 = p2.loads(s1) - other_a2 = p2.loads(s2) - back_a1 = p1.loads(p2.dumps(other_a1)) - other_a3 = p2.loads(p1.dumps(a3)) - back_a3 = p1.loads(p2.dumps(other_a3)) - back_a2 = p1.loads(p2.dumps(other_a2)) - back_a1 = p1.loads(p2.dumps(other_a1)) - assert back_a1 is a1 - assert back_a2 is a2 - -def test_pickling_concurrently(): - p1 = ImmutablePickler(uneven=False) - p2 = ImmutablePickler(uneven=True) - - a1 = A() - a1.hasattr = 42 - a2 = A() - - s1 = p1.dumps(a1) - s2 = p2.dumps(a2) - other_a1 = p2.loads(s1) - other_a2 = p1.loads(s2) - a1_back = p1.loads(p2.dumps(other_a1)) - -def test_self_memoize(): - p1 = ImmutablePickler(uneven=False) - a1 = A() - p1.selfmemoize(a1) - x = p1.loads(p1.dumps(a1)) - assert x is a1 - -TESTTIMEOUT = 2.0 -class TestPickleChannelFunctional: - def setup_class(cls): - cls.gw = execnet.PopenGateway() - cls.gw.remote_exec( - "import py ; py.path.local(%r).pyimport()" %(__file__) - ) - cls.gw.remote_init_threads(5) - # we need the remote test code to import - # the same test module here - - def test_popen_send_instance(self): - channel = self.gw.remote_exec(""" - from xdist.mypickle import PickleChannel - channel = PickleChannel(channel) - from testing.test_mypickle import A - a1 = A() - a1.hello = 10 - channel.send(a1) - a2 = channel.receive() - channel.send(a2 is a1) - """) - channel = PickleChannel(channel) - a_received = channel.receive() - assert isinstance(a_received, A) - assert a_received.hello == 10 - channel.send(a_received) - remote_a2_is_a1 = channel.receive() - assert remote_a2_is_a1 - - def test_send_concurrent(self): - channel = self.gw.remote_exec(""" - from xdist.mypickle import PickleChannel - channel = PickleChannel(channel) - from testing.test_mypickle import A - l = [A() for i in range(10)] - channel.send(l) - other_l = channel.receive() - channel.send((l, other_l)) - channel.send(channel.receive()) - channel.receive() - """) - channel = PickleChannel(channel) - l = [A() for i in range(10)] - channel.send(l) - other_l = channel.receive() - channel.send(other_l) - ret = channel.receive() - assert ret[0] is other_l - assert ret[1] is l - back = channel.receive() - assert other_l is other_l - channel.send(None) - - #s1 = p1.dumps(a1) - #s2 = p2.dumps(a2) - #other_a1 = p2.loads(s1) - #other_a2 = p1.loads(s2) - #a1_back = p1.loads(p2.dumps(other_a1)) - - def test_popen_with_callback(self): - channel = self.gw.remote_exec(""" - from xdist.mypickle import PickleChannel - channel = PickleChannel(channel) - from testing.test_mypickle import A - a1 = A() - a1.hello = 10 - channel.send(a1) - a2 = channel.receive() - channel.send(a2 is a1) - """) - channel = PickleChannel(channel) - queue = Queue() - channel.setcallback(queue.put) - a_received = queue.get(timeout=TESTTIMEOUT) - assert isinstance(a_received, A) - assert a_received.hello == 10 - channel.send(a_received) - #remote_a2_is_a1 = queue.get(timeout=TESTTIMEOUT) - #assert remote_a2_is_a1 - - def test_popen_with_callback_with_endmarker(self): - channel = self.gw.remote_exec(""" - from xdist.mypickle import PickleChannel - channel = PickleChannel(channel) - from testing.test_mypickle import A - a1 = A() - a1.hello = 10 - channel.send(a1) - a2 = channel.receive() - channel.send(a2 is a1) - """) - channel = PickleChannel(channel) - queue = Queue() - channel.setcallback(queue.put, endmarker=-1) - - a_received = queue.get(timeout=TESTTIMEOUT) - assert isinstance(a_received, A) - assert a_received.hello == 10 - channel.send(a_received) - remote_a2_is_a1 = queue.get(timeout=TESTTIMEOUT) - assert remote_a2_is_a1 - endmarker = queue.get(timeout=TESTTIMEOUT) - assert endmarker == -1 - - def test_popen_with_callback_with_endmarker_and_unpickling_error(self): - channel = self.gw.remote_exec(""" - from xdist.mypickle import PickleChannel - channel = PickleChannel(channel) - from testing.test_mypickle import A - a1 = A() - channel.send(a1) - channel.send(a1) - """) - channel = PickleChannel(channel) - queue = Queue() - a = channel.receive() - channel._ipickle._unpicklememo.clear() - channel.setcallback(queue.put, endmarker=-1) - next = queue.get(timeout=TESTTIMEOUT) - assert next == -1 - error = channel._getremoteerror() - assert isinstance(error, UnpickleError) - - def test_popen_with_various_methods(self): - channel = self.gw.remote_exec(""" - from xdist.mypickle import PickleChannel - channel = PickleChannel(channel) - channel.receive() - """) - channel = PickleChannel(channel) - assert not channel.isclosed() - assert not channel._getremoteerror() - channel.send(2) - channel.waitclose(timeout=2) - - diff --git a/testing/test_pickling.py b/testing/test_pickling.py deleted file mode 100644 index 9e6442c..0000000 --- a/testing/test_pickling.py +++ /dev/null @@ -1,82 +0,0 @@ -import py -import pickle - -def setglobals(request): - oldconfig = py.test.config - print("setting py.test.config to None") - py.test.config = None - def resetglobals(): - py.builtin.print_("setting py.test.config to", oldconfig) - py.test.config = oldconfig - request.addfinalizer(resetglobals) - -def pytest_funcarg__testdir(request): - setglobals(request) - return request.getfuncargvalue("testdir") - -class ImmutablePickleTransport: - def __init__(self, request): - from xdist.mypickle import ImmutablePickler - self.p1 = ImmutablePickler(uneven=0) - self.p2 = ImmutablePickler(uneven=1) - setglobals(request) - - def p1_to_p2(self, obj): - return self.p2.loads(self.p1.dumps(obj)) - - def p2_to_p1(self, obj): - return self.p1.loads(self.p2.dumps(obj)) - - def unifyconfig(self, config): - p2config = self.p1_to_p2(config) - p2config._initafterpickle(config.topdir) - return p2config - -pytest_funcarg__pickletransport = ImmutablePickleTransport - -class TestImmutablePickling: - def test_pickle_config(self, testdir, pickletransport): - config1 = testdir.parseconfig() - assert config1.topdir == testdir.tmpdir - testdir.chdir() - p2config = pickletransport.p1_to_p2(config1) - assert p2config.topdir.realpath() == config1.topdir.realpath() - config_back = pickletransport.p2_to_p1(p2config) - assert config_back is config1 - - def test_pickle_modcol(self, testdir, pickletransport): - modcol1 = testdir.getmodulecol("def test_one(): pass") - modcol2a = pickletransport.p1_to_p2(modcol1) - modcol2b = pickletransport.p1_to_p2(modcol1) - assert modcol2a is modcol2b - - modcol1_back = pickletransport.p2_to_p1(modcol2a) - assert modcol1_back - - def test_pickle_func(self, testdir, pickletransport): - modcol1 = testdir.getmodulecol("def test_one(): pass") - item = modcol1.collect_by_name("test_one") - testdir.chdir() - item2a = pickletransport.p1_to_p2(item) - assert item is not item2a # of course - assert item2a.name == item.name - modback = pickletransport.p2_to_p1(item2a.parent) - assert modback is modcol1 - - -def test_config__setstate__wired_correctly_in_childprocess(testdir): - execnet = py.test.importorskip("execnet") - from xdist.mypickle import PickleChannel - gw = execnet.makegateway() - channel = gw.remote_exec(""" - import py - from xdist.mypickle import PickleChannel - channel = PickleChannel(channel) - config = channel.receive() - assert py.test.config == config - """) - channel = PickleChannel(channel) - config = testdir.parseconfig() - channel.send(config) - channel.waitclose() # this will potentially raise - gw.exit() diff --git a/testing/test_plugin.py b/testing/test_plugin.py index ccc8cb1..00bfa3a 100644 --- a/testing/test_plugin.py +++ b/testing/test_plugin.py @@ -4,8 +4,8 @@ import execnet from xdist.nodemanage import NodeManager def test_dist_incompatibility_messages(testdir): - Error = py.test.config.Error - py.test.raises(Error, "testdir.parseconfigure('--pdb', '--looponfail')") + result = testdir.runpytest("--pdb", "--looponfail") + assert result.ret != 0 result = testdir.runpytest("--pdb", "-n", "3") assert result.ret != 0 assert "incompatible" in result.stderr.str() diff --git a/testing/test_remote.py b/testing/test_remote.py new file mode 100644 index 0000000..6743c75 --- /dev/null +++ b/testing/test_remote.py @@ -0,0 +1,198 @@ +import py +from xdist.remote import SlaveController +from xdist.remote import serialize_report, unserialize_report +import execnet +queue = py.builtin._tryimport("queue", "Queue") +from py.builtin import print_ +import marshal + +def check_marshallable(d): + try: + marshal.dumps(d) + except ValueError: + py.std.pprint.pprint(d) + raise ValueError("not marshallable") + +class EventCall: + def __init__(self, eventcall): + self.name, self.kwargs = eventcall + + def __str__(self): + return "" %(self.name, self.kwargs) + +class SlaveSetup: + use_callback = False + + def __init__(self, request): + self.testdir = testdir = request.getfuncargvalue("testdir") + self.request = request + self.events = queue.Queue() + + def setup(self, ): + self.testdir.chdir() + #import os ; os.environ['EXECNET_DEBUG'] = "2" + self.gateway = execnet.makegateway() + self.config = config = self.testdir.parseconfig() + putevent = self.use_callback and self.events.put or None + self.slp = SlaveController(None, self.gateway, config, putevent) + self.request.addfinalizer(self.slp.ensure_teardown) + self.slp.setup() + + def popevent(self, name=None): + while 1: + if self.use_callback: + data = self.events.get(timeout=2) + else: + data = self.slp.channel.receive(timeout=2) + ev = EventCall(data) + if name is None or ev.name == name: + return ev + print("skipping %s" % (ev,)) + + def sendcommand(self, name, **kwargs): + self.slp.sendcommand(name, **kwargs) + +def pytest_funcarg__slave(request): + return SlaveSetup(request) + +def test_remoteinitconfig(testdir): + from xdist.remote import remote_initconfig + config1 = testdir.parseconfig() + config2 = testdir.parseconfig("-x") + cfg = remote_initconfig(config2, config1.option.__dict__, config1.args) + assert cfg == config2 + assert cfg.option.__dict__ == config1.option.__dict__ + +class TestReportSerialization: + def test_itemreport_outcomes(self, testdir): + reprec = testdir.inline_runsource(""" + import py + def test_pass(): pass + def test_fail(): 0/0 + @py.test.mark.skipif("True") + def test_skip(): pass + def test_skip_imperative(): + py.test.skip("hello") + @py.test.mark.xfail("True") + def test_xfail(): 0/0 + def test_xfail_imperative(): + py.test.xfail("hello") + """) + reports = reprec.getreports("pytest_runtest_logreport") + assert len(reports) == 6 + for rep in reports: + d = serialize_report(rep) + check_marshallable(d) + newrep = unserialize_report(d) + assert newrep.passed == rep.passed + assert newrep.failed == rep.failed + assert newrep.skipped == rep.skipped + assert newrep.outcome == rep.outcome + assert newrep.when == rep.when + assert newrep.keywords == rep.keywords + if rep.failed: + assert newrep.longrepr == str(rep.longrepr) + + def test_collectreport_passed(self, testdir): + reprec = testdir.inline_runsource("def test_func(): pass") + reports = reprec.getreports("pytest_collectreport") + for rep in reports: + d = serialize_report(rep) + check_marshallable(d) + newrep = unserialize_report(d) + assert newrep.passed == rep.passed + assert newrep.failed == rep.failed + assert newrep.skipped == rep.skipped + + def test_collectreport_fail(self, testdir): + reprec = testdir.inline_runsource("qwe abc") + reports = reprec.getreports("pytest_collectreport") + assert reports + for rep in reports: + d = serialize_report(rep) + check_marshallable(d) + newrep = unserialize_report(d) + assert newrep.passed == rep.passed + assert newrep.failed == rep.failed + assert newrep.skipped == rep.skipped + if rep.failed: + assert newrep.longrepr == str(rep.longrepr) + +class TestSlaveInteractor: + def test_basic_collect_and_runtests(self, slave): + p = slave.testdir.makepyfile(""" + def test_func(): + pass + """) + slave.setup() + ev = slave.popevent() + assert ev.name == "slaveready" + ev = slave.popevent() + assert ev.name == "collectionstart" + assert not ev.kwargs + ev = slave.popevent("collectionfinish") + assert ev.kwargs['topdir'] == slave.testdir.tmpdir + ids = ev.kwargs['ids'] + assert len(ids) == 1 + slave.sendcommand("runtests", ids=ids) + ev = slave.popevent("testreport") + assert ev.name == "testreport" + rep = unserialize_report(ev.kwargs['data']) + assert rep.nodeid.endswith("::test_func") + assert rep.passed + assert rep.when == "call" + slave.sendcommand("shutdown") + ev = slave.popevent("slavefinished") + assert 'slaveoutput' in ev.kwargs + + def test_remote_collect_skip(self, slave): + p = slave.testdir.makepyfile(""" + import py + py.test.skip("hello") + """) + slave.setup() + ev = slave.popevent("collectionstart") + assert not ev.kwargs + ev = slave.popevent() + assert ev.name == "collectreport" + rep = unserialize_report(ev.kwargs['data']) + assert rep.skipped + ev = slave.popevent("collectionfinish") + print ev.kwargs + assert not ev.kwargs['ids'] + + def test_remote_collect_fail(self, slave): + p = slave.testdir.makepyfile("""aasd qwe""") + slave.setup() + ev = slave.popevent("collectionstart") + assert not ev.kwargs + ev = slave.popevent() + assert ev.name == "collectreport" + rep = unserialize_report(ev.kwargs['data']) + assert rep.failed + ev = slave.popevent("collectionfinish") + print ev.kwargs + assert not ev.kwargs['ids'] + + def test_happy_run_events_converted(self, testdir, slave): + py.test.xfail("implement a simple test for event production") + assert not slave.use_callback + p = slave.testdir.makepyfile(""" + def test_func(): + pass + """) + slave.setup() + hookrec = testdir.getreportrecorder(slave.config) + for data in slave.slp.channel: + slave.slp.process_from_remote(data) + slave.slp.process_from_remote(slave.slp.ENDMARK) + py.std.pprint.pprint(hookrec.hookrecorder.calls) + hookrec.hookrecorder.contains([ + ("pytest_collectstart", "collector.fspath == aaa"), + ("pytest_pycollect_makeitem", "name == 'test_func'"), + ("pytest_collectreport", "report.collector.fspath == aaa"), + ("pytest_collectstart", "collector.fspath == bbb"), + ("pytest_pycollect_makeitem", "name == 'test_func'"), + ("pytest_collectreport", "report.collector.fspath == bbb"), + ]) + diff --git a/xdist/dsession.py b/xdist/dsession.py index a0f8cfc..b2c47bf 100644 --- a/xdist/dsession.py +++ b/xdist/dsession.py @@ -1,81 +1,135 @@ import py -from py._test import session +import sys from xdist.nodemanage import NodeManager +from py._test import session +import kwlog queue = py.builtin._tryimport('queue', 'Queue') -debug_file = None # open('/tmp/loop.log', 'w') -def debug(*args): - if debug_file is not None: - s = " ".join(map(str, args)) - debug_file.write(s+"\n") - debug_file.flush() +def dsession_main(config): + config.pluginmanager.do_configure(config) + session = DSession(config) + trdist = TerminalDistReporter(config) + config.pluginmanager.register(trdist, "terminaldistreporter") + exitcode = session.main() + config.pluginmanager.do_unconfigure(config) + return exitcode -class LoopState(object): - def __init__(self, dsession, colitems): - self.dsession = dsession - self.colitems = colitems - self.exitstatus = None - # loopstate.dowork is False after reschedule events - # because otherwise we might very busily loop - # waiting for a host to become ready. - self.dowork = True - self.shuttingdown = False - self.testsfailed = 0 - - def __repr__(self): - return "" % ( - self.exitstatus, self.shuttingdown, len(self.colitems)) - - def pytest_runtest_logreport(self, report): - if report.item in self.dsession.item2nodes: - if report.when != "teardown": # otherwise we already managed it - self.dsession.removeitem(report.item, report.node) - if report.failed: - self.testsfailed += 1 - - def pytest_collectreport(self, report): - if report.passed: - self.colitems.extend(report.result) - - def pytest_testnodeready(self, node): - self.dsession.addnode(node) - - def pytest_testnodedown(self, node, error=None): - pending = self.dsession.removenode(node) - if pending: - if error: - crashitem = pending[0] - debug("determined crashitem", crashitem) - self.dsession.handle_crashitem(crashitem, node) - # XXX recovery handling for "each"? - # currently pending items are not retried - if self.dsession.config.option.dist == "load": - self.colitems.extend(pending[1:]) - - def pytest_rescheduleitems(self, items): - self.colitems[:] = items + self.colitems - for pending in self.dsession.node2pending.values(): - if pending: - self.dowork = False # avoid busywait, nodes still have work - -class DSession(session.Session): - """ - Session drives the collection and running of tests - and generates test events for reporters. - """ +class LoadScheduling: LOAD_THRESHOLD_NEWITEMS = 5 ITEM_CHUNKSIZE = 10 - def __init__(self, config): - self.queue = queue.Queue() + def __init__(self, numnodes, log=None): + self.numnodes = numnodes self.node2pending = {} + self.node2collection = {} + self.pending = [] + if log is None: + self.log = kwlog.Producer("loadsched") + else: + self.log = log.loadsched + + def hasnodes(self): + return bool(self.node2pending) + + def addnode(self, node): + self.node2pending[node] = [] + + def collection_is_completed(self): + return len(self.node2collection) == self.numnodes + + def tests_finished(self): + if not self.collection_is_completed() or self.pending: + return False + for items in self.node2pending.values(): + if items: + return False + return True + + def addnode_collection(self, node, collection): + assert node in self.node2pending + self.node2collection[node] = list(collection) + + def remove_item(self, node, item): + if item not in self.item2nodes: + raise AssertionError(item, self.item2nodes) + nodes = self.item2nodes[item] + if node in nodes: # the node might have gone down already + nodes.remove(node) + #if not nodes: + # del self.item2nodes[item] + pending = self.node2pending[node] + pending.remove(item) + # pre-load items-to-test if the node may become ready + if self.pending and len(pending) < self.LOAD_THRESHOLD_NEWITEMS: + item = self.pending.pop(0) + pending.append(item) + self.item2nodes.setdefault(item, []).append(node) + node.send_runtest(item) + + def remove_node(self, node): + pending = self.node2pending.pop(node) + # KeyError if we didn't get an addnode() yet + for item in pending: + l = self.item2nodes[item] + l.remove(node) + if not l: + del self.item2nodes[item] + if not pending: + return + crashitem = pending.pop(0) + self.pending.extend(pending) + return crashitem + + def init_distribute(self): + assert self.collection_is_completed() + assert not hasattr(self, 'item2nodes') self.item2nodes = {} - super(DSession, self).__init__(config=config) + # XXX allow nodes to have different collections + col = list(self.node2collection.values())[0] + for node, collection in self.node2collection.items(): + assert collection == col + self.pending = col + + def triggertesting(self): + if not self.pending: + return + available = [] + for node, pending in self.node2pending.items(): + if len(pending) < self.LOAD_THRESHOLD_NEWITEMS: + available.append((node, pending)) + num_available = len(available) + if num_available: + max_one_round = num_available * self.ITEM_CHUNKSIZE -1 + for i, item in enumerate(self.pending): + nodeindex = i % num_available + node, pending = available[nodeindex] + node.send_runtest(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 self.pending[:i+1] + if self.pending: + self.log.debug("triggertesting remaining:", len(self.pending)) + +class Interrupted(KeyboardInterrupt): + """ signals an immediate interruption. """ + +class DSession: + def __init__(self, config): + self.config = config + self.log = kwlog.Producer("dsession") + #kwlog.setconsumer(self.log, kwlog.Path("/tmp/x.log")) + kwlog.setconsumer(self.log, 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 - self._nodesready = py.std.threading.Event() def report_line(self, line): if self.terminal: @@ -90,98 +144,75 @@ class DSession(session.Session): # targets = ", ".join(["[%s]" % gw.id for gw in gateways]) # self.write_line("rsyncfinish: %s -> %s" %(source, targets)) - def main(self, colitems): - self.sessionstarts() + def main(self): + self.config.hook.pytest_sessionstart(session=self) self.setup() - allitems = self.collect_all_items(colitems) - exitstatus = self.loop(allitems) + exitstatus = self.loop() self.teardown() - self.sessionfinishes(exitstatus=exitstatus) + self.config.hook.pytest_sessionfinish(session=self, + exitstatus=exitstatus,) return exitstatus - def collect_all_items(self, colitems): - verbose = self.config.getvalue("verbose") - if verbose: - self.report_line("[master] starting full item collection ...") - allitems = list(self.collect(colitems)) - if verbose: - self.report_line("[master] collected %d items" %(len(allitems))) - return allitems + def slave_slaveready(self, node): + self.sched.addnode(node) + if self.shuttingdown: + node.sendcommand("shutdown") - def loop_once(self, loopstate): - if loopstate.shuttingdown: - return self.loop_once_shutdown(loopstate) - colitems = loopstate.colitems - if self._nodesready.isSet() and loopstate.dowork and colitems: - self.triggertesting(loopstate.colitems) - colitems[:] = [] - # we use a timeout here so that control-C gets through - while 1: - try: - eventcall = self.queue.get(timeout=2.0) - break - except queue.Empty: - continue - loopstate.dowork = True + def slave_slavefinished(self, node): + crashitem = self.sched.remove_node(node) + assert not crashitem, (crashitem, node) + if self.shuttingdown and not self.sched.hasnodes(): + self.session_finished = True - callname, args, kwargs = eventcall - if callname is not None: - call = getattr(self.config.hook, callname) - assert not args - call(**kwargs) + def slave_errordown(self, node, error): + self.report_line("node %r down on error: %s" %(node.gateway.id, error,)) + crashitem = self.sched.remove_node(node) + if crashitem: + self.handle_crashitem(crashitem, node) + #self.report_line("item crashed on node: %s" % crashitem) + if not self.sched.hasnodes(): + self.session_finished = True - # termination conditions - maxfail = self.config.getvalue("maxfail") - if (not self.node2pending or - (loopstate.testsfailed and maxfail and - loopstate.testsfailed >= maxfail) or - (not self.item2nodes and not colitems and not self.queue.qsize())): - if maxfail and loopstate.testsfailed >= maxfail: - raise self.Interrupted("stopping after %d failures" % ( - loopstate.testsfailed)) - self.triggershutdown() - loopstate.shuttingdown = True - if not self.node2pending: - loopstate.exitstatus = session.EXIT_NOHOSTS + def slave_collectionfinish(self, node, ids): + self.sched.addnode_collection(node, ids) + self.report_line("[%s] collected %d test items" %( + node.gateway.id, len(ids))) - def loop_once_shutdown(self, loopstate): - # once we are in shutdown mode we dont send - # events other than HostDown upstream - eventname, args, kwargs = self.queue.get() - if eventname == "pytest_testnodedown": - self.config.hook.pytest_testnodedown(**kwargs) - self.removenode(kwargs['node']) - elif eventname == "pytest_runtest_logreport": - # might be some teardown report - self.config.hook.pytest_runtest_logreport(**kwargs) - elif eventname == "pytest_internalerror": - self.config.hook.pytest_internalerror(**kwargs) - loopstate.exitstatus = session.EXIT_INTERNALERROR - elif eventname == "pytest__teardown_final_logerror": - self.config.hook.pytest__teardown_final_logerror(**kwargs) - loopstate.exitstatus = session.EXIT_TESTSFAILED - if not self.node2pending: - # finished - if loopstate.testsfailed: - loopstate.exitstatus = session.EXIT_TESTSFAILED - else: - loopstate.exitstatus = session.EXIT_OK - #self.config.pluginmanager.unregister(loopstate) + if self.sched.collection_is_completed(): + self.sched.init_distribute() + self.sched.triggertesting() - def _initloopstate(self, colitems): - loopstate = LoopState(self, colitems) - self.config.pluginmanager.register(loopstate) - return loopstate + def slave_logstart(self, node, nodeid, location): + self.config.hook.pytest_runtest_logstart( + nodeid=nodeid, location=location) - def loop(self, colitems): + def slave_testreport(self, node, rep): + self.sched.remove_item(node, rep.nodeid) + #self.report_line("testreport %s: %s" %(rep.id, rep.status)) + self.config.hook.pytest_runtest_logreport(report=rep) + self._handlefailures(rep) + + def slave_collectreport(self, node, rep): + #self.report_line("collectreport %s: %s" %(rep.id, rep.status)) + self._handlefailures(rep) + + def _handlefailures(self, rep): + if rep.failed: + self.countfailures += 1 + if self.maxfail and self.countfailures >= self.maxfail: + self.shouldstop = "stopping after %d failures" % ( + self.countfailures) + + def loop(self): + self.sched = LoadScheduling(len(self.config.option.tx), log=self.log) + self.shouldstop = False + self.session_finished = False + exitstatus = 0 try: - loopstate = self._initloopstate(colitems) - loopstate.dowork = False # first receive at least one HostUp events - while 1: - self.loop_once(loopstate) - if loopstate.exitstatus is not None: - exitstatus = loopstate.exitstatus - break + while not (self.session_finished or self.shouldstop): + self.loop_once() + if self.shouldstop: + raise Interrupted(str(self.shouldstop)) except KeyboardInterrupt: excinfo = py.code.ExceptionInfo() self.config.hook.pytest_keyboard_interrupt(excinfo=excinfo) @@ -189,103 +220,40 @@ class DSession(session.Session): except: self.config.pluginmanager.notify_exception() exitstatus = session.EXIT_INTERNALERROR - self.config.pluginmanager.unregister(loopstate) - if exitstatus == 0 and self._testsfailed: + #self.config.pluginmanager.unregister(loopstate) + if exitstatus == 0 and self.countfailures: exitstatus = session.EXIT_TESTSFAILED return exitstatus + def loop_once(self): + 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.debug("calling method: %s(**%s)" % (method, kwargs)) + call(**kwargs) + if self.sched.tests_finished(): + self.triggershutdown() + def triggershutdown(self): - for node in self.node2pending: + self.shuttingdown = True + for node in self.sched.node2pending: node.shutdown() - def addnode(self, node): - assert node not in self.node2pending - self.node2pending[node] = [] - if (not hasattr(self, 'nodemanager') or - len(self.node2pending) == len(self.nodemanager.gwmanager.group)): - self._nodesready.set() - - def removenode(self, node): - try: - pending = self.node2pending.pop(node) - except KeyError: - # this happens if we didn't receive a testnodeready event yet - return [] - for item in pending: - l = self.item2nodes[item] - l.remove(node) - if not l: - del self.item2nodes[item] - return pending - - def triggertesting(self, colitems): - # for now we don't allow sending collectors - for next in colitems: - assert isinstance(next, py.test.collect.Item), next - senditems = list(colitems) - if self.config.option.dist == "each": - self.senditems_each(senditems) - else: - # XXX assert self.config.option.dist == "load" - self.senditems_load(senditems) - - def queueevent(self, eventname, **kwargs): - self.queue.put((eventname, (), kwargs)) - - def senditems_each(self, tosend): - if not tosend: - return - for node, pending in self.node2pending.items(): - 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(): - 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) - - def removeitem(self, item, node): - if item not in self.item2nodes: - raise AssertionError(item, self.item2nodes) - nodes = self.item2nodes[item] - if node in nodes: # the node might have gone down already - nodes.remove(node) - if not nodes: - del self.item2nodes[item] - pending = self.node2pending[node] - pending.remove(item) - - def handle_crashitem(self, item, node): - runner = item.config.pluginmanager.getplugin("runner") - info = "!!! Node %r crashed during running of test %r" %(node, item) - rep = runner.ItemTestReport(item=item, excinfo=info, when="???") - rep.node = node - item.ihook.pytest_runtest_logreport(report=rep) + def handle_crashitem(self, nodeid, slave): + # XXX get more reporting info by recording pytest_runtest_logstart? + runner = self.config.pluginmanager.getplugin("runner") + fspath = nodeid.split("::")[0] + msg = "Slave %r crashed while running %r" %(slave.gateway.id, nodeid) + rep = runner.TestReport(nodeid, (), fspath, (fspath, None, fspath), (), + "failed", msg, "???") + self.config.hook.pytest_runtest_logreport(report=rep) def setup(self): """ setup any neccessary resources ahead of the test run. """ @@ -298,3 +266,55 @@ class DSession(session.Session): def teardown(self): """ teardown any resources after a test run. """ self.nodemanager.teardown_nodes() + +class TerminalDistReporter: + def __init__(self, config): + self.gateway2info = {} + self.config = config + self.tplugin = config.pluginmanager.getplugin("terminal") + self.tr = config.pluginmanager.getplugin("terminalreporter") + + def write_line(self, msg): + self.tr.write_line(msg) + + def pytest_itemstart(self, __multicall__): + try: + __multicall__.methods.remove(self.tr.pytest_itemstart) + except KeyError: + pass + + def pytest_runtest_logreport(self, report): + if hasattr(report, 'node'): + report.headerlines.append(self.gateway2info.get( + report.node.gateway, + "node %r (platinfo not found? strange)")) + + def pytest_gwmanage_newgateway(self, gateway, platinfo): + #self.write_line("%s instantiated gateway from spec %r" %(gateway.id, gateway.spec._spec)) + d = {} + d['version'] = self.tplugin.repr_pythonversion(platinfo.version_info) + d['id'] = gateway.id + d['spec'] = gateway.spec._spec + d['platform'] = platinfo.platform + if self.config.option.verbose: + d['extra'] = "- " + platinfo.executable + else: + d['extra'] = "" + d['cwd'] = platinfo.cwd + infoline = ("[%(id)s] %(spec)s -- platform %(platform)s, " + "Python %(version)s " + "cwd: %(cwd)s" + "%(extra)s" % d) + if self.config.getvalue("verbose"): + self.write_line(infoline) + self.gateway2info[gateway] = infoline + + def pytest_testnodeready(self, node): + if self.config.getvalue("verbose"): + self.write_line( + "[%s] txnode ready to receive tests" %(node.gateway.id,)) + + def pytest_testnodedown(self, node, error): + if not error: + return + self.write_line("[%s] node down, error: %s" %(node.gateway.id, error)) diff --git a/xdist/looponfail.py b/xdist/looponfail.py index df0dd10..3abd256 100644 --- a/xdist/looponfail.py +++ b/xdist/looponfail.py @@ -184,8 +184,9 @@ class SlaveFailSession: self.config.hook.pytest_cmdline_main(config=self.config) trails, failreports = [], [] for rep in self.recorded_failures: - trails.append(self.collection.getid(rep.getnode())) - loc = rep._getcrashline() + trails.append(rep.nodeid) + loc = rep.longrepr + loc = str(getattr(loc, 'reprcrash', loc)) failreports.append(loc) topdir = str(self.topdir) self.channel.send((topdir, trails, failreports, self.collection_failed)) diff --git a/xdist/mypickle.py b/xdist/mypickle.py deleted file mode 100644 index 4edec91..0000000 --- a/xdist/mypickle.py +++ /dev/null @@ -1,183 +0,0 @@ -""" - - Pickling support for two processes that want to exchange - *immutable* object instances. Immutable in the sense - that the receiving side of an object can modify its - copy but when it sends it back the original sending - side will continue to see its unmodified version - (and no actual state will go over the wire). - - This module also implements an experimental - execnet pickling channel using this idea. - -""" - -import py -import sys, os, struct -#debug = open("log-mypickle-%d" % os.getpid(), 'w') - -if sys.version_info >= (3,0): - makekey = lambda x: x - fromkey = lambda x: x - from pickle import _Pickler as Pickler - from pickle import _Unpickler as Unpickler -else: - makekey = str - fromkey = int - from pickle import Pickler, Unpickler - - -class MyPickler(Pickler): - """ Pickler with a custom memoize() - to take care of unique ID creation. - See the usage in ImmutablePickler - """ - def __init__(self, immo, file, protocol, uneven): - Pickler.__init__(self, file, protocol) - self.uneven = uneven - self._unpicklememo = immo._unpicklememo - self.memo = immo._picklememo - - def memoize(self, obj): - if self.fast: - return - assert id(obj) not in self.memo - memo_len = len(self.memo) - key = memo_len * 2 + self.uneven - self.write(self.put(key)) - self.memo[id(obj)] = key, obj - key = makekey(key) - if key in self._unpicklememo: - assert self._unpicklememo[key] is obj - dict.__setitem__(self._unpicklememo, key, obj) - - #if sys.version_info < (3,0): - # def save_string(self, obj, pack=struct.pack): - # obj = unicode(obj) - # self.save_unicode(obj, pack=pack) - # Pickler.dispatch[str] = save_string - -class UnpicklingDict(dict): - def __init__(self, picklememo): - super(UnpicklingDict, self).__init__() - self._picklememo = picklememo - - def __setitem__(self, key, obj): - super(UnpicklingDict, self).__setitem__(key, obj) - self._picklememo[id(obj)] = (fromkey(key), obj) - -class ImmutablePickler: - def __init__(self, uneven, protocol=0): - """ ImmutablePicklers are instantiated in Pairs. - The two sides need to create unique IDs - while pickling their objects. This is - done by using either even or uneven - numbers, depending on the instantiation - parameter. - """ - self._picklememo = {} - self._unpicklememo = UnpicklingDict(self._picklememo) - self._protocol = protocol - self.uneven = uneven and 1 or 0 - - def selfmemoize(self, obj): - # this is for feeding objects to ourselfes - # which be the case e.g. if you want to pickle - # from a forked process back to the original - f = py.io.BytesIO() - pickler = MyPickler(self, f, self._protocol, uneven=self.uneven) - pickler.memoize(obj) - - def dumps(self, obj): - f = py.io.BytesIO() - pickler = MyPickler(self, f, self._protocol, uneven=self.uneven) - pickler.dump(obj) - #print >>debug, "dumped", obj - #print >>debug, "picklememo", self._picklememo - return f.getvalue() - - def loads(self, string): - f = py.io.BytesIO(string) - unpickler = Unpickler(f) - unpickler.memo = self._unpicklememo - res = unpickler.load() - #print >>debug, "loaded", res - #print >>debug, "unpicklememo", self._unpicklememo - return res - - -NO_ENDMARKER_WANTED = object() - -class UnpickleError(Exception): - """ Problems while unpickling. """ - def __init__(self, formatted): - self.formatted = formatted - Exception.__init__(self, formatted) - def __str__(self): - return self.formatted - -class PickleChannel(object): - """ PickleChannels wrap execnet channels - and allow to send/receive by using - "immutable pickling". - """ - _unpicklingerror = None - def __init__(self, channel): - self._channel = channel - # we use the fact that each side of a - # gateway connection counts with uneven - # or even numbers depending on which - # side it is (for the purpose of creating - # unique ids - which is what we need it here for) - uneven = channel.gateway._channelfactory.count % 2 - self._ipickle = ImmutablePickler(uneven=uneven) - self.RemoteError = channel.RemoteError - - def send(self, obj): - pickled_obj = self._ipickle.dumps(obj) - self._channel.send(pickled_obj) - - def receive(self): - pickled_obj = self._channel.receive() - return self._unpickle(pickled_obj) - - def _unpickle(self, pickled_obj): - if isinstance(pickled_obj, self._channel.__class__): - return pickled_obj - return self._ipickle.loads(pickled_obj) - - def _getremoteerror(self): - return self._unpicklingerror or self._channel._getremoteerror() - - def close(self): - return self._channel.close() - - def isclosed(self): - return self._channel.isclosed() - - def waitclose(self, timeout=None): - return self._channel.waitclose(timeout=timeout) - - def setcallback(self, callback, endmarker=NO_ENDMARKER_WANTED): - if endmarker is NO_ENDMARKER_WANTED: - def unpickle_callback(pickled_obj): - obj = self._unpickle(pickled_obj) - callback(obj) - self._channel.setcallback(unpickle_callback) - return - uniqueendmarker = object() - def unpickle_callback(pickled_obj): - if pickled_obj is uniqueendmarker: - return callback(endmarker) - try: - obj = self._unpickle(pickled_obj) - except KeyboardInterrupt: - raise - except: - excinfo = py.code.ExceptionInfo() - formatted = str(excinfo.getrepr(showlocals=True,funcargs=True)) - self._unpicklingerror = UnpickleError(formatted) - callback(endmarker) - else: - callback(obj) - self._channel.setcallback(unpickle_callback, uniqueendmarker) diff --git a/xdist/newhooks.py b/xdist/newhooks.py index f97a0cc..75a2df4 100644 --- a/xdist/newhooks.py +++ b/xdist/newhooks.py @@ -16,8 +16,3 @@ def pytest_testnodeready(node): def pytest_testnodedown(node, error): """ Test Node is down. """ - -def pytest_rescheduleitems(items): - """ reschedule Items from a node that went down. """ - - diff --git a/xdist/nodemanage.py b/xdist/nodemanage.py index be0f9b7..b7ca99a 100644 --- a/xdist/nodemanage.py +++ b/xdist/nodemanage.py @@ -1,7 +1,8 @@ import py import sys, os import xdist -from xdist.txnode import TXNode +from xdist.remote import SlaveController + from xdist.gwmanage import GatewayManager import execnet @@ -48,18 +49,19 @@ class NodeManager(object): # pick it up as the right topdir # (for other gateways this chdir is irrelevant) self.trace("making gateways") - old = self.config.topdir.chdir() - try: - self.gwmanager.makegateways() - finally: - old.chdir() + #old = self.config.topdir.chdir() + #try: + self.gwmanager.makegateways() + #finally: + # old.chdir() def setup_nodes(self, putevent): self.rsync_roots() self.trace("setting up nodes") for gateway in self.gwmanager.group: - node = TXNode(self, gateway, self.config, putevent) + node = SlaveController(self, gateway, self.config, putevent) gateway.node = node # to keep node alive + node.setup() self.trace("started node %r" % node) def teardown_nodes(self): diff --git a/xdist/plugin.py b/xdist/plugin.py index 6c0e8d2..6dbd72f 100644 --- a/xdist/plugin.py +++ b/xdist/plugin.py @@ -183,7 +183,18 @@ def pytest_addhooks(pluginmanager): # ------------------------------------------------------------------------- # distributed testing initialization # ------------------------------------------------------------------------- -def pytest_configure(config): + +def pytest_cmdline_main(config): + check_options(config) + if config.getvalue("looponfail"): + from xdist.looponfail import looponfail_main + looponfail_main(config) + return 2 # looponfail only can get stop with ctrl-C anyway + elif config.getvalue("dist") != "no": + from xdist.dsession import dsession_main + return dsession_main(config) + +def check_options(config): if config.option.numprocesses: config.option.dist = "load" config.option.tx = ['popen'] * int(config.option.numprocesses) @@ -199,32 +210,6 @@ def pytest_configure(config): if usepdb: raise config.Error("--pdb incompatible with distributing tests.") -def pytest_cmdline_main(config): - if config.getvalue("looponfail"): - from xdist.looponfail import looponfail_main - looponfail_main(config) - return 2 # looponfail only can get stop with ctrl-C anyway - elif config.getvalue("dist"): - pass - return - from py._test.session import Session, Collection - collection = Collection(config) - # instantiate session already because it - # records failures and implements maxfail handling - session = Session(config, collection) - exitstatus = collection.do_collection() - if not exitstatus: - exitstatus = session.main() - return exitstatus - -def pytest_sessionstart(session): - config = session.config - if hasattr(config, '_isdistsession'): - if not config.pluginmanager.hasplugin("terminal") or \ - not config.pluginmanager.hasplugin("terminalreporter"): - return - trdist = TerminalDistReporter(config) - config.pluginmanager.register(trdist, "terminaldistreporter") def pytest_runtest_protocol(item): if item.config.getvalue("boxed"): @@ -238,23 +223,20 @@ def forked_run_report(item): # XXX optionally allow sharing of setup/teardown from py._plugin.pytest_runner import runtestprotocol EXITSTATUS_TESTEXIT = 4 - from xdist.mypickle import ImmutablePickler - ipickle = ImmutablePickler(uneven=0) - ipickle.selfmemoize(item.config) - # XXX workaround the issue that 2.6 cannot pickle - # instances of classes defined in global conftest.py files - ipickle.selfmemoize(item) + import marshal + from xdist.remote import serialize_report, unserialize_report def runforked(): try: reports = runtestprotocol(item, log=False) except KeyboardInterrupt: py.std.os._exit(EXITSTATUS_TESTEXIT) - return ipickle.dumps(reports) + return marshal.dumps([serialize_report(x) for x in reports]) ff = py.process.ForkedFunc(runforked) result = ff.waitfinish() if result.retval is not None: - return ipickle.loads(result.retval) + report_dumps = marshal.loads(result.retval) + return [unserialize_report(x) for x in report_dumps] else: if result.exitstatus == EXITSTATUS_TESTEXIT: py.test.exit("forked test item %s raised Exit" %(item,)) @@ -264,62 +246,8 @@ def report_process_crash(item, result): path, lineno = item._getfslineno() info = "%s:%s: running the test CRASHED with signal %d" %( path, lineno, result.signal) - from py._plugin.pytest_runner import ItemTestReport - return ItemTestReport(item, excinfo=info, when="???") - -class TerminalDistReporter: - def __init__(self, config): - self.gateway2info = {} - self.config = config - self.tplugin = config.pluginmanager.getplugin("terminal") - self.tr = config.pluginmanager.getplugin("terminalreporter") - - def write_line(self, msg): - self.tr.write_line(msg) - - def pytest_itemstart(self, __multicall__): - try: - __multicall__.methods.remove(self.tr.pytest_itemstart) - except KeyError: - pass - - def pytest_runtest_logreport(self, report): - if hasattr(report, 'node'): - report.headerlines.append(self.gateway2info.get( - report.node.gateway, - "node %r (platinfo not found? strange)")) - - def pytest_gwmanage_newgateway(self, gateway, platinfo): - #self.write_line("%s instantiated gateway from spec %r" %(gateway.id, gateway.spec._spec)) - d = {} - d['version'] = self.tplugin.repr_pythonversion(platinfo.version_info) - d['id'] = gateway.id - d['spec'] = gateway.spec._spec - d['platform'] = platinfo.platform - if self.config.option.verbose: - d['extra'] = "- " + platinfo.executable - else: - d['extra'] = "" - d['cwd'] = platinfo.cwd - infoline = ("[%(id)s] %(spec)s -- platform %(platform)s, " - "Python %(version)s " - "cwd: %(cwd)s" - "%(extra)s" % d) - if self.config.getvalue("verbose"): - self.write_line(infoline) - self.gateway2info[gateway] = infoline - - def pytest_testnodeready(self, node): - if self.config.getvalue("verbose"): - self.write_line( - "[%s] txnode ready to receive tests" %(node.gateway.id,)) - - def pytest_testnodedown(self, node, error): - if not error: - return - self.write_line("[%s] node down, error: %s" %(node.gateway.id, error)) - - def pytest_rescheduleitems(self, items): - if self.config.option.debug: - self.write_sep("!", "RESCHEDULING %s " %(items,)) - + from py._plugin import pytest_runner as runner + call = runner.CallInfo(lambda: 0/0, "???") + call.excinfo = info + rep = runner.pytest_runtest_makereport(item, call) + return rep diff --git a/xdist/remote.py b/xdist/remote.py new file mode 100644 index 0000000..421f273 --- /dev/null +++ b/xdist/remote.py @@ -0,0 +1,213 @@ +""" + Implement --dist=* testing +""" + +import py +import sys +import execnet +import kwlog +from py._plugin import pytest_runner as runner # XXX load dynamically + +class SlaveController(object): + ENDMARK = -1 + + def __init__(self, nodemanager, gateway, config, putevent): + #self.nodemanager = nodemanager + self.putevent = putevent + self.gateway = gateway + self.config = config + self.status = None + self._down = False + self.status = "gateway-init" + + def __repr__(self): + return "<%s id=%s status=%s>" %(self.__class__.__name__, + self.gateway.id, self.status) + + def trace(self, *args): + if self.config.option.debug: + msg = " ".join([str(x) for x in args]) + py.builtin.print_("SlaveController:", msg) + + def setup(self): + self.trace("setting up slave session") + assert self.status == "gateway-init" + self.channel = self.gateway.remote_exec(init_slave_session, + args=self.config.args, + option_dict=vars(self.config.option), + ) + self.status = "slave-init" + if self.putevent: + self.channel.setcallback(self.process_from_remote, + endmarker=self.ENDMARK) + + def ensure_teardown(self): + if hasattr(self, 'channel'): + if not self.channel.isclosed(): + self.trace("closing", self.channel) + self.channel.close() + #del self.channel + if hasattr(self, 'gateway'): + self.trace("exiting", self.gateway) + self.gateway.exit() + #del self.gateway + + def send_runtest(self, nodeid): + self.sendcommand("runtests", ids=[nodeid]) + + def shutdown(self): + if not self._down and not self.channel.isclosed(): + self.sendcommand("shutdown") + + def sendcommand(self, name, **kwargs): + """ send a named parametrized command to the other side. """ + self.trace("sending command %s(**%s)" % (name, kwargs)) + self.channel.send((name, kwargs)) + + def notify_inproc(self, eventname, **kwargs): + self.trace("queuing %s(**%s)" % (eventname, kwargs)) + self.putevent((eventname, kwargs)) + + def process_from_remote(self, eventcall): + """ this gets called for each object we receive from + the other side and if the channel closes. + + Note that channel callbacks run in the receiver + thread of execnet gateways - we need to + avoid raising exceptions or doing heavy work. + """ + try: + if eventcall == self.ENDMARK: + err = self.channel._getremoteerror() + if not self._down: + if not err or isinstance(err, EOFError): + err = "Not properly terminated" # lost connection? + self.notify_inproc("errordown", node=self, error=err) + self._down = True + return + eventname, kwargs = eventcall + if eventname in ("collectionstart"): + self.trace("ignoring %s(%s)" %(eventname, kwargs)) + elif eventname == "slaveready": + self.notify_inproc(eventname, node=self) + elif eventname == "slavefinished": + self._down = True + self.slaveoutput = kwargs['slaveoutput'] + self.notify_inproc("slavefinished", node=self) + #elif eventname == "logstart": + # self.notify_inproc(eventname, node=self, **kwargs) + elif eventname in ("testreport", "collectreport"): + rep = unserialize_report(kwargs['data']) + self.notify_inproc(eventname, node=self, rep=rep) + elif eventname == "collectionfinish": + self.notify_inproc(eventname, node=self, ids=kwargs['ids']) + else: + raise ValueError("unknown event: %s" %(eventname,)) + except KeyboardInterrupt: + # should not land in receiver-thread + raise + except: + excinfo = py.code.ExceptionInfo() + py.builtin.print_("!" * 20, excinfo) + self.config.pluginmanager.notify_exception(excinfo) + +def init_slave_session(channel, args, option_dict): + import py + #outchannel = channel.gateway.newchannel() + #sys.stdout = sys.stderr = outchannel.makefile('w') + #channel.send(outchannel) + #fullwidth, hasmarkup = channel.receive() + from xdist.remote import remote_initconfig, SlaveInteractor + config = remote_initconfig(py.test.config, option_dict, args) + interactor = SlaveInteractor(config, channel) + config.hook.pytest_cmdline_main(config=config) + +def remote_initconfig(config, option_dict, args): + config._preparse(args) + config.option.__dict__.update(option_dict) + config.option.looponfail = False + config.option.usepdb = False + config.option.dist = "no" + config.option.distload = False + config.option.numprocesses = None + #kwlog.Producer("slave").DEBUG("option dict", config.option.__dict__) + config.args = args + return config + +class SlaveInteractor: + def __init__(self, config, channel): + self.config = config + self.log = kwlog.Producer("slave") + kwlog.setconsumer(self.log, None) + self.log.info("initializing SlaveInteractor") + self.channel = channel + config.pluginmanager.register(self) + + def sendevent(self, name, **kwargs): + self.log.debug("sending", name, kwargs) + self.channel.send((name, kwargs)) + + def pytest_internalerror(self, excrepr): + for line in str(excrepr).split("\n"): + self.log.debug("IERROR> " + line) + + def pytest_sessionstart(self, session): + self.session = session + self.collection = session.collection + self.sendevent("slaveready") + + def pytest_sessionfinish(self): + self.sendevent("slavefinished", slaveoutput={}) + + def pytest_perform_collection(self, session): + self.sendevent("collectionstart") + + def pytest_runtest_mainloop(self, session): + self.log.debug("entering main loop") + while 1: + name, kwargs = self.channel.receive() + self.log.debug("received command %s(**%s)" % (name, kwargs)) + if name == "runtests": + ids = kwargs['ids'] + for nodeid in ids: + for item in self.collection.getbyid(nodeid): + self.config.hook.pytest_runtest_protocol(item=item) + elif name == "shutdown": + break + return True + + def pytest_log_finishcollection(self, collection): + self.log.debug("pytest_log_finishcollection") + ids = [collection.getid(item) for item in collection.items] + self.sendevent("collectionfinish", + topdir=str(collection.topdir), + ids=ids) + + #def pytest_runtest_logstart(self, nodeid, location): + # self.sendevent("logstart", nodeid=nodeid, location=location) + + def pytest_runtest_logreport(self, report): + data = serialize_report(report) + self.sendevent("testreport", data=data) + + def pytest_collectreport(self, report): + data = serialize_report(report) + self.sendevent("collectreport", data=data) + +def serialize_report(rep): + d = rep.__dict__.copy() + d['longrepr'] = rep.longrepr and str(rep.longrepr) or None + for name in d: + if isinstance(d[name], py.path.local): + d[name] = str(d[name]) + elif name == "result": + d[name] = None # for now + return d + +def unserialize_report(reportdict): + d = reportdict + if 'result' in d: + return runner.CollectReport(**d) + else: + return runner.TestReport(**d) + diff --git a/xdist/txnode.py b/xdist/txnode.py index 09a777b..acceb0c 100644 --- a/xdist/txnode.py +++ b/xdist/txnode.py @@ -2,7 +2,6 @@ Manage setup, running and local representation of remote nodes/processes. """ import py -from xdist.mypickle import PickleChannel from py._test.session import Session class TXNode(object): @@ -19,7 +18,7 @@ class TXNode(object): self.putevent = putevent self.gateway = gateway self.slaveinput = {} - self.channel = install_slave(self) + self.channel = self.setup() self.channel.setcallback(self.callback, endmarker=self.ENDMARK) self._down = False @@ -84,44 +83,47 @@ class TXNode(object): else: self.channel.send(None) -# configuring and setting up slave node -def install_slave(node): - channel = node.gateway.remote_exec(source=""" - import os, sys - sys.path.insert(0, os.getcwd()) - from xdist.mypickle import PickleChannel - from xdist.txnode import SlaveSession - channel.send("basicimport") - channel = PickleChannel(channel) - import py - config, slaveinput, basetemp, nodeid = channel.receive() - config.slaveinput = slaveinput - config.slaveoutput = {} - if basetemp: - config.basetemp = py.path.local(basetemp) - config.nodeid = nodeid - config.pluginmanager.do_configure(config) - session = SlaveSession(config, channel, nodeid) - session.dist_main() - """) - channel.receive() - channel = PickleChannel(channel) - basetemp = None - config = node.config - config.hook.pytest_configure_node(node=node) - if node.gateway.spec.popen: - popenbase = config.ensuretemp("popen") - basetemp = py.path.local.make_numbered_dir(prefix="slave-", - keep=0, rootdir=popenbase) - basetemp = str(basetemp) - channel.send((config, node.slaveinput, basetemp, node.gateway.id)) - return channel + # configuring and setting up slave node + def setup(self): + basetemp = None + config = self.config + config.hook.pytest_configure_node(node=self) + if self.gateway.spec.popen: + popenbase = config.ensuretemp("popen") + basetemp = py.path.local.make_numbered_dir(prefix="slave-", + keep=0, rootdir=popenbase) + basetemp = str(basetemp) + return self.gateway.remote_exec(init_slave_session, + args=self.config.args, + option_dict=vars(self.config.option), + slaveinput={}, # XXX, + basetemp=basetemp, + nodeid=self.gateway.id, + ) -class SlaveSession(Session): - def __init__(self, config, channel, nodeid): +def init_slave_session(channel, args, option_dict, + slaveinput, basetemp, nodeid): + import os, sys + #sys.path.insert(0, os.getcwd()) + from xdist.txnode import SlaveSession + import py + config = py.test.config + config.option.__dict__.update(option_dict) + config._preparse(args) + config.args = args + config.slaveinput = slaveinput + config.slaveoutput = {} + if basetemp: + config.basetemp = py.path.local(basetemp) + config.nodeid = nodeid + return SlaveSession(config, channel).dist_main() + +class SlaveSession: + def __init__(self, config, channel): self.channel = channel - self.nodeid = nodeid - super(SlaveSession, self).__init__(config=config) + self.config = config + self.runner = self.config.pluginmanager.getplugin("pytest_runner") + config.pluginmanager.register(self, "slavesession") def __repr__(self): return "<%s channel=%s>" %(self.__class__.__name__, self.channel) @@ -143,7 +145,6 @@ class SlaveSession(Session): self.sendevent("pytest_internalerror", excrepr=excrepr) def dist_main(self): - self.runner = self.config.pluginmanager.getplugin("pytest_runner") self.sendevent("slaveready") self.main(None) error = getattr(self, '_slaveerror', None)