just use the released version fo execnet

is separately tested)

.hgignore

@@ -19,3 +19,4 @@ dist/
 pytest_xdist.egg-info
 issue/
 3rdparty/
+.tox

.hgtags

@@ -0,0 +1,6 @@
+42c6503ee48fae9c4c96d406afb12bfc86f15803 1.0
+eca7ce17eabf296983c36812c8b8be901e7055a3 1.1
+56d8e5280be224a0ad3220a9deed55334710bd23 1.2
+e6c4ce20db4bf65086ff55807a3c306cad7ca393 1.3
+e6c4ce20db4bf65086ff55807a3c306cad7ca393 1.3
+eaf8b1cb7c312883598677231be5bbeea3b5c127 1.3

CHANGELOG

@@ -1,3 +1,46 @@
+1.4
+-------------------------
+
+- perform distributed testing related reporting in the plugin 
+  rather than having dist-related code in the generic py.test 
+  distribution
+
+- depend on execnet-1.0.7 which adds "env1:NAME=value" keys to 
+  gateway specification strings. 
+
+- show detailed gateway setup and platform information only when 
+  "-v" or "--verbose" is specified. 
+
+1.3
+-------------------------
+
+- fix --looponfailing - it would not actually run against the fully changed
+  source tree when initial conftest files load application state. 
+
+- adapt for py-1.3.1's new --maxfailure option 
+
+1.2
+-------------------------
+
+- fix issue79: sessionfinish/teardown hooks are now called systematically 
+  on the slave side 
+- introduce a new data input/output mechanism to allow the master side 
+  to send and receive data from a slave. 
+- fix race condition in underlying pickling/unpickling handling 
+- use and require new register hooks facility of py.test>=1.3.0 
+- require improved execnet>=1.0.6 because of various race conditions 
+  that can arise in xdist testing modes. 
+- fix some python3 related pickling related race conditions 
+- fix PyPI description 
+
+1.1
+-------------------------
+
+- fix an indefinite hang which would wait for events although no events
+  are pending - this happened if items arrive very quickly while 
+  the "reschedule-event" tried unconditionally avoiding a busy-loop
+  and not schedule new work. 
+
 1.0
 -------------------------
 

setup.py

@@ -7,11 +7,12 @@ for the new options.
 """
 
 from setuptools import setup
+from xdist import __version__
 
 setup(
     name="pytest-xdist",
-    version="1.0",
+    version=__version__,
-    description='py.test figleaf coverage plugin',
+    description='py.test xdist plugin for distributed testing and loop-on-failing modes',
     long_description=__doc__,
     license='GPLv2 or later',
     author='holger krekel and contributors',
@@ -21,9 +22,9 @@ setup(
     packages = ['xdist'],
     entry_points = {'pytest11': ['xdist = xdist.plugin'],},
     zip_safe=False,
-    install_requires = ['execnet>=1.0.3', 'py>=1.2.0'],
+    install_requires = ['execnet>=1.0.7', 'py>1.3.1'],
     classifiers=[
-    'Development Status :: 4 - Beta',
+    'Development Status :: 5 - Production/Stable',
     'Intended Audience :: Developers',
     'License :: OSI Approved :: GNU General Public License (GPL)',
     'Operating System :: POSIX',
@@ -33,5 +34,6 @@ setup(
     'Topic :: Software Development :: Quality Assurance',
     'Topic :: Utilities',
     'Programming Language :: Python',
+    'Programming Language :: Python :: 3',
     ],
 )

testing/acceptance_test.py

@@ -1,4 +1,5 @@
 import py
+import sys
 
 class TestDistribution:
     def test_manytests_to_one_popen(self, testdir):
@@ -14,7 +15,7 @@ class TestDistribution:
                     py.test.skip("hello")
             """, 
         )
-        result = testdir.runpytest(p1, '-d', '--tx=popen', '--tx=popen')
+        result = testdir.runpytest(p1, "-v", '-d', '--tx=popen', '--tx=popen')
         result.stdout.fnmatch_lines([
             "*0*popen*Python*",
             "*1*popen*Python*",
@@ -38,7 +39,7 @@ class TestDistribution:
         testdir.makeconftest("""
             option_tx = 'popen popen popen'.split()
         """)
-        result = testdir.runpytest(p1, '-d')
+        result = testdir.runpytest(p1, '-d', "-v")
         result.stdout.fnmatch_lines([
             "*0*popen*Python*",
             "*1*popen*Python*",
@@ -47,6 +48,7 @@ class TestDistribution:
         ])
         assert result.ret == 1
 
+    @py.test.mark.xfail("sys.platform.startswith('java')")
     def test_dist_tests_with_crash(self, testdir):
         if not hasattr(py.std.os, 'kill'):
             py.test.skip("no os.kill")
@@ -68,7 +70,7 @@ class TestDistribution:
                     os.kill(os.getpid(), 15)
             """
         )
-        result = testdir.runpytest(p1, '-d', '--tx=3*popen')
+        result = testdir.runpytest(p1, "-v", '-d', '--tx=3*popen')
         result.stdout.fnmatch_lines([
             "*popen*Python*",
             "*popen*Python*",
@@ -85,7 +87,7 @@ class TestDistribution:
         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("-d", "--rsyncdir=%(subdir)s" % locals(), 
+        result = testdir.runpytest("-v", "-d", "--rsyncdir=%(subdir)s" % locals(), 
             "--tx=popen//chdir=%(dest)s" % locals(), p)
         assert result.ret == 0
         result.stdout.fnmatch_lines([
@@ -110,10 +112,119 @@ class TestDistribution:
                 print("%s...%s" % sys.version_info[:2])
                 assert 0
         """)
-        args = ["--dist=each"]
+        args = ["--dist=each", "-v"]
         args += ["--tx", "popen//python=%s" % interpreters[0]]
         args += ["--tx", "popen//python=%s" % interpreters[1]]
         result = testdir.runpytest(*args)
         s = result.stdout.str()
         assert "2.4" in s
         assert "2.5" in s
+
+    def test_data_exchange(self, testdir):
+        c1 = testdir.makeconftest("""
+            # This hook only called on master.
+            def pytest_configure_node(node):
+                node.slaveinput['a'] = 42
+                node.slaveinput['b'] = 7
+
+            # This hook only takes action on slave.
+            def pytest_configure(config):
+                if hasattr(config, 'slaveinput'):
+                    a = config.slaveinput['a']
+                    b = config.slaveinput['b']
+                    r = a + b
+                    config.slaveoutput['r'] = r
+
+            # This hook only called on master.
+            def pytest_testnodedown(node, error):
+                node.config.calc_result = node.slaveoutput['r']
+
+            # This hook only takes action on master.
+            def pytest_terminal_summary(terminalreporter):
+                if not hasattr(terminalreporter.config, 'slaveinput'):
+                    calc_result = terminalreporter.config.calc_result
+                    terminalreporter._tw.sep('-', 
+                        'calculated result is %s' % calc_result)
+        """)
+        p1 = testdir.makepyfile("def test_func(): pass")
+        result = testdir.runpytest("-v", p1, '-d', '--tx=popen')
+        result.stdout.fnmatch_lines([
+            "*popen*Python*",
+            "*calculated result is 49*",
+            "*1 passed*"
+        ])
+        assert result.ret == 0
+
+    def test_keyboardinterrupt_hooks_issue79(self, testdir):
+        testdir.makepyfile(__init__="", test_one="""
+            def test_hello():
+                raise KeyboardInterrupt()
+        """)
+        testdir.makeconftest("""
+            def pytest_sessionfinish(session):
+                if hasattr(session.config, 'slaveoutput'):
+                    session.config.slaveoutput['s2'] = 42
+            def pytest_testnodedown(node, error):
+                assert node.slaveoutput['s2'] == 42
+                print ("s2call-finished")
+        """)
+        args = ["-n1"]
+        result = testdir.runpytest(*args)
+        s = result.stdout.str()
+        assert result.ret 
+        assert 'SIGINT' in s
+        assert 's2call' in s
+
+    def test_keyboard_interrupt_dist(self, testdir):
+        # xxx could be refined to check for return code 
+        p = testdir.makepyfile("""
+            def test_sleep():
+                import time
+                time.sleep(10)
+        """)
+        child = testdir.spawn_pytest("-n1")
+        child.expect(".*test session starts.*")
+        child.kill(2) # keyboard interrupt
+        child.expect(".*KeyboardInterrupt.*")
+        #child.expect(".*seconds.*")
+        child.close()
+        #assert ret == 2 
+
+class TestTerminalReporting:
+    def test_pass_skip_fail(self, testdir):
+        p = testdir.makepyfile("""
+            import py
+            def test_ok():
+                pass
+            def test_skip():
+                py.test.skip("xx")
+            def test_func():
+                assert 0
+        """)
+        result = testdir.runpytest("-n1", "-v")
+        expected = [
+            "*PASS*test_pass_skip_fail.py:2: *test_ok*", 
+            "*SKIP*test_pass_skip_fail.py:4: *test_skip*", 
+            "*FAIL*test_pass_skip_fail.py:6: *test_func*", 
+        ]
+        for line in expected:
+            result.stdout.fnmatch_lines([line])
+        result.stdout.fnmatch_lines([
+            "    def test_func():",
+            ">       assert 0",
+            "E       assert 0",
+        ])
+
+    def test_fail_platinfo(self, testdir):
+        p = testdir.makepyfile("""
+            def test_func():
+                assert 0
+        """)
+        result = testdir.runpytest("-n1", "-v")
+        result.stdout.fnmatch_lines([
+            "*FAIL*test_fail_platinfo.py:1: *test_func*", 
+            "*popen*Python*",
+            "    def test_func():",
+            ">       assert 0",
+            "E       assert 0",
+        ])

testing/conftest.py

@@ -2,7 +2,6 @@ import py
 import execnet
 
 pytest_plugins = "pytester"
-option_report = 'skipped'
     
 #rsyncdirs = ['.', '../xdist', py.path.local(execnet.__file__).dirpath()]
 

testing/test_boxed.py

@@ -8,7 +8,7 @@ def test_functional_boxed(testdir):
             os.kill(os.getpid(), 15)
     """)
     result = testdir.runpytest(p1, "--boxed")
-    assert result.stdout.fnmatch_lines([
+    result.stdout.fnmatch_lines([
         "*CRASHED*",
         "*1 failed*"
     ])

testing/test_deprecated.py

@@ -1,6 +1,7 @@
 import py
 
-def test_dist_conftest_options(testdir):
+@py.test.mark.xfail("sys.version_info[:2] == (2,7)")
+def test_dist_conftest_options(testdir, recwarn):
     p1 = testdir.tmpdir.ensure("dir", 'p1.py')
     p1.dirpath("__init__.py").write("")
     p1.dirpath("conftest.py").write(py.code.Source("""

testing/test_dsession.py

@@ -1,5 +1,5 @@
 from xdist.dsession import DSession
-from py._test import outcome
+from py._test import session as outcome
 import py
 import execnet
 
@@ -110,6 +110,7 @@ class TestDSession:
         assert name == "pytest_rescheduleitems"
         assert kwargs['items'] == [item]
 
+    
     def test_keyboardinterrupt(self, testdir):
         item = testdir.getitem("def test_func(): pass")
         session = DSession(item.config)
@@ -134,14 +135,22 @@ class TestDSession:
         loopstate = session._initloopstate([])
         session.queueevent("pytest_rescheduleitems", items=[item])
         session.loop_once(loopstate)
-        # check that RescheduleEvents are not immediately
+        # we need to do work because nothing is pending / we would not wake up
-        # rescheduled if there are no nodes
+        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]
+        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 
@@ -250,7 +259,7 @@ class TestDSession:
             def test_pass(): 
                 pass
         """)
-        modcol.config.option.exitfirst = True
+        modcol.config.option.maxfail = 1
         session = DSession(modcol.config)
         node = MockNode()
         session.addnode(node)
@@ -262,12 +271,44 @@ class TestDSession:
         # run tests ourselves and produce reports 
         ev1 = run(items[0], node, "fail")
         ev2 = run(items[1], node, None)
+        session.queueevent("pytest_runtest_logreport", report=ev1) 
+        session.queueevent("pytest_runtest_logreport", report=ev2)
+        # now call the loop
+        loopstate = session._initloopstate(items)
+        py.test.raises(session.Interrupted, "session.loop_once(loopstate)")
+        assert loopstate.testsfailed
+        #assert loopstate.shuttingdown
+
+    def test_maxfail(self, testdir):
+        modcol = testdir.getmodulecol("""
+            def test_fail1(): 
+                assert 0
+            def test_fail2(): 
+                assert 0
+            def test_pass(): 
+                pass
+        """)
+        modcol.config.option.maxfail = 2
+        session = DSession(modcol.config)
+        node = MockNode()
+        session.addnode(node)
+        items = modcol.config.hook.pytest_make_collect_report(collector=modcol).result
+
+        # trigger testing  - this sends tests to the node
+        session.triggertesting(items)
+
+        # run tests ourselves and produce reports 
+        ev1 = run(items[0], node, "fail")
+        ev2 = run(items[1], node, "fail")
         session.queueevent("pytest_runtest_logreport", report=ev1) # a failing one
         session.queueevent("pytest_runtest_logreport", report=ev2)
         # now call the loop
         loopstate = session._initloopstate(items)
-        from xdist.dsession import ExitFirstInterrupt
+        try:
-        py.test.raises(ExitFirstInterrupt, "session.loop_once(loopstate)")
+            session.loop_once(loopstate)
+        except session.Interrupted:
+            py.test.fail("raised Interrupted but shouildn't")
+        py.test.raises(session.Interrupted, "session.loop_once(loopstate)")
         assert loopstate.testsfailed
         #assert loopstate.shuttingdown
 

testing/test_gwmanage.py

@@ -3,6 +3,7 @@ import os
 from xdist.gwmanage import GatewayManager, HostRSync
 from py._test.pluginmanager import HookRelay, Registry
 from py._plugin import hookspec
+from xdist import newhooks
 import execnet
 
 def pytest_funcarg__hookrecorder(request):
@@ -11,7 +12,7 @@ def pytest_funcarg__hookrecorder(request):
     return _pytest.gethookrecorder(hook)
 
 def pytest_funcarg__hook(request):
-    return HookRelay(hookspec, Registry())
+    return HookRelay([hookspec, newhooks], Registry())
 
 class TestGatewayManagerPopen:
     def test_popen_no_default_chdir(self, hook):

testing/test_remote.py

@@ -61,7 +61,6 @@ class TestLooponFailing:
         """)
         session = LooponfailingSession(modcol.config)
         loopstate = LoopState()
-        session.remotecontrol.setup()
         session.loop_once(loopstate)
         assert len(loopstate.colitems) == 1
  
@@ -83,7 +82,6 @@ class TestLooponFailing:
         """)
         session = LooponfailingSession(modcol.config)
         loopstate = LoopState()
-        session.remotecontrol.setup()
         loopstate.colitems = []
         session.loop_once(loopstate)
         assert len(loopstate.colitems) == 1
@@ -110,7 +108,6 @@ class TestLooponFailing:
         """)
         session = LooponfailingSession(modcol.config)
         loopstate = LoopState()
-        session.remotecontrol.setup()
         loopstate.colitems = []
         session.loop_once(loopstate)
         assert len(loopstate.colitems) == 2
@@ -135,7 +132,7 @@ class TestLooponFailing:
                 x = 0
                 assert x == 1
         """)
-        child = testdir.spawn_pytest("-f %s" % p)
+        child = testdir.spawn_pytest("-f %s --traceconfig" % p)
         child.expect("def test_one")
         child.expect("x == 1")
         child.expect("1 failed")
@@ -149,3 +146,16 @@ class TestLooponFailing:
         child.expect(".*1 passed.*")
         child.kill(15)
 
+    def test_looponfail_xfail_passes(self, testdir):
+        p = testdir.makepyfile("""
+            import py
+            @py.test.mark.xfail
+            def test_one():
+                pass
+        """)
+        child = testdir.spawn_pytest("-f %s" % p)
+        child.expect("1 xpass")
+        child.expect("### LOOPONFAILING ####")
+        child.expect("waiting for changes")
+        child.kill(15)
+

testing/test_txnode.py

@@ -12,7 +12,7 @@ class EventQueue:
         self.queue = queue
         registry.register(self)
 
-    def geteventargs(self, eventname, timeout=2.0):
+    def geteventargs(self, eventname, timeout=10.0):
         events = []
         while 1:
             try:
@@ -38,21 +38,22 @@ class MySetup:
         self.id = 0
         self.request = request
 
-    def geteventargs(self, eventname, timeout=2.0):
+    def geteventargs(self, eventname, timeout=10.0):
         eq = EventQueue(self.config.pluginmanager, self.queue)
         return eq.geteventargs(eventname, timeout=timeout)
 
-    def makenode(self, config=None):
+    def makenode(self, config=None, xspec="popen"):
         if config is None:
             testdir = self.request.getfuncargvalue("testdir")
             config = testdir.reparseconfig([])
         self.config = config
         self.queue = Queue()
-        self.xspec = execnet.XSpec("popen")
+        self.xspec = execnet.XSpec(xspec)
         self.gateway = execnet.makegateway(self.xspec)
         self.id += 1
         self.gateway.id = str(self.id)
-        self.node = TXNode(self.gateway, self.config, putevent=self.queue.put)
+        self.nodemanager = None
+        self.node = TXNode(self.nodemanager, self.gateway, self.config, putevent=self.queue.put)
         assert not self.node.channel.isclosed()
         return self.node 
 
@@ -80,7 +81,7 @@ class TestMasterSlaveConnection:
         node.send(123) # invalid item 
         kwargs = mysetup.geteventargs("pytest_testnodedown")
         assert kwargs['node'] is node 
-        assert isinstance(kwargs['error'], execnet.RemoteError)
+        #assert isinstance(kwargs['error'], execnet.RemoteError)
 
     def test_crash_killed(self, testdir, mysetup):
         if not hasattr(py.std.os, 'kill'):
@@ -146,3 +147,26 @@ class TestMasterSlaveConnection:
         for outcome in "passed failed skipped".split():
             rep = mysetup.geteventargs("pytest_runtest_logreport")['report']
             assert getattr(rep, outcome) 
+
+    def test_send_one_with_env(self, testdir, mysetup, monkeypatch):
+        if execnet.XSpec("popen").env is None:
+            py.test.skip("requires execnet 1.0.7 or above")
+        monkeypatch.delenv('ENV1', raising=False)
+        monkeypatch.delenv('ENV2', raising=False)
+        monkeypatch.setenv('ENV3', 'var3')
+
+        item = testdir.getitem("""
+            def test_func():
+                import os
+                # ENV1, ENV2 set by xspec; ENV3 inherited from parent process
+                assert os.getenv('ENV2') == 'var2'
+                assert os.getenv('ENV1') == 'var1'
+                assert os.getenv('ENV3') == 'var3'
+        """)
+        node = mysetup.makenode(item.config,
+                xspec="popen//env:ENV1=var1//env:ENV2=var2")
+        node.send(item)
+        kwargs = mysetup.geteventargs("pytest_runtest_logreport")
+        rep = kwargs['report']
+        assert rep.passed
+

tox.ini

@@ -0,0 +1,35 @@
+[tox]
+distshare={homedir}/.tox/distshare
+envlist=py26,py31,py27,py25,py24
+[tox:hudson]
+distshare={toxworkdir}/distshare
+sdistsrc={distshare}/pytest-xdist-*
+
+[testenv]
+changedir=testing
+deps= 
+    {distshare}/py-*
+commands=
+    py.test -rsfxX --tools-on-path \
+            --junitxml={envlogdir}/junit-{envname}.xml []
+[testenv:py27]
+basepython=python2.7
+[testenv:py26]
+basepython=python2.6
+deps=
+    {distshare}/py-*
+    pexpect
+#[testenv:py26-py132]
+#basepython=python2.6
+#deps= py==1.3.2
+#      execnet==1.0.6
+[testenv:py25]
+basepython=python2.5
+[testenv:py24]
+basepython=python2.4
+[testenv:py31]
+basepython=python3.1
+#[testenv:pypy]
+#basepython=pypy-c
+[testenv:jython]
+basepython=jython

xdist/__init__.py

@@ -1,3 +1,3 @@
 #
-__version__ = "1.0"
+__version__ = "1.4"
 

xdist/dsession.py

@@ -1,6 +1,5 @@
 import py
-from py._test.session import Session
+from py._test import session 
-from py._test import outcome 
 from xdist.nodemanage import NodeManager
 queue = py.builtin._tryimport('queue', 'Queue')
 
@@ -21,7 +20,7 @@ class LoopState(object):
         # waiting for a host to become ready.  
         self.dowork = True
         self.shuttingdown = False
-        self.testsfailed = False
+        self.testsfailed = 0
 
     def __repr__(self):
         return "<LoopState exitstatus=%r shuttingdown=%r len(colitems)=%d>" % (
@@ -32,7 +31,7 @@ class LoopState(object):
             if report.when != "teardown": # otherwise we already managed it
                 self.dsession.removeitem(report.item, report.node)
         if report.failed:
-            self.testsfailed = True
+            self.testsfailed += 1
 
     def pytest_collectreport(self, report):
         if report.passed:
@@ -55,12 +54,11 @@ class LoopState(object):
 
     def pytest_rescheduleitems(self, items):
         self.colitems[:] = items + self.colitems
-        self.dowork = False # avoid busywait
+        for pending in self.dsession.node2pending.values():
+            if pending:
+                self.dowork = False # avoid busywait, nodes still have work
 
-class ExitFirstInterrupt(KeyboardInterrupt):
+class DSession(session.Session):
-    pass
-
-class DSession(Session):
     """ 
         Session drives the collection and running of tests
         and generates test events for reporters. 
@@ -102,9 +100,12 @@ class DSession(Session):
         return exitstatus
 
     def collect_all_items(self, colitems):
-        self.report_line("[master] starting full item collection ...")
+        verbose = self.config.getvalue("verbose")
+        if verbose:
+            self.report_line("[master] starting full item collection ...")
         allitems = list(self.collect(colitems))
-        self.report_line("[master] collected %d items" %(len(allitems)))
+        if verbose:
+            self.report_line("[master] collected %d items" %(len(allitems)))
         return allitems
 
     def loop_once(self, loopstate):
@@ -130,14 +131,18 @@ class DSession(Session):
             call(**kwargs)
 
         # termination conditions
-        if ((loopstate.testsfailed and self.config.option.exitfirst) or 
+        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 self.config.option.exitfirst:
+            if maxfail and loopstate.testsfailed >= maxfail:
-                raise ExitFirstInterrupt()
+                raise self.Interrupted("stopping after %d failures" % (
+                    loopstate.testsfailed))
             self.triggershutdown()
             loopstate.shuttingdown = True
-        elif not self.node2pending:
+            if not self.node2pending:
-            loopstate.exitstatus = outcome.EXIT_NOHOSTS
+                loopstate.exitstatus = session.EXIT_NOHOSTS
                 
     def loop_once_shutdown(self, loopstate):
         # once we are in shutdown mode we dont send 
@@ -151,16 +156,16 @@ class DSession(Session):
             self.config.hook.pytest_runtest_logreport(**kwargs)
         elif eventname == "pytest_internalerror":
             self.config.hook.pytest_internalerror(**kwargs)
-            loopstate.exitstatus = outcome.EXIT_INTERNALERROR
+            loopstate.exitstatus = session.EXIT_INTERNALERROR
         elif eventname == "pytest__teardown_final_logerror":
             self.config.hook.pytest__teardown_final_logerror(**kwargs)
-            loopstate.exitstatus = outcome.EXIT_TESTSFAILED
+            loopstate.exitstatus = session.EXIT_TESTSFAILED
         if not self.node2pending:
             # finished
             if loopstate.testsfailed:
-                loopstate.exitstatus = outcome.EXIT_TESTSFAILED
+                loopstate.exitstatus = session.EXIT_TESTSFAILED
             else:
-                loopstate.exitstatus = outcome.EXIT_OK
+                loopstate.exitstatus = session.EXIT_OK
         #self.config.pluginmanager.unregister(loopstate)
 
     def _initloopstate(self, colitems):
@@ -179,17 +184,14 @@ class DSession(Session):
                     break 
         except KeyboardInterrupt:
             excinfo = py.code.ExceptionInfo()
-            if excinfo.errisinstance(ExitFirstInterrupt):
+            self.config.hook.pytest_keyboard_interrupt(excinfo=excinfo)
-                exitstatus = outcome.EXIT_TESTSFAILED
+            exitstatus = session.EXIT_INTERRUPTED
-            else:
-                self.config.hook.pytest_keyboard_interrupt(excinfo=excinfo)
-                exitstatus = outcome.EXIT_INTERRUPTED
         except:
             self.config.pluginmanager.notify_exception()
-            exitstatus = outcome.EXIT_INTERNALERROR
+            exitstatus = session.EXIT_INTERNALERROR
         self.config.pluginmanager.unregister(loopstate)
         if exitstatus == 0 and self._testsfailed:
-            exitstatus = outcome.EXIT_TESTSFAILED
+            exitstatus = session.EXIT_TESTSFAILED
         return exitstatus
 
     def triggershutdown(self):
@@ -287,6 +289,9 @@ class DSession(Session):
 
     def setup(self):
         """ setup any neccessary resources ahead of the test run. """
+        if not self.config.getvalue("verbose"):
+            self.report_line("instantiating gateways (use -v for details): %s" % 
+                ",".join(self.config.option.tx))
         self.nodemanager = NodeManager(self.config)
         self.nodemanager.setup_nodes(putevent=self.queue.put)
 

xdist/mypickle.py

@@ -31,13 +31,12 @@ class MyPickler(Pickler):
     """ Pickler with a custom memoize()
         to take care of unique ID creation. 
         See the usage in ImmutablePickler
-        XXX we could probably extend Pickler 
-            and Unpickler classes to directly
-            update the other'S memos. 
     """
-    def __init__(self, file, protocol, uneven):
+    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:
@@ -47,6 +46,10 @@ class MyPickler(Pickler):
         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):
@@ -54,6 +57,15 @@ class MyPickler(Pickler):
     #        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. 
@@ -64,7 +76,7 @@ class ImmutablePickler:
             parameter.
         """
         self._picklememo = {}
-        self._unpicklememo = {}
+        self._unpicklememo = UnpicklingDict(self._picklememo)
         self._protocol = protocol
         self.uneven = uneven and 1 or 0
 
@@ -73,18 +85,13 @@ class ImmutablePickler:
         # 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(f, self._protocol, uneven=self.uneven)
+        pickler = MyPickler(self, f, self._protocol, uneven=self.uneven)
-        pickler.memo = self._picklememo
         pickler.memoize(obj)
-        self._updateunpicklememo()
 
     def dumps(self, obj):
         f = py.io.BytesIO()
-        pickler = MyPickler(f, self._protocol, uneven=self.uneven)
+        pickler = MyPickler(self, f, self._protocol, uneven=self.uneven)
-        pickler.memo = self._picklememo
         pickler.dump(obj)
-        if obj is not None:
-            self._updateunpicklememo()
         #print >>debug, "dumped", obj 
         #print >>debug, "picklememo", self._picklememo
         return f.getvalue()
@@ -94,21 +101,10 @@ class ImmutablePickler:
         unpickler = Unpickler(f)
         unpickler.memo = self._unpicklememo
         res = unpickler.load()
-        self._updatepicklememo()
         #print >>debug, "loaded", res
         #print >>debug, "unpicklememo", self._unpicklememo
         return res
 
-    def _updatepicklememo(self):
-        for x, obj in self._unpicklememo.items():
-            self._picklememo[id(obj)] = (fromkey(x), obj)
-
-    def _updateunpicklememo(self):
-        for key,obj in self._picklememo.values():
-            key = makekey(key) 
-            if key in self._unpicklememo:
-                assert self._unpicklememo[key] is obj
-            self._unpicklememo[key] = obj
 
 NO_ENDMARKER_WANTED = object()
 

xdist/newhooks.py

@@ -0,0 +1,23 @@
+
+def pytest_gwmanage_newgateway(gateway, platinfo):
+    """ called on new raw gateway creation. """ 
+
+def pytest_gwmanage_rsyncstart(source, gateways):
+    """ called before rsyncing a directory to remote gateways takes place. """
+
+def pytest_gwmanage_rsyncfinish(source, gateways):
+    """ called after rsyncing a directory to remote gateways takes place. """
+
+def pytest_configure_node(node):
+    """ configure node information before it gets instantiated. """
+
+def pytest_testnodeready(node):
+    """ Test Node is ready to operate. """
+
+def pytest_testnodedown(node, error):
+    """ Test Node is down. """
+
+def pytest_rescheduleitems(items):
+    """ reschedule Items from a node that went down. """
+
+

xdist/nodemanage.py

@@ -58,7 +58,7 @@ class NodeManager(object):
         self.rsync_roots()
         self.trace("setting up nodes")
         for gateway in self.gwmanager.group:
-            node = TXNode(gateway, self.config, putevent)
+            node = TXNode(self, gateway, self.config, putevent)
             gateway.node = node  # to keep node alive 
             self.trace("started node %r" % node)
 

xdist/plugin.py

@@ -120,11 +120,11 @@ Specifying test exec environments in a conftest.py
 Instead of specifying command line options, you can 
 put options values in a ``conftest.py`` file like this::
 
-    pytest_option_tx = ['ssh=myhost//python=python2.5', 'popen//python=python2.5']
+    option_tx = ['ssh=myhost//python=python2.5', 'popen//python=python2.5']
-    pytest_option_dist = True
+    option_dist = True
 
-Any commandline ``--tx`` specifictions  will add to the list of available execution
+Any commandline ``--tx`` specifictions  will add to the list of
-environments. 
+available execution environments. 
 
 Specifying "rsync" dirs in a conftest.py
 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
@@ -173,6 +173,16 @@ def pytest_addoption(parser):
     group.addoption('--rsyncdir', action="append", default=[], metavar="dir1", 
            help="add directory for rsyncing to remote tx nodes.")
 
+# -------------------------------------------------------------------------
+# distributed testing hooks
+# -------------------------------------------------------------------------
+def pytest_addhooks(pluginmanager):
+    from xdist import newhooks
+    pluginmanager.addhooks(newhooks)
+
+# -------------------------------------------------------------------------
+# distributed testing initialization
+# -------------------------------------------------------------------------
 def pytest_configure(config):
     if config.option.numprocesses:
         config.option.dist = "load"
@@ -187,11 +197,23 @@ def pytest_configure(config):
                 raise config.Error("--pdb incompatible with --looponfail.")
             from xdist.remote import LooponfailingSession
             config.setsessionclass(LooponfailingSession)
+            config._isdistsession = True
         elif val("dist") != "no":
             if usepdb:
                 raise config.Error("--pdb incompatible with distributing tests.")
             from xdist.dsession import DSession
             config.setsessionclass(DSession)
+            config._isdistsession = True
+
+
+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"):
@@ -234,3 +256,59 @@ def report_process_crash(item, result):
     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,))
+

xdist/remote.py

@@ -24,7 +24,6 @@ class LooponfailingSession(Session):
     def main(self, initialitems):
         try:
             self.loopstate = loopstate = LoopState([])
-            self.remotecontrol.setup()
             while 1:
                 self.loop_once(loopstate)
                 if not loopstate.colitems and loopstate.wasfailing:
@@ -34,10 +33,10 @@ class LooponfailingSession(Session):
             print
 
     def loop_once(self, loopstate):
+        self.remotecontrol.setup()
         colitems = loopstate.colitems
         loopstate.wasfailing = colitems and len(colitems)
         loopstate.colitems = self.remotecontrol.runsession(colitems or ())
-        self.remotecontrol.setup()
 
 class LoopState:
     def __init__(self, colitems=None):
@@ -170,8 +169,21 @@ def slave_runsession(channel, config, fullwidth, hasmarkup):
 
     DEBUG("SLAVE: starting session.main()")
     session.main(colitems)
-    session.config.hook.pytest_looponfailinfo(
+    repr_pytest_looponfailinfo(
         failreports=list(failreports), 
         rootdirs=[config.topdir])
     rootcol = session.config._rootcol
     channel.send([rootcol.totrail(rep.getnode()) for rep in failreports])
+
+
+def repr_pytest_looponfailinfo(failreports, rootdirs):
+    tr = py.io.TerminalWriter()
+    if failreports:
+        tr.sep("#", "LOOPONFAILING", red=True)
+        for report in failreports:
+            loc = report._getcrashline()
+            if loc:
+                tr.line(loc, red=True)
+    tr.sep("#", "waiting for changes")
+    for rootdir in rootdirs:
+        tr.line("### Watching:   %s" %(rootdir,), bold=True)

xdist/txnode.py

@@ -3,7 +3,7 @@
 """
 import py
 from xdist.mypickle import PickleChannel
-from py._test import outcome
+from py._test.session import Session
 
 class TXNode(object):
     """ Represents a Test Execution environment in the controlling process. 
@@ -13,11 +13,13 @@ class TXNode(object):
     """
     ENDMARK = -1
 
-    def __init__(self, gateway, config, putevent):
+    def __init__(self, nodemanager, gateway, config, putevent):
+        self.nodemanager = nodemanager
         self.config = config 
         self.putevent = putevent 
         self.gateway = gateway
-        self.channel = install_slave(gateway, config)
+        self.slaveinput = {}
+        self.channel = install_slave(self)
         self.channel.setcallback(self.callback, endmarker=self.ENDMARK)
         self._down = False
 
@@ -43,7 +45,7 @@ class TXNode(object):
                 err = self.channel._getremoteerror()
                 if not self._down:
                     if not err or isinstance(err, EOFError):
-                        err = "Not properly terminated"
+                        err = "Not properly terminated" # lost connection? 
                     self.notify("pytest_testnodedown", node=self, error=err)
                     self._down = True
                 return
@@ -52,7 +54,9 @@ class TXNode(object):
                 self.notify("pytest_testnodeready", node=self)
             elif eventname == "slavefinished":
                 self._down = True
-                self.notify("pytest_testnodedown", error=None, node=self)
+                self.slaveoutput = kwargs['slaveoutput']
+                error = kwargs['error']
+                self.notify("pytest_testnodedown", error=error, node=self)
             elif eventname in ("pytest_runtest_logreport", 
                                "pytest__teardown_final_logerror"):
                 kwargs['report'].node = self
@@ -80,32 +84,44 @@ class TXNode(object):
         else:
             self.channel.send(None)
 
-# setting up slave code 
+# configuring and setting up slave node 
-def install_slave(gateway, config):
+def install_slave(node):
-    channel = gateway.remote_exec(source="""
+    channel = node.gateway.remote_exec(source="""
         import os, sys 
         sys.path.insert(0, os.getcwd()) 
         from xdist.mypickle import PickleChannel
-        from xdist.txnode import SlaveNode
+        from xdist.txnode import SlaveSession
         channel.send("basicimport")
         channel = PickleChannel(channel)
-        slavenode = SlaveNode(channel)
+        import py
-        slavenode.run()
+        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
-    if gateway.spec.popen:
+    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, basetemp, gateway.id))
+    channel.send((config, node.slaveinput, basetemp, node.gateway.id))
     return channel
 
-class SlaveNode(object):
+class SlaveSession(Session):
-    def __init__(self, channel):
+    def __init__(self, config, channel, nodeid):
         self.channel = channel
+        self.nodeid = nodeid
+        super(SlaveSession, self).__init__(config=config)
 
     def __repr__(self):
         return "<%s channel=%s>" %(self.__class__.__name__, self.channel)
@@ -119,37 +135,31 @@ class SlaveNode(object):
     def pytest__teardown_final_logerror(self, report):
         self.sendevent("pytest__teardown_final_logerror", report=report)
 
-    def run(self):
+    def pytest_keyboard_interrupt(self, excinfo):
-        channel = self.channel
+        self._slaveerror = "SIGINT"
-        self.config, basetemp, self.nodeid = channel.receive()
+
-        if basetemp:
+    def pytest_internalerror(self, excrepr):
-            self.config.basetemp = py.path.local(basetemp)
+        self._slaveerror = "internal-error"
-        self.config.pluginmanager.do_configure(self.config)
+        self.sendevent("pytest_internalerror", excrepr=excrepr)
-        self.config.pluginmanager.register(self)
+
+    def dist_main(self):
         self.runner = self.config.pluginmanager.getplugin("pytest_runner")
         self.sendevent("slaveready")
-        try:
+        self.main(None)
-            self.config.hook.pytest_sessionstart(session=self)
+        error = getattr(self, '_slaveerror', None)
-            while 1:
+        self.sendevent("slavefinished", error=error, 
-                task = channel.receive()
+                       slaveoutput=self.config.slaveoutput)
-                if task is None: 
+        
-                    break
+    def _mainloop(self, colitems):
-                if isinstance(task, list):
+        while 1:
-                    for item in task:
+            task = self.channel.receive()
-                        self.run_single(item=item)
+            if task is None: 
-                else:
+                break
-                    self.run_single(item=task)
+            if isinstance(task, list):
-            self.config.hook.pytest_sessionfinish(
+                for item in task:
-                session=self, 
+                    self.run_single(item=item)
-                exitstatus=outcome.EXIT_OK)
+            else:
-        except KeyboardInterrupt:
+                self.run_single(item=task)
-            raise
-        except:
-            er = py.code.ExceptionInfo().getrepr(funcargs=True, showlocals=True)
-            self.sendevent("pytest_internalerror", excrepr=er)
-            raise
-        else:
-            self.sendevent("slavefinished")
 
     def run_single(self, item):
         call = self.runner.CallInfo(item._reraiseunpicklingproblem, when='setup')