Don't show stderr output when running test suite

kombu/tests/test_utils.py

@@ -6,6 +6,8 @@ from StringIO import StringIO
 from kombu import utils
 from kombu.utils.functional import wraps
 
+from kombu.tests.utils import redirect_stdouts
+
 partition = utils._compat_partition
 rpartition = utils._compat_rpartition
 
@@ -116,16 +118,19 @@ class MyStringIO(StringIO):
         pass
 
 
-
 class test_emergency_dump_state(unittest.TestCase):
 
-    def test_dump(self):
+    @redirect_stdouts
+    def test_dump(self, stdout, stderr):
         fh = MyStringIO()
 
         utils.emergency_dump_state({"foo": "bar"}, open_file=lambda n, m : fh)
         self.assertDictEqual(pickle.loads(fh.getvalue()), {"foo": "bar"})
+        self.assertTrue(stderr.getvalue())
+        self.assertFalse(stdout.getvalue())
 
-    def test_dump_second_strategy(self):
+    @redirect_stdouts
+    def test_dump_second_strategy(self, stdout, stderr):
         fh = MyStringIO()
 
         def raise_something(*args, **kwargs):
@@ -134,7 +139,8 @@ class test_emergency_dump_state(unittest.TestCase):
         utils.emergency_dump_state({"foo": "bar"}, open_file=lambda n, m: fh,
                                                    dump=raise_something)
         self.assertIn("'foo': 'bar'", fh.getvalue())
-
+        self.assertTrue(stderr.getvalue())
+        self.assertFalse(stdout.getvalue())
 
 
 _tried_to_sleep = [None]

kombu/tests/test_virtual.py

@@ -5,6 +5,7 @@ from kombu.transport import virtual
 from kombu.utils import gen_unique_id
 
 from kombu.tests.mocks import Channel
+from kombu.tests.utils import redirect_stdouts
 
 
 def client():
@@ -41,7 +42,8 @@ class test_QoS(unittest.TestCase):
         self.assertFalse(self.q._delivered.restored)
         self.assertTrue(self.q._on_collect)
 
-    def test_can_consume(self):
+    @redirect_stdouts
+    def test_can_consume(self, stdout, stderr):
 
         _restored = []
         class RestoreChannel(virtual.Channel):
@@ -80,6 +82,9 @@ class test_QoS(unittest.TestCase):
         self.q._delivered.restored = False
         self.q.restore_unacked_once()
 
+        self.assertTrue(stderr.getvalue())
+        self.assertFalse(stdout.getvalue())
+
 
 class test_Message(unittest.TestCase):
 

kombu/tests/utils.py

@@ -0,0 +1,23 @@
+import sys
+
+from StringIO import StringIO
+
+from kombu.utils.functional import wraps
+
+
+def redirect_stdouts(fun):
+
+    @wraps(fun)
+    def _inner(*args, **kwargs):
+        sys.stdout = StringIO()
+        sys.stderr = StringIO()
+        try:
+            return fun(*args, **dict(kwargs, stdout=sys.stdout,
+                                             stderr=sys.stderr))
+        finally:
+            sys.stdout = sys.__stdout__
+            sys.stderr = sys.__stderr__
+
+    return _inner
+
+