# Copyright (c) 2010-2016 testtools developers. See LICENSE for details. """Tests for the DeferredRunTest single test execution logic.""" import os import signal from extras import try_import from testtools import ( skipIf, TestCase, TestResult, ) from testtools.matchers import ( AfterPreprocessing, Contains, ContainsAll, ContainsDict, EndsWith, Equals, Is, KeysEqual, MatchesDict, MatchesException, MatchesListwise, Not, Raises, ) from testtools.runtest import RunTest from testtools.testresult.doubles import ExtendedTestResult from testtools.tests.helpers import ( AsText, MatchesEvents, ) from ._helpers import NeedsTwistedTestCase DebugTwisted = try_import( 'testtools.twistedsupport._deferreddebug.DebugTwisted') assert_fails_with = try_import('testtools.twistedsupport.assert_fails_with') AsynchronousDeferredRunTest = try_import( 'testtools.twistedsupport.AsynchronousDeferredRunTest') flush_logged_errors = try_import( 'testtools.twistedsupport.flush_logged_errors') SynchronousDeferredRunTest = try_import( 'testtools.twistedsupport.SynchronousDeferredRunTest') defer = try_import('twisted.internet.defer') failure = try_import('twisted.python.failure') log = try_import('twisted.python.log') DelayedCall = try_import('twisted.internet.base.DelayedCall') _get_global_publisher_and_observers = try_import( 'testtools.twistedsupport._runtest._get_global_publisher_and_observers') class X(object): """Tests that we run as part of our tests, nested to avoid discovery.""" # XXX: After testing-cabal/testtools#165 lands, fix up all of these to be # scenario tests for RunTest. class Base(TestCase): def setUp(self): super(X.Base, self).setUp() self.calls = ['setUp'] self.addCleanup(self.calls.append, 'clean-up') def test_something(self): self.calls.append('test') def tearDown(self): 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)] def setUp(self): super(X.ErrorInSetup, self).setUp() raise RuntimeError("Error in setUp") class ErrorInTest(Base): expected_calls = ['setUp', 'tearDown', 'clean-up'] expected_results = [('addError', RuntimeError)] def test_something(self): raise RuntimeError("Error in test") class FailureInTest(Base): expected_calls = ['setUp', 'tearDown', 'clean-up'] expected_results = [('addFailure', AssertionError)] def test_something(self): self.fail("test failed") class ErrorInTearDown(Base): expected_calls = ['setUp', 'test', 'clean-up'] expected_results = [('addError', RuntimeError)] def tearDown(self): raise RuntimeError("Error in tearDown") class ErrorInCleanup(Base): expected_calls = ['setUp', 'test', 'tearDown', 'clean-up'] expected_results = [('addError', ZeroDivisionError)] def test_something(self): self.calls.append('test') self.addCleanup(lambda: 1/0) class ExpectThatFailure(Base): """Calling expectThat with a failing match fails the test.""" expected_calls = ['setUp', 'test', 'tearDown', 'clean-up'] expected_results = [('addFailure', AssertionError)] def test_something(self): self.calls.append('test') self.expectThat(object(), Is(object())) class TestIntegration(NeedsTwistedTestCase): def assertResultsMatch(self, test, result): events = list(result._events) self.assertEqual(('startTest', test), events.pop(0)) for expected_result in test.expected_results: result = events.pop(0) if len(expected_result) == 1: self.assertEqual((expected_result[0], test), result) else: self.assertEqual((expected_result[0], test), result[:2]) error_type = expected_result[1] self.assertIn(error_type.__name__, str(result[2])) self.assertEqual([('stopTest', test)], events) def test_runner(self): result = ExtendedTestResult() test = self.test_factory('test_something', runTest=self.runner) 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) def make_integration_tests(): from unittest import TestSuite from testtools import clone_test_with_new_id runners = [ ('RunTest', RunTest), ('SynchronousDeferredRunTest', SynchronousDeferredRunTest), ('AsynchronousDeferredRunTest', AsynchronousDeferredRunTest), ] tests = [ X.BaseExceptionRaised, X.ErrorInSetup, X.ErrorInTest, X.ErrorInTearDown, X.FailureInTest, X.ErrorInCleanup, X.ExpectThatFailure, ] base_test = X.TestIntegration('test_runner') integration_tests = [] for runner_name, runner in runners: for test in tests: new_test = clone_test_with_new_id( base_test, '%s(%s, %s)' % ( base_test.id(), runner_name, test.__name__)) new_test.test_factory = test new_test.runner = runner integration_tests.append(new_test) return TestSuite(integration_tests) class TestSynchronousDeferredRunTest(NeedsTwistedTestCase): def make_result(self): return ExtendedTestResult() def make_runner(self, test): return SynchronousDeferredRunTest(test, test.exception_handlers) def test_success(self): class SomeCase(TestCase): def test_success(self): return defer.succeed(None) test = SomeCase('test_success') runner = self.make_runner(test) result = self.make_result() runner.run(result) self.assertThat( result._events, Equals([ ('startTest', test), ('addSuccess', test), ('stopTest', test)])) def test_failure(self): class SomeCase(TestCase): def test_failure(self): return defer.maybeDeferred(self.fail, "Egads!") test = SomeCase('test_failure') runner = self.make_runner(test) result = self.make_result() runner.run(result) self.assertThat( [event[:2] for event in result._events], Equals([ ('startTest', test), ('addFailure', test), ('stopTest', test)])) def test_setUp_followed_by_test(self): class SomeCase(TestCase): def setUp(self): super(SomeCase, self).setUp() return defer.succeed(None) def test_failure(self): return defer.maybeDeferred(self.fail, "Egads!") test = SomeCase('test_failure') runner = self.make_runner(test) result = self.make_result() runner.run(result) self.assertThat( [event[:2] for event in result._events], Equals([ ('startTest', test), ('addFailure', test), ('stopTest', test)])) class TestAsynchronousDeferredRunTest(NeedsTwistedTestCase): def make_reactor(self): from twisted.internet import reactor return reactor def make_result(self): return ExtendedTestResult() def make_runner(self, test, timeout=None, suppress_twisted_logging=True, store_twisted_logs=True): if timeout is None: timeout = self.make_timeout() return AsynchronousDeferredRunTest( test, test.exception_handlers, timeout=timeout, suppress_twisted_logging=suppress_twisted_logging, store_twisted_logs=store_twisted_logs, ) def make_timeout(self): return 0.005 def test_setUp_returns_deferred_that_fires_later(self): # setUp can return a Deferred that might fire at any time. # AsynchronousDeferredRunTest will not go on to running the test until # the Deferred returned by setUp actually fires. call_log = [] marker = object() d = defer.Deferred().addCallback(call_log.append) class SomeCase(TestCase): def setUp(self): super(SomeCase, self).setUp() call_log.append('setUp') return d def test_something(self): call_log.append('test') def fire_deferred(): self.assertThat(call_log, Equals(['setUp'])) d.callback(marker) test = SomeCase('test_something') timeout = self.make_timeout() runner = self.make_runner(test, timeout=timeout) result = self.make_result() reactor = self.make_reactor() reactor.callLater(timeout, fire_deferred) runner.run(result) self.assertThat(call_log, Equals(['setUp', marker, 'test'])) def test_calls_setUp_test_tearDown_in_sequence(self): # setUp, the test method and tearDown can all return # Deferreds. AsynchronousDeferredRunTest will make sure that each of # these are run in turn, only going on to the next stage once the # Deferred from the previous stage has fired. call_log = [] a = defer.Deferred() a.addCallback(lambda x: call_log.append('a')) b = defer.Deferred() b.addCallback(lambda x: call_log.append('b')) c = defer.Deferred() c.addCallback(lambda x: call_log.append('c')) class SomeCase(TestCase): def setUp(self): super(SomeCase, self).setUp() call_log.append('setUp') return a def test_success(self): call_log.append('test') return b def tearDown(self): super(SomeCase, self).tearDown() call_log.append('tearDown') return c test = SomeCase('test_success') timeout = self.make_timeout() runner = self.make_runner(test, timeout) result = self.make_result() reactor = self.make_reactor() def fire_a(): self.assertThat(call_log, Equals(['setUp'])) a.callback(None) def fire_b(): self.assertThat(call_log, Equals(['setUp', 'a', 'test'])) b.callback(None) def fire_c(): self.assertThat( call_log, Equals(['setUp', 'a', 'test', 'b', 'tearDown'])) c.callback(None) reactor.callLater(timeout * 0.25, fire_a) reactor.callLater(timeout * 0.5, fire_b) reactor.callLater(timeout * 0.75, fire_c) runner.run(result) self.assertThat( call_log, Equals(['setUp', 'a', 'test', 'b', 'tearDown', 'c'])) def test_async_cleanups(self): # Cleanups added with addCleanup can return # Deferreds. AsynchronousDeferredRunTest will run each of them in # turn. class SomeCase(TestCase): def test_whatever(self): pass test = SomeCase('test_whatever') call_log = [] a = defer.Deferred().addCallback(lambda x: call_log.append('a')) b = defer.Deferred().addCallback(lambda x: call_log.append('b')) c = defer.Deferred().addCallback(lambda x: call_log.append('c')) test.addCleanup(lambda: a) test.addCleanup(lambda: b) test.addCleanup(lambda: c) def fire_a(): self.assertThat(call_log, Equals([])) a.callback(None) def fire_b(): self.assertThat(call_log, Equals(['a'])) b.callback(None) def fire_c(): self.assertThat(call_log, Equals(['a', 'b'])) c.callback(None) timeout = self.make_timeout() reactor = self.make_reactor() reactor.callLater(timeout * 0.25, fire_a) reactor.callLater(timeout * 0.5, fire_b) reactor.callLater(timeout * 0.75, fire_c) runner = self.make_runner(test, timeout) result = self.make_result() runner.run(result) self.assertThat(call_log, Equals(['a', 'b', 'c'])) def test_clean_reactor(self): # If there's cruft left over in the reactor, the test fails. reactor = self.make_reactor() timeout = self.make_timeout() class SomeCase(TestCase): def test_cruft(self): reactor.callLater(timeout * 10.0, lambda: None) test = SomeCase('test_cruft') runner = self.make_runner(test, timeout) result = self.make_result() runner.run(result) self.assertThat( [event[:2] for event in result._events], Equals( [('startTest', test), ('addError', test), ('stopTest', test)])) error = result._events[1][2] self.assertThat(error, KeysEqual('traceback', 'twisted-log')) def test_exports_reactor(self): # The reactor is set as an attribute on the test case. reactor = self.make_reactor() timeout = self.make_timeout() class SomeCase(TestCase): def test_cruft(self): self.assertIs(reactor, self.reactor) test = SomeCase('test_cruft') runner = self.make_runner(test, timeout) result = TestResult() runner.run(result) self.assertEqual([], result.errors) self.assertEqual([], result.failures) def test_unhandled_error_from_deferred(self): # If there's a Deferred with an unhandled error, the test fails. Each # unhandled error is reported with a separate traceback. # We're interested in the behavior when debugging is disabled. When # debugging is enabled, we get more stack traces. self.useFixture(DebugTwisted(False)) class SomeCase(TestCase): def test_cruft(self): # Note we aren't returning the Deferred so that the error will # be unhandled. defer.maybeDeferred(lambda: 1/0) defer.maybeDeferred(lambda: 2/0) test = SomeCase('test_cruft') runner = self.make_runner(test) result = self.make_result() runner.run(result) error = result._events[1][2] result._events[1] = ('addError', test, None) self.assertThat(result._events, Equals( [('startTest', test), ('addError', test, None), ('stopTest', test)])) self.assertThat( error, KeysEqual( 'twisted-log', 'unhandled-error-in-deferred', 'unhandled-error-in-deferred-1', )) def test_unhandled_error_from_deferred_combined_with_error(self): # If there's a Deferred with an unhandled error, the test fails. Each # unhandled error is reported with a separate traceback, and the error # is still reported. # We're interested in the behavior when debugging is disabled. When # debugging is enabled, we get more stack traces. self.useFixture(DebugTwisted(False)) class SomeCase(TestCase): def test_cruft(self): # Note we aren't returning the Deferred so that the error will # be unhandled. defer.maybeDeferred(lambda: 1/0) 2 / 0 test = SomeCase('test_cruft') runner = self.make_runner(test) result = self.make_result() runner.run(result) error = result._events[1][2] result._events[1] = ('addError', test, None) self.assertThat(result._events, Equals( [('startTest', test), ('addError', test, None), ('stopTest', test)])) self.assertThat( error, KeysEqual( 'traceback', 'twisted-log', 'unhandled-error-in-deferred', )) @skipIf(os.name != "posix", "Sending SIGINT with os.kill is posix only") def test_keyboard_interrupt_stops_test_run(self): # If we get a SIGINT during a test run, the test stops and no more # tests run. SIGINT = getattr(signal, 'SIGINT', None) if not SIGINT: raise self.skipTest("SIGINT unavailable") class SomeCase(TestCase): def test_pause(self): return defer.Deferred() test = SomeCase('test_pause') reactor = self.make_reactor() timeout = self.make_timeout() runner = self.make_runner(test, timeout * 5) result = self.make_result() reactor.callLater(timeout, os.kill, os.getpid(), SIGINT) runner.run(result) self.assertThat(result.shouldStop, Equals(True)) @skipIf(os.name != "posix", "Sending SIGINT with os.kill is posix only") def test_fast_keyboard_interrupt_stops_test_run(self): # If we get a SIGINT during a test run, the test stops and no more # tests run. SIGINT = getattr(signal, 'SIGINT', None) if not SIGINT: raise self.skipTest("SIGINT unavailable") class SomeCase(TestCase): def test_pause(self): return defer.Deferred() test = SomeCase('test_pause') reactor = self.make_reactor() timeout = self.make_timeout() runner = self.make_runner(test, timeout * 5) result = self.make_result() reactor.callWhenRunning(os.kill, os.getpid(), SIGINT) runner.run(result) self.assertThat(result.shouldStop, Equals(True)) def test_timeout_causes_test_error(self): # If a test times out, it reports itself as having failed with a # TimeoutError. class SomeCase(TestCase): def test_pause(self): return defer.Deferred() test = SomeCase('test_pause') runner = self.make_runner(test) result = self.make_result() runner.run(result) error = result._events[1][2] self.assertThat( [event[:2] for event in result._events], Equals( [('startTest', test), ('addError', test), ('stopTest', test)])) self.assertIn('TimeoutError', str(error['traceback'])) def test_convenient_construction(self): # As a convenience method, AsynchronousDeferredRunTest has a # classmethod that returns an AsynchronousDeferredRunTest # factory. This factory has the same API as the RunTest constructor. reactor = object() timeout = object() handler = object() factory = AsynchronousDeferredRunTest.make_factory(reactor, timeout) runner = factory(self, [handler]) self.assertIs(reactor, runner._reactor) self.assertIs(timeout, runner._timeout) self.assertIs(self, runner.case) self.assertEqual([handler], runner.handlers) def test_use_convenient_factory(self): # Make sure that the factory can actually be used. factory = AsynchronousDeferredRunTest.make_factory() class SomeCase(TestCase): run_tests_with = factory def test_something(self): pass case = SomeCase('test_something') case.run() def test_convenient_construction_default_reactor(self): # As a convenience method, AsynchronousDeferredRunTest has a # classmethod that returns an AsynchronousDeferredRunTest # factory. This factory has the same API as the RunTest constructor. reactor = object() handler = object() factory = AsynchronousDeferredRunTest.make_factory(reactor=reactor) runner = factory(self, [handler]) self.assertIs(reactor, runner._reactor) self.assertIs(self, runner.case) self.assertEqual([handler], runner.handlers) def test_convenient_construction_default_timeout(self): # As a convenience method, AsynchronousDeferredRunTest has a # classmethod that returns an AsynchronousDeferredRunTest # factory. This factory has the same API as the RunTest constructor. timeout = object() handler = object() factory = AsynchronousDeferredRunTest.make_factory(timeout=timeout) runner = factory(self, [handler]) self.assertIs(timeout, runner._timeout) self.assertIs(self, runner.case) self.assertEqual([handler], runner.handlers) def test_convenient_construction_default_debugging(self): # As a convenience method, AsynchronousDeferredRunTest has a # classmethod that returns an AsynchronousDeferredRunTest # factory. This factory has the same API as the RunTest constructor. handler = object() factory = AsynchronousDeferredRunTest.make_factory(debug=True) runner = factory(self, [handler]) self.assertIs(self, runner.case) self.assertEqual([handler], runner.handlers) self.assertEqual(True, runner._debug) def test_deferred_error(self): class SomeTest(TestCase): def test_something(self): return defer.maybeDeferred(lambda: 1/0) test = SomeTest('test_something') runner = self.make_runner(test) result = self.make_result() runner.run(result) self.assertThat( [event[:2] for event in result._events], Equals([ ('startTest', test), ('addError', test), ('stopTest', test)])) error = result._events[1][2] self.assertThat(error, KeysEqual('traceback', 'twisted-log')) def test_only_addError_once(self): # Even if the reactor is unclean and the test raises an error and the # cleanups raise errors, we only called addError once per test. # We're interested in the behavior when debugging is disabled. When # debugging is enabled, we get more stack traces. self.useFixture(DebugTwisted(False)) reactor = self.make_reactor() class WhenItRains(TestCase): def it_pours(self): # Add a dirty cleanup. self.addCleanup(lambda: 3 / 0) # Dirty the reactor. from twisted.internet.protocol import ServerFactory reactor.listenTCP(0, ServerFactory(), interface='127.0.0.1') # Unhandled error. defer.maybeDeferred(lambda: 2 / 0) # Actual error. raise RuntimeError("Excess precipitation") test = WhenItRains('it_pours') runner = self.make_runner(test) result = self.make_result() runner.run(result) self.assertThat( [event[:2] for event in result._events], Equals([ ('startTest', test), ('addError', test), ('stopTest', test)])) error = result._events[1][2] self.assertThat( error, KeysEqual( 'traceback', 'traceback-1', 'traceback-2', 'twisted-log', 'unhandled-error-in-deferred', )) def test_log_err_is_error(self): # An error logged during the test run is recorded as an error in the # tests. class LogAnError(TestCase): def test_something(self): try: 1/0 except ZeroDivisionError: f = failure.Failure() log.err(f) test = LogAnError('test_something') runner = self.make_runner(test, store_twisted_logs=False) result = self.make_result() runner.run(result) self.assertThat( result._events, MatchesEvents( ('startTest', test), ('addError', test, { 'logged-error': AsText(ContainsAll([ 'Traceback (most recent call last):', 'ZeroDivisionError', ])), }), ('stopTest', test))) def test_log_err_flushed_is_success(self): # An error logged during the test run is recorded as an error in the # tests. class LogAnError(TestCase): def test_something(self): try: 1/0 except ZeroDivisionError: f = failure.Failure() log.err(f) flush_logged_errors(ZeroDivisionError) test = LogAnError('test_something') runner = self.make_runner(test, store_twisted_logs=False) result = self.make_result() runner.run(result) self.assertThat( result._events, MatchesEvents( ('startTest', test), ('addSuccess', test), ('stopTest', test))) def test_log_in_details(self): # If store_twisted_logs is True, we include the Twisted logs as # details. class LogAnError(TestCase): def test_something(self): log.msg("foo") 1/0 test = LogAnError('test_something') runner = self.make_runner(test, store_twisted_logs=True) result = self.make_result() runner.run(result) self.assertThat( result._events, MatchesEvents( ('startTest', test), ('addError', test, { 'traceback': Not(Is(None)), 'twisted-log': AsText(EndsWith(' foo\n')), }), ('stopTest', test))) def test_do_not_log_to_twisted(self): # If suppress_twisted_logging is True, we don't log anything to the # default Twisted loggers. messages = [] publisher, _ = _get_global_publisher_and_observers() publisher.addObserver(messages.append) self.addCleanup(publisher.removeObserver, messages.append) class LogSomething(TestCase): def test_something(self): log.msg("foo") test = LogSomething('test_something') runner = self.make_runner(test, suppress_twisted_logging=True) result = self.make_result() runner.run(result) self.assertThat(messages, Equals([])) def test_log_to_twisted(self): # If suppress_twisted_logging is False, we log to the default Twisted # loggers. messages = [] publisher, _ = _get_global_publisher_and_observers() publisher.addObserver(messages.append) class LogSomething(TestCase): def test_something(self): log.msg("foo") test = LogSomething('test_something') runner = self.make_runner(test, suppress_twisted_logging=False) result = self.make_result() runner.run(result) self.assertThat( messages, MatchesListwise([ContainsDict({'message': Equals(('foo',))})])) def test_restore_observers(self): # We restore the original observers. publisher, observers = _get_global_publisher_and_observers() class LogSomething(TestCase): def test_something(self): pass test = LogSomething('test_something') runner = self.make_runner(test) result = self.make_result() runner.run(result) self.assertThat( _get_global_publisher_and_observers()[1], Equals(observers)) def test_restore_observers_after_timeout(self): # We restore the original observers even if the test times out. publisher, observers = _get_global_publisher_and_observers() class LogSomething(TestCase): def test_something(self): return defer.Deferred() test = LogSomething('test_something') runner = self.make_runner(test, timeout=0.0001) result = self.make_result() runner.run(result) self.assertThat( _get_global_publisher_and_observers()[1], Equals(observers)) def test_debugging_unchanged_during_test_by_default(self): debugging = [(defer.Deferred.debug, DelayedCall.debug)] class SomeCase(TestCase): def test_debugging_enabled(self): debugging.append((defer.Deferred.debug, DelayedCall.debug)) test = SomeCase('test_debugging_enabled') runner = AsynchronousDeferredRunTest( test, handlers=test.exception_handlers, reactor=self.make_reactor(), timeout=self.make_timeout()) runner.run(self.make_result()) self.assertEqual(debugging[0], debugging[1]) def test_debugging_enabled_during_test_with_debug_flag(self): self.patch(defer.Deferred, 'debug', False) self.patch(DelayedCall, 'debug', False) debugging = [] class SomeCase(TestCase): def test_debugging_enabled(self): debugging.append((defer.Deferred.debug, DelayedCall.debug)) test = SomeCase('test_debugging_enabled') runner = AsynchronousDeferredRunTest( test, handlers=test.exception_handlers, reactor=self.make_reactor(), timeout=self.make_timeout(), debug=True) runner.run(self.make_result()) self.assertEqual([(True, True)], debugging) self.assertEqual(False, defer.Deferred.debug) self.assertEqual(False, defer.Deferred.debug) class TestAssertFailsWith(NeedsTwistedTestCase): """Tests for `assert_fails_with`.""" if SynchronousDeferredRunTest is not None: run_tests_with = SynchronousDeferredRunTest def test_assert_fails_with_success(self): # assert_fails_with fails the test if it's given a Deferred that # succeeds. marker = object() d = assert_fails_with(defer.succeed(marker), RuntimeError) def check_result(failure): failure.trap(self.failureException) self.assertThat( str(failure.value), Equals("RuntimeError not raised (%r returned)" % (marker,))) d.addCallbacks( lambda x: self.fail("Should not have succeeded"), check_result) return d def test_assert_fails_with_success_multiple_types(self): # assert_fails_with fails the test if it's given a Deferred that # succeeds. marker = object() d = assert_fails_with( defer.succeed(marker), RuntimeError, ZeroDivisionError) def check_result(failure): failure.trap(self.failureException) self.assertThat( str(failure.value), Equals("RuntimeError, ZeroDivisionError not raised " "(%r returned)" % (marker,))) d.addCallbacks( lambda x: self.fail("Should not have succeeded"), check_result) return d def test_assert_fails_with_wrong_exception(self): # assert_fails_with fails the test if it's given a Deferred that # succeeds. d = assert_fails_with( defer.maybeDeferred(lambda: 1/0), RuntimeError, KeyboardInterrupt) def check_result(failure): failure.trap(self.failureException) lines = str(failure.value).splitlines() self.assertThat( lines[:2], Equals([ ("ZeroDivisionError raised instead of RuntimeError, " "KeyboardInterrupt:"), " Traceback (most recent call last):", ])) d.addCallbacks( lambda x: self.fail("Should not have succeeded"), check_result) return d def test_assert_fails_with_expected_exception(self): # assert_fails_with calls back with the value of the failure if it's # one of the expected types of failures. try: 1/0 except ZeroDivisionError: f = failure.Failure() d = assert_fails_with(defer.fail(f), ZeroDivisionError) return d.addCallback(self.assertThat, Equals(f.value)) def test_custom_failure_exception(self): # If assert_fails_with is passed a 'failureException' keyword # argument, then it will raise that instead of `AssertionError`. class CustomException(Exception): pass marker = object() d = assert_fails_with( defer.succeed(marker), RuntimeError, failureException=CustomException) def check_result(failure): failure.trap(CustomException) self.assertThat( str(failure.value), Equals("RuntimeError not raised (%r returned)" % (marker,))) return d.addCallbacks( lambda x: self.fail("Should not have succeeded"), check_result) class TestRunWithLogObservers(NeedsTwistedTestCase): def test_restores_observers(self): from testtools.deferredruntest import run_with_log_observers from twisted.python import log # Make sure there's at least one observer. This reproduces bug # #926189. log.addObserver(lambda *args: None) observers = list(log.theLogPublisher.observers) run_with_log_observers([], lambda: None) self.assertEqual(observers, log.theLogPublisher.observers) class TestNoTwistedLogObservers(NeedsTwistedTestCase): """Tests for _NoTwistedLogObservers.""" def _get_logged_messages(self, function, *args, **kwargs): """Run ``function`` and return ``(ret, logged_messages)``.""" messages = [] publisher, _ = _get_global_publisher_and_observers() publisher.addObserver(messages.append) try: ret = function(*args, **kwargs) finally: publisher.removeObserver(messages.append) return (ret, messages) def test_default(self): # The default behaviour is for messages logged to Twisted to actually # go to the Twisted logs. class SomeTest(TestCase): def test_something(self): log.msg('foo') _, messages = self._get_logged_messages(SomeTest('test_something').run) self.assertThat( messages, MatchesListwise([ContainsDict({'message': Equals(('foo',))})])) def test_nothing_logged(self): # Using _NoTwistedLogObservers means that nothing is logged to # Twisted. from testtools.twistedsupport._runtest import _NoTwistedLogObservers class SomeTest(TestCase): def test_something(self): self.useFixture(_NoTwistedLogObservers()) log.msg('foo') _, messages = self._get_logged_messages(SomeTest('test_something').run) self.assertThat(messages, Equals([])) def test_logging_restored(self): # _NoTwistedLogObservers restores the original log observers. from testtools.twistedsupport._runtest import _NoTwistedLogObservers class SomeTest(TestCase): def test_something(self): self.useFixture(_NoTwistedLogObservers()) log.msg('foo') def run_then_log(): SomeTest('test_something').run() log.msg('bar') _, messages = self._get_logged_messages(run_then_log) self.assertThat( messages, MatchesListwise([ContainsDict({'message': Equals(('bar',))})])) class TestTwistedLogObservers(NeedsTwistedTestCase): """Tests for _TwistedLogObservers.""" def test_logged_messages_go_to_observer(self): # Using _TwistedLogObservers means messages logged to Twisted go to # that observer while the fixture is active. from testtools.twistedsupport._runtest import _TwistedLogObservers messages = [] class SomeTest(TestCase): def test_something(self): self.useFixture(_TwistedLogObservers([messages.append])) log.msg('foo') SomeTest('test_something').run() log.msg('bar') self.assertThat( messages, MatchesListwise([ContainsDict({'message': Equals(('foo',))})])) class TestErrorObserver(NeedsTwistedTestCase): """Tests for _ErrorObserver.""" def test_captures_errors(self): # _ErrorObserver stores all errors logged while it is active. from testtools.twistedsupport._runtest import ( _ErrorObserver, _LogObserver, _NoTwistedLogObservers) log_observer = _LogObserver() error_observer = _ErrorObserver(log_observer) exception = ValueError('bar') class SomeTest(TestCase): def test_something(self): # Temporarily suppress default log observers to avoid spewing # to stderr. self.useFixture(_NoTwistedLogObservers()) self.useFixture(error_observer) log.msg('foo') log.err(exception) SomeTest('test_something').run() self.assertThat( error_observer.flush_logged_errors(), MatchesListwise([ AfterPreprocessing(lambda x: x.value, Equals(exception))])) class TestCaptureTwistedLogs(NeedsTwistedTestCase): """Tests for CaptureTwistedLogs.""" def test_captures_logs(self): # CaptureTwistedLogs stores all Twisted log messages as a detail. from testtools.twistedsupport import CaptureTwistedLogs class SomeTest(TestCase): def test_something(self): self.useFixture(CaptureTwistedLogs()) log.msg('foo') test = SomeTest('test_something') test.run() self.assertThat( test.getDetails(), MatchesDict({ 'twisted-log': AsText(Contains('foo')), })) def test_suite(): from unittest2 import TestLoader, TestSuite return TestLoader().loadTestsFromName(__name__) def load_tests(loader, tests, pattern): tests.addTest(make_integration_tests()) return tests