Merge "tests: catch eventlet.Timeout exception"

neutron/tests/base.py

@@ -16,8 +16,11 @@
 """Base test cases for all neutron tests.
 """
 
+import abc
 import contextlib
+import functools
 import gc
+import inspect
 import os
 import os.path
 import sys
@@ -115,6 +118,32 @@ class AttributeDict(dict):
         raise AttributeError(_("Unknown attribute '%s'.") % name)
 
 
+def _catch_timeout(f):
+    @functools.wraps(f)
+    def func(self, *args, **kwargs):
+        try:
+            return f(self, *args, **kwargs)
+        except eventlet.timeout.Timeout as e:
+            self.fail('Execution of this test timed out: %s' % e)
+    return func
+
+
+class _CatchTimeoutMetaclass(abc.ABCMeta):
+    def __init__(cls, name, bases, dct):
+        super(_CatchTimeoutMetaclass, cls).__init__(name, bases, dct)
+        for name, method in inspect.getmembers(
+                # NOTE(ihrachys): we should use isroutine because it will catch
+                # both unbound methods (python2) and functions (python3)
+                cls, predicate=inspect.isroutine):
+            if name.startswith('test_'):
+                setattr(cls, name, _catch_timeout(method))
+
+
+# Test worker cannot survive eventlet's Timeout exception, which effectively
+# kills the whole worker, with all test cases scheduled to it. This metaclass
+# makes all test cases convert Timeout exceptions into unittest friendly
+# failure mode (self.fail).
+@six.add_metaclass(_CatchTimeoutMetaclass)
 class DietTestCase(base.BaseTestCase):
     """Same great taste, less filling.
 

neutron/tests/unit/tests/test_base.py

@@ -16,6 +16,8 @@
 """Tests to test the test framework"""
 
 import sys
+
+import eventlet.timeout
 import unittest2
 
 from neutron.tests import base
@@ -70,3 +72,20 @@ class SystemExitTestCase(base.DietTestCase):
         self.assertEqual([], result.errors)
         self.assertItemsEqual(set(id(t) for t in expectedFails),
                               set(id(t) for (t, traceback) in result.failures))
+
+
+class CatchTimeoutTestCase(base.DietTestCase):
+    # Embedded to hide from the regular test discovery
+    class MyTestCase(base.DietTestCase):
+        def test_case(self):
+            raise eventlet.timeout.Timeout()
+
+        def runTest(self):
+            return self.test_case()
+
+    def test_catch_timeout(self):
+        try:
+            result = self.MyTestCase().run()
+            self.assertFalse(result.wasSuccessful())
+        except eventlet.timeout.Timeout:
+            self.fail('Timeout escaped!')