further tweaks to parallel tests

* skip crash tests, due to undesirable crash reports, suggesting something went wrong
* test_abort: increase wait to improve likelihood of first abort
* test_magic_px: flush&sleep to ensure stdout arrives first
* test_push_tracked: don't test equality of ar.sent/ar._tracker.done, due to race condition

IPython/parallel/tests/test_lbview.py

@@ -20,6 +20,7 @@
 import time
 
 import zmq
+from nose import SkipTest
 
 from IPython import parallel  as pmod
 from IPython.parallel import error
@@ -39,6 +40,7 @@ def setUp(self):
 
     def test_z_crash_task(self):
         """test graceful handling of engine death (balanced)"""
+        raise SkipTest("crash tests disabled, due to undesirable crash reports")
         # self.add_engines(1)
         ar = self.view.apply_async(crash)
         self.assertRaisesRemote(error.EngineError, ar.get, 10)
@@ -60,6 +62,7 @@ def test_abort(self):
         view = self.view
         ar = self.client[:].apply_async(time.sleep, .5)
         ar = self.client[:].apply_async(time.sleep, .5)
+        time.sleep(0.2)
         ar2 = view.apply_async(lambda : 2)
         ar3 = view.apply_async(lambda : 3)
         view.abort(ar2)

IPython/parallel/tests/test_view.py

@@ -22,6 +22,7 @@
 from StringIO import StringIO
 
 import zmq
+from nose import SkipTest
 
 from IPython import parallel  as pmod
 from IPython.parallel import error
@@ -40,6 +41,7 @@ class TestView(ClusterTestCase):
 
     def test_z_crash_mux(self):
         """test graceful handling of engine death (direct)"""
+        raise SkipTest("crash tests disabled, due to undesirable crash reports")
         # self.add_engines(1)
         eid = self.client.ids[-1]
         ar = self.client[eid].apply_async(crash)
@@ -185,8 +187,8 @@ def test_push_tracked(self):
 
         ar = v.push(ns, block=False, track=True)
         self.assertTrue(isinstance(ar._tracker, zmq.MessageTracker))
-        self.assertEquals(ar.sent, ar._tracker.done)
         ar._tracker.wait()
+        self.assertEquals(ar.sent, ar._tracker.done)
         self.assertTrue(ar.sent)
         ar.get()
 
@@ -271,7 +273,7 @@ def test_execute(self):
 
     def test_abort(self):
         view = self.client[-1]
-        ar = view.execute('import time; time.sleep(0.25)', block=False)
+        ar = view.execute('import time; time.sleep(1)', block=False)
         ar2 = view.apply_async(lambda : 2)
         ar3 = view.apply_async(lambda : 3)
         view.abort(ar2)
@@ -316,7 +318,9 @@ def test_magic_px_blocking(self):
         sio = StringIO()
         savestdout = sys.stdout
         sys.stdout = sio
-        ip.magic_px('print a')
+        # just 'print a' worst ~99% of the time, but this ensures that
+        # the stdout message has arrived when the result is finished:
+        ip.magic_px('import sys,time;print a; sys.stdout.flush();time.sleep(0.2)')
         sys.stdout = savestdout
         buf = sio.getvalue()
         self.assertTrue('[stdout:' in buf, buf)