IQ.Lvbs/pytest-xdist
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

adapt to new refactored py baseline.

Browse Source 
remove pickling in distributed testing
refactor internal mechanism enough to work for -n1
* * *
fix handling of crashing items and slave down events
This commit is contained in:
holger krekel
2010-09-27 16:13:54 +02:00
parent e03f6d76b6
commit 2efb059d75
17 changed files with 881 additions and 1010 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
ISSUES.txt
View File

@@ -1,3 +1,11 @@
rename / hooks
-----------------------------------------------
tag: bug
node -> slave
transition for hooks?
configure_node -> configure_slave
allow to run xdist tests with xdist
-----------------------------------------------

2
setup.py
View File

@@ -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',

62
testing/acceptance_test.py
View File

@@ -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([

2
testing/test_boxed.py
View File

@@ -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):

170
testing/test_dsession.py
View File

@@ -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

254
testing/test_mypickle.py
View File

@@ -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)

82
testing/test_pickling.py
View File

@@ -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()

4
testing/test_plugin.py
View File

@@ -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()

198
testing/test_remote.py Normal file
View File

@@ -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 "<EventCall %s(**%s)>" %(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"),
])

492
xdist/dsession.py
View File

@@ -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 "<LoopState exitstatus=%r shuttingdown=%r len(colitems)=%d>" % (
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))

5
xdist/looponfail.py
View File

@@ -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))

183
xdist/mypickle.py
View File

@@ -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)

5
xdist/newhooks.py
View File

@@ -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. """

16
xdist/nodemanage.py
View File

@@ -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):

116
xdist/plugin.py
View File

@@ -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

213
xdist/remote.py Normal file
View File

@@ -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)

79
xdist/txnode.py
View File

@@ -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)