openstack/deb-python-testtools
Code Issues Proposed changes

Fix handling of uncatchable exceptions.

Browse Source 
Fix a long-standing bug where tearDown and cleanUps would not be
called if the test run was interrupted. This should fix leaking
external resources from interrupted tests.
(Robert Collins, #1364188)

Fix a long-standing bug where calling sys.exit(0) from within a test
would cause the test suite to exit with 0, without reporting a failure
of that test. We still allow the test suite to be exited (since
catching higher order exceptions requires exceptional circumstances)
but we now call a last-resort handler on the TestCase, resulting in an
error being reported for the test.
(Robert Collins, #1364188)

Change-Id: I0700f33fe7ed01416b37c21eb3f3fd0a7ea917eb
This commit is contained in:
Robert Collins
2014-09-02 14:16:33 +12:00
parent b04e714e6a
commit 18bc5741cf
8 changed files with 132 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
NEWS
View File

@@ -10,6 +10,18 @@ NEXT
Improvements
------------
* Fix a long-standing bug where tearDown and cleanUps would not be called if the
test run was interrupted. This should fix leaking external resources from
interrupted tests.
(Robert Collins, #1364188)
* Fix a long-standing bug where calling sys.exit(0) from within a test would
cause the test suite to exit with 0, without reporting a failure of that
test. We still allow the test suite to be exited (since catching higher order
exceptions requires exceptional circumstances) but we now call a last-resort
handler on the TestCase, resulting in an error being reported for the test.
(Robert Collins, #1364188)
* Fix an issue where tests skipped with the ``skip``* family of decorators would
still have their ``setUp`` and ``tearDown`` functions called.
(Thomi Richards, #https://github.com/testing-cabal/testtools/issues/86)

6
doc/for-framework-folk.rst
View File

@@ -50,9 +50,9 @@ provide a custom ``RunTest`` to a ``TestCase``. The ``RunTest`` object can
change everything about how the test executes.
To work with ``testtools.TestCase``, a ``RunTest`` must have a factory that
takes a test and an optional list of exception handlers. Instances returned
by the factory must have a ``run()`` method that takes an optional ``TestResult``
object.
takes a test and an optional list of exception handlers and an optional
last_resort handler. Instances returned by the factory must have a ``run()``
method that takes an optional ``TestResult`` object.
The default is ``testtools.runtest.RunTest``, which calls ``setUp``, the test
method, ``tearDown`` and clean ups (see :ref:`addCleanup`) in the normal, vanilla

18
testtools/deferredruntest.py
View File

@@ -89,14 +89,21 @@ class AsynchronousDeferredRunTest(_DeferredRunTest):
This is highly experimental code. Use at your own risk.
"""
def __init__(self, case, handlers=None, reactor=None, timeout=0.005,
debug=False):
def __init__(self, case, handlers=None, last_resort=None, reactor=None,
timeout=0.005, debug=False):
"""Construct an `AsynchronousDeferredRunTest`.
Please be sure to always use keyword syntax, not positional, as the
base class may add arguments in future - and for core code
compatibility with that we have to insert them before the local
parameters.
:param case: The `TestCase` to run.
:param handlers: A list of exception handlers (ExceptionType, handler)
where 'handler' is a callable that takes a `TestCase`, a
``testtools.TestResult`` and the exception raised.
:param last_resort: Handler to call before re-raising uncatchable
exceptions (those for which there is no handler).
:param reactor: The Twisted reactor to use. If not given, we use the
default reactor.
:param timeout: The maximum time allowed for running a test. The
@@ -105,7 +112,8 @@ class AsynchronousDeferredRunTest(_DeferredRunTest):
to get information about unhandled Deferreds and left-over
DelayedCalls. Defaults to False.
"""
super(AsynchronousDeferredRunTest, self).__init__(case, handlers)
super(AsynchronousDeferredRunTest, self).__init__(
case, handlers, last_resort)
if reactor is None:
from twisted.internet import reactor
self._reactor = reactor
@@ -119,8 +127,8 @@ class AsynchronousDeferredRunTest(_DeferredRunTest):
# will be able to be assigned to a class variable *and* also be
# invoked directly.
class AsynchronousDeferredRunTestFactory:
def __call__(self, case, handlers=None):
return cls(case, handlers, reactor, timeout, debug)
def __call__(self, case, handlers=None, last_resort=None):
return cls(case, handlers, last_resort, reactor, timeout, debug)
return AsynchronousDeferredRunTestFactory()
@defer.deferredGenerator

23
testtools/runtest.py
View File

@@ -47,17 +47,23 @@ class RunTest(object):
reporting of error/failure/skip etc.
"""
def __init__(self, case, handlers=None):
def __init__(self, case, handlers=None, last_resort=None):
"""Create a RunTest to run a case.
:param case: A testtools.TestCase test case object.
:param handlers: Exception handlers for this RunTest. These are stored
in self.handlers and can be modified later if needed.
:param last_resort: A handler of last resort: any exception which is
not handled by handlers will cause the last resort handler to be
called as last_resort(exc_info), and then the exception will be
raised - aborting the test run as this is inside the runner
machinery rather than the confined context of the test.
"""
self.case = case
self.handlers = handlers or []
self.exception_caught = object()
self._exceptions = []
self.last_resort = last_resort or (lambda case, result, exc: None)
def run(self, result=None):
"""Run self.case reporting activity to result.
@@ -106,6 +112,9 @@ class RunTest(object):
if isinstance(e, exc_class):
handler(self.case, self.result, e)
break
else:
self.last_resort(self.case, self.result, e)
raise e
finally:
result.stopTest(self.case)
return result
@@ -178,8 +187,6 @@ class RunTest(object):
"""
try:
return fn(*args, **kwargs)
except KeyboardInterrupt:
raise
except:
return self._got_user_exception(sys.exc_info())
@@ -204,11 +211,11 @@ class RunTest(object):
self.case.onException(exc_info, tb_label=tb_label)
finally:
del exc_info
for exc_class, handler in self.handlers:
if isinstance(e, exc_class):
self._exceptions.append(e)
return self.exception_caught
raise e
self._exceptions.append(e)
# Yes, this means we catch everything - we re-raise KeyBoardInterrupt
# etc later, after tearDown and cleanUp - since those may be cleaning up
# external processes.
return self.exception_caught
def _raise_force_fail_error():

25
testtools/testcase.py
View File

@@ -126,9 +126,16 @@ def run_test_with(test_runner, **kwargs):
def decorator(function):
# Set an attribute on 'function' which will inform TestCase how to
# make the runner.
function._run_test_with = (
lambda case, handlers=None:
test_runner(case, handlers=handlers, **kwargs))
def _run_test_with(case, handlers=None, last_resort=None):
try:
return test_runner(
case, handlers=handlers, last_resort=last_resort,
**kwargs)
except TypeError:
# Backwards compat: if we can't call the constructor
# with last_resort, try without that.
return test_runner(case, handlers=handlers, **kwargs)
function._run_test_with = _run_test_with
return function
return decorator
@@ -209,7 +216,10 @@ class TestCase(unittest.TestCase):
self.__RunTest = runTest
if getattr(test_method, '__unittest_expecting_failure__', False):
setattr(self, self._testMethodName, _expectedFailure(test_method))
# Used internally for onException processing - used to gather extra
# data from exceptions.
self.__exception_handlers = []
# Passed to RunTest to map exceptions to result actions
self.exception_handlers = [
(self.skipException, self._report_skip),
(self.failureException, self._report_failure),
@@ -582,7 +592,14 @@ class TestCase(unittest.TestCase):
result.addUnexpectedSuccess(self, details=self.getDetails())
def run(self, result=None):
return self.__RunTest(self, self.exception_handlers).run(result)
try:
run_test = self.__RunTest(
self, self.exception_handlers, last_resort=self._report_error)
except TypeError:
# Backwards compat: if we can't call the constructor
# with last_resort, try without that.
run_test = self.__RunTest(self, self.exception_handlers)
return run_test.run(result)
def _run_setup(self, result):
"""Run the setUp function for this test.

12
testtools/tests/test_deferredruntest.py
View File

@@ -53,6 +53,12 @@ class X(object):
self.calls.append('tearDown')
super(X.Base, self).tearDown()
class BaseExceptionRaised(Base):
expected_calls = ['setUp', 'tearDown', 'clean-up']
expected_results = [('addError', SystemExit)]
def test_something(self):
raise SystemExit(0)
class ErrorInSetup(Base):
expected_calls = ['setUp', 'clean-up']
expected_results = [('addError', RuntimeError)]
@@ -103,7 +109,10 @@ class X(object):
def test_runner(self):
result = ExtendedTestResult()
test = self.test_factory('test_something', runTest=self.runner)
test.run(result)
if self.test_factory is X.BaseExceptionRaised:
self.assertRaises(SystemExit, test.run, result)
else:
test.run(result)
self.assertEqual(test.calls, self.test_factory.expected_calls)
self.assertResultsMatch(test, result)
@@ -118,6 +127,7 @@ def make_integration_tests():
]
tests = [
X.BaseExceptionRaised,
X.ErrorInSetup,
X.ErrorInTest,
X.ErrorInTearDown,

58
testtools/tests/test_runtest.py
View File

@@ -34,6 +34,12 @@ class TestRunTest(TestCase):
run = RunTest("bar", handlers)
self.assertEqual(handlers, run.handlers)
def test__init____handlers_last_resort(self):
handlers = [("quux", "baz")]
last_resort = "foo"
run = RunTest("bar", handlers, last_resort)
self.assertEqual(last_resort, run.last_resort)
def test_run_with_result(self):
# test.run passes result down to _run_test_method.
log = []
@@ -61,15 +67,19 @@ class TestRunTest(TestCase):
run.run()
self.assertEqual(['foo'], log)
def test__run_user_does_not_catch_keyboard(self):
case = self.make_case()
def raises():
raise KeyboardInterrupt("yo")
run = RunTest(case, None)
def test__run_prepared_result_does_not_mask_keyboard(self):
class Case(TestCase):
def test(self):
raise KeyboardInterrupt("go")
case = Case('test')
run = RunTest(case)
run.result = ExtendedTestResult()
self.assertThat(lambda: run._run_user(raises),
self.assertThat(lambda: run._run_prepared_result(run.result),
Raises(MatchesException(KeyboardInterrupt)))
self.assertEqual([], run.result._events)
self.assertEqual(
[('startTest', case), ('stopTest', case)], run.result._events)
# tearDown is still run though!
self.assertEqual(True, getattr(case, '_TestCase__teardown_called'))
def test__run_user_calls_onException(self):
case = self.make_case()
@@ -103,21 +113,43 @@ class TestRunTest(TestCase):
self.assertEqual([], run.result._events)
self.assertEqual([], log)
def test__run_user_uncaught_Exception_raised(self):
case = self.make_case()
def test__run_prepared_result_uncaught_Exception_raised(self):
e = KeyError('Yo')
def raises():
raise e
class Case(TestCase):
def test(self):
raise e
case = Case('test')
log = []
def log_exc(self, result, err):
log.append((result, err))
run = RunTest(case, [(ValueError, log_exc)])
run.result = ExtendedTestResult()
self.assertThat(lambda: run._run_user(raises),
self.assertThat(lambda: run._run_prepared_result(run.result),
Raises(MatchesException(KeyError)))
self.assertEqual([], run.result._events)
self.assertEqual(
[('startTest', case), ('stopTest', case)], run.result._events)
self.assertEqual([], log)
def test__run_prepared_result_uncaught_Exception_triggers_error(self):
# https://bugs.launchpad.net/testtools/+bug/1364188
# When something isn't handled, the test that was
# executing has errored, one way or another.
e = SystemExit(0)
class Case(TestCase):
def test(self):
raise e
case = Case('test')
log = []
def log_exc(self, result, err):
log.append((result, err))
run = RunTest(case, [], log_exc)
run.result = ExtendedTestResult()
self.assertThat(lambda: run._run_prepared_result(run.result),
Raises(MatchesException(SystemExit)))
self.assertEqual(
[('startTest', case), ('stopTest', case)], run.result._events)
self.assertEqual([(run.result, e)], log)
def test__run_user_uncaught_Exception_from_exception_handler_raised(self):
case = self.make_case()
def broken_handler(exc_info):

12
testtools/tests/test_testcase.py
View File

@@ -909,6 +909,18 @@ class TestAddCleanup(TestCase):
set(self.logging_result._events[1][2].keys()))
class TestRunTestUsage(TestCase):
def test_last_resort_in_place(self):
class TestBase(TestCase):
def test_base_exception(self):
raise SystemExit(0)
result = ExtendedTestResult()
test = TestBase("test_base_exception")
self.assertRaises(SystemExit, test.run, result)
self.assertFalse(result.wasSuccessful())
class TestWithDetails(TestCase):
run_test_with = FullStackRunTest