openstack/deb-python-testtools
Code Issues Proposed changes

Merge changes that make Twisted support much, much more robust, along

Browse Source 
with some conveniences.
This commit is contained in:
Jonathan Lange
2010-10-31 12:42:36 -04:00
parent 332d99a0d4 cb70880ad8
commit 9dde40dfa2
8 changed files with 316 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
NEWS
View File

@@ -10,7 +10,7 @@ Improvements
* Experimental support for running tests that return Deferreds.
(Jonathan Lange, Martin [gz])
* Provide a per-test decoractor, run_test_with, to specify which RunTest
* Provide a per-test decorator, run_test_with, to specify which RunTest
object to use for a given test. (Jonathan Lange, #657780)
* Fix the runTest parameter of TestCase to actually work, rather than raising
@@ -30,6 +30,11 @@ Improvements
* Update documentation to say how to use testtools.run() on Python 2.4.
(Jonathan Lange, #501174)
* ``text_content`` conveniently converts a Python string to a Content object.
(Jonathan Lange, James Westby)
* `New `KeysEqual`` matcher. (Jonathan Lange)
0.9.7
~~~~~

11
testtools/_spinner.py
View File

@@ -194,7 +194,16 @@ class Spinner(object):
self._stop_reactor()
def _clean(self):
"""Clean up any junk in the reactor."""
"""Clean up any junk in the reactor.
This will *always* spin the reactor at least once. We do this in
order to give the reactor a chance to do the disconnections and
terminations that were asked of it.
"""
# If we've just run a method that calls, say, loseConnection and then
# returns, then the reactor might not have had a chance to actually
# close the connection yet. Here we explicitly give it such a chance.
self._reactor.iterate(0)
junk = []
for delayed_call in self._reactor.getDelayedCalls():
delayed_call.cancel()

10
testtools/content.py
View File

@@ -5,7 +5,7 @@
import codecs
from testtools.compat import _b
from testtools.content_type import ContentType
from testtools.content_type import ContentType, UTF8_TEXT
from testtools.testresult import TestResult
@@ -89,3 +89,11 @@ class TracebackContent(Content):
value = self._result._exc_info_to_unicode(err, test)
super(TracebackContent, self).__init__(
content_type, lambda: [value.encode("utf8")])
def text_content(text):
"""Create a `Content` object from some text.
This is useful for adding details which are short strings.
"""
return Content(UTF8_TEXT, lambda: [text.encode('utf8')])

100
testtools/deferredruntest.py
View File

@@ -12,8 +12,17 @@ __all__ = [
'SynchronousDeferredRunTest',
]
try:
from StringIO import StringIO
except ImportError:
from io import StringIO
import sys
from testtools.content import (
Content,
text_content,
)
from testtools.content_type import UTF8_TEXT
from testtools.runtest import RunTest
from testtools._spinner import (
extract_result,
@@ -24,10 +33,10 @@ from testtools._spinner import (
)
from twisted.internet import defer
from twisted.python import log
from twisted.trial.unittest import _LogObserver
# TODO: Need a conversion guide for flushLoggedErrors
class SynchronousDeferredRunTest(RunTest):
"""Runner for tests that return synchronous Deferreds."""
@@ -41,6 +50,27 @@ class SynchronousDeferredRunTest(RunTest):
return result
def run_with_log_observers(observers, function, *args, **kwargs):
"""Run 'function' with the given Twisted log observers."""
real_observers = log.theLogPublisher.observers
for observer in real_observers:
log.theLogPublisher.removeObserver(observer)
for observer in observers:
log.theLogPublisher.addObserver(observer)
try:
return function(*args, **kwargs)
finally:
for observer in observers:
log.theLogPublisher.removeObserver(observer)
for observer in real_observers:
log.theLogPublisher.addObserver(observer)
# Observer of the Twisted log that we install during tests.
_log_observer = _LogObserver()
class AsynchronousDeferredRunTest(RunTest):
"""Runner for tests that return Deferreds that fire asynchronously.
@@ -72,11 +102,23 @@ class AsynchronousDeferredRunTest(RunTest):
self._reactor = reactor
self._timeout = timeout
def _got_user_failure(self, failure, tb_label='traceback'):
"""We got a failure from user code."""
# XXX: We don't always get tracebacks from these.
return self._got_user_exception(
(failure.type, failure.value, failure.tb), tb_label=tb_label)
@classmethod
def make_factory(cls, reactor=None, timeout=0.005):
"""Make a factory that conforms to the RunTest factory interface."""
return lambda case, handlers=None: AsynchronousDeferredRunTest(
case, handlers, reactor, timeout)
# This is horrible, but it means that the return value of the method
# will be able to be assigned to a class variable *and* also be
# invoked directly.
class AsynchronousDeferredRunTestFactory:
def __call__(self, case, handlers=None):
return AsynchronousDeferredRunTest(
case, handlers, reactor, timeout)
return AsynchronousDeferredRunTestFactory()
@defer.deferredGenerator
def _run_cleanups(self):
@@ -151,34 +193,53 @@ class AsynchronousDeferredRunTest(RunTest):
except e.__class__:
self._got_user_exception(sys.exc_info())
def _run_core(self):
spinner = Spinner(self._reactor)
def _blocking_run_deferred(self, spinner):
try:
successful, unhandled = trap_unhandled_errors(
return trap_unhandled_errors(
spinner.run, self._timeout, self._run_deferred)
except NoResultError:
# We didn't get a result at all! This could be for any number of
# reasons, but most likely someone hit Ctrl-C during the test.
raise KeyboardInterrupt
except TimeoutError:
# The function took too long to run. No point reporting about
# junk and we don't have any information about unhandled errors in
# deferreds. Report the timeout and skip to the end.
# The function took too long to run.
self._log_user_exception(TimeoutError(self.case, self._timeout))
return
return False, []
def _run_core(self):
# Add an observer to trap all logged errors.
error_observer = _log_observer
full_log = StringIO()
full_observer = log.FileLogObserver(full_log)
spinner = Spinner(self._reactor)
successful, unhandled = run_with_log_observers(
[error_observer.gotEvent, full_observer.emit],
self._blocking_run_deferred, spinner)
self.case.addDetail(
'twisted-log', Content(UTF8_TEXT, full_log.readlines))
logged_errors = error_observer.flushErrors()
for logged_error in logged_errors:
successful = False
self._got_user_failure(logged_error, tb_label='logged-error')
if unhandled:
successful = False
# XXX: Maybe we could log creator & invoker here as well if
# present.
for debug_info in unhandled:
f = debug_info.failResult
self._got_user_exception(
(f.type, f.value, f.tb), 'unhandled-error-in-deferred')
info = debug_info._getDebugTracebacks()
if info:
self.case.addDetail(
'unhandled-error-in-deferred-debug',
text_content(info))
self._got_user_failure(f, 'unhandled-error-in-deferred')
junk = spinner.clear_junk()
if junk:
successful = False
self._log_user_exception(UncleanReactorError(junk))
if successful:
self.result.addSuccess(self.case, details=self.case.getDetails())
@@ -188,9 +249,8 @@ class AsynchronousDeferredRunTest(RunTest):
This just makes sure that it returns a Deferred, regardless of how the
user wrote it.
"""
return defer.maybeDeferred(
super(AsynchronousDeferredRunTest, self)._run_user,
function, *args)
d = defer.maybeDeferred(function, *args)
return d.addErrback(self._got_user_failure)
def assert_fails_with(d, *exc_types, **kwargs):
@@ -227,6 +287,10 @@ def assert_fails_with(d, *exc_types, **kwargs):
return d.addCallbacks(got_success, got_failure)
def flush_logged_errors(*error_types):
return _log_observer.flushErrors(*error_types)
class UncleanReactorError(Exception):
"""Raised when the reactor has junk in it."""

28
testtools/matchers.py
View File

@@ -342,6 +342,34 @@ class StartsWith(Matcher):
return None
class KeysEqual(Matcher):
"""Checks whether a dict has particular keys."""
def __init__(self, *expected):
"""Create a `KeysEqual` Matcher.
:param *expected: The keys the dict is expected to have. If a dict,
then we use the keys of that dict, if a collection, we assume it
is a collection of expected keys.
"""
try:
self.expected = expected.keys()
except AttributeError:
self.expected = list(expected)
def __str__(self):
return "KeysEqual(%s)" % ', '.join(map(repr, self.expected))
def match(self, matchee):
expected = sorted(self.expected)
matched = Equals(expected).match(sorted(matchee.keys()))
if matched:
return AnnotatedMismatch(
'Keys not equal',
_BinaryMismatch(expected, 'does not match', matchee))
return None
class Annotate(object):
"""Annotates a matcher with a descriptive string.

12
testtools/tests/test_content.py
View File

@@ -3,8 +3,8 @@
import unittest
from testtools import TestCase
from testtools.compat import _u
from testtools.content import Content, TracebackContent
from testtools.content_type import ContentType
from testtools.content import Content, TracebackContent, text_content
from testtools.content_type import ContentType, UTF8_TEXT
from testtools.tests.helpers import an_exc_info
@@ -68,6 +68,14 @@ class TestTracebackContent(TestCase):
self.assertEqual(expected, ''.join(list(content.iter_text())))
class TestBytesContent(TestCase):
def test_bytes(self):
data = _u("some data")
expected = Content(UTF8_TEXT, lambda: [data.encode('utf8')])
self.assertEqual(expected, text_content(data))
def test_suite():
from unittest import TestLoader
return TestLoader().loadTestsFromName(__name__)

170
testtools/tests/test_deferredruntest.py
View File

@@ -9,19 +9,24 @@ from testtools import (
skipIf,
TestCase,
)
from testtools.content import (
text_content,
)
from testtools.deferredruntest import (
assert_fails_with,
AsynchronousDeferredRunTest,
flush_logged_errors,
SynchronousDeferredRunTest,
)
from testtools.tests.helpers import ExtendedTestResult
from testtools.matchers import (
Equals,
KeysEqual,
)
from testtools.runtest import RunTest
from twisted.internet import defer
from twisted.python import failure
from twisted.python import failure, log
class X(object):
@@ -38,10 +43,6 @@ class X(object):
self.calls.append('tearDown')
super(X.Base, self).tearDown()
class Success(Base):
expected_calls = ['setUp', 'test', 'tearDown', 'clean-up']
expected_results = [['addSuccess']]
class ErrorInSetup(Base):
expected_calls = ['setUp', 'clean-up']
expected_results = [('addError', RuntimeError)]
@@ -107,7 +108,6 @@ def make_integration_tests():
]
tests = [
X.Success,
X.ErrorInSetup,
X.ErrorInTest,
X.ErrorInTearDown,
@@ -281,21 +281,21 @@ class TestAsynchronousDeferredRunTest(TestCase):
def test_whatever(self):
pass
test = SomeCase('test_whatever')
log = []
a = defer.Deferred().addCallback(lambda x: log.append('a'))
b = defer.Deferred().addCallback(lambda x: log.append('b'))
c = defer.Deferred().addCallback(lambda x: log.append('c'))
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(log, Equals([]))
self.assertThat(call_log, Equals([]))
a.callback(None)
def fire_b():
self.assertThat(log, Equals(['a']))
self.assertThat(call_log, Equals(['a']))
b.callback(None)
def fire_c():
self.assertThat(log, Equals(['a', 'b']))
self.assertThat(call_log, Equals(['a', 'b']))
c.callback(None)
timeout = self.make_timeout()
reactor = self.make_reactor()
@@ -305,7 +305,7 @@ class TestAsynchronousDeferredRunTest(TestCase):
runner = self.make_runner(test, timeout)
result = self.make_result()
runner.run(result)
self.assertThat(log, Equals(['a', 'b', 'c']))
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.
@@ -313,18 +313,19 @@ class TestAsynchronousDeferredRunTest(TestCase):
timeout = self.make_timeout()
class SomeCase(TestCase):
def test_cruft(self):
reactor.callLater(timeout * 2.0, lambda: None)
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]
result._events[1] = ('addError', test, None)
self.assertThat(result._events, Equals(
[('startTest', test),
('addError', test, None),
('stopTest', test)]))
self.assertThat(list(error.keys()), Equals(['traceback']))
self.assertThat(error, KeysEqual('traceback', 'twisted-log'))
def test_unhandled_error_from_deferred(self):
# If there's a Deferred with an unhandled error, the test fails. Each
@@ -346,10 +347,38 @@ class TestAsynchronousDeferredRunTest(TestCase):
('addError', test, None),
('stopTest', test)]))
self.assertThat(
list(error.keys()), Equals([
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.
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):
@@ -419,6 +448,16 @@ class TestAsynchronousDeferredRunTest(TestCase):
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
@@ -443,6 +482,23 @@ class TestAsynchronousDeferredRunTest(TestCase):
self.assertIs(self, runner.case)
self.assertEqual([handler], runner.handlers)
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.
@@ -470,12 +526,76 @@ class TestAsynchronousDeferredRunTest(TestCase):
('stopTest', test)]))
error = result._events[1][2]
self.assertThat(
sorted(error.keys()), Equals([
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)
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('logged-error', 'twisted-log'))
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)
result = self.make_result()
runner.run(result)
self.assertThat(
result._events,
Equals([
('startTest', test),
('addSuccess', test, {'twisted-log': text_content('')}),
('stopTest', test)]))
def test_log_in_details(self):
class LogAnError(TestCase):
def test_something(self):
log.msg("foo")
1/0
test = LogAnError('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'))
class TestAssertFailsWith(TestCase):

26
testtools/tests/test_matchers.py
View File

@@ -13,6 +13,7 @@ from testtools.matchers import (
Equals,
DocTestMatches,
DoesNotStartWith,
KeysEqual,
Is,
LessThan,
MatchesAny,
@@ -211,6 +212,31 @@ class TestMatchesAllInterface(TestCase, TestMatchersInterface):
1, MatchesAll(NotEquals(1), NotEquals(2)))]
class TestKeysEqual(TestCase, TestMatchersInterface):
matches_matcher = KeysEqual('foo', 'bar')
matches_matches = [
{'foo': 0, 'bar': 1},
]
matches_mismatches = [
{},
{'foo': 0},
{'bar': 1},
{'foo': 0, 'bar': 1, 'baz': 2},
{'a': None, 'b': None, 'c': None},
]
str_examples = [
("KeysEqual('foo', 'bar')", KeysEqual('foo', 'bar')),
]
describe_examples = [
("['bar', 'foo'] does not match {'baz': 2, 'foo': 0, 'bar': 1}: "
"Keys not equal",
{'foo': 0, 'bar': 1, 'baz': 2}, KeysEqual('foo', 'bar')),
]
class TestAnnotate(TestCase, TestMatchersInterface):
matches_matcher = Annotate("foo", Equals(1))