openstack/deb-python-testtools
Code Issues Proposed changes

Merge trunk.

Browse Source
This commit is contained in:
Jonathan Lange
2010-12-22 20:22:27 +00:00
parent 42ff4f229a c777b9da74
commit 4a7356d37b
23 changed files with 911 additions and 308 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.bzrignore
View File

@@ -7,3 +7,4 @@ TAGS
apidocs
_trial_temp
doc/_build
./.testrepository

4
.testr.conf
View File

@@ -1,2 +1,4 @@
[DEFAULT]
test_command=PYTHONPATH=. python -m subunit.run testtools.tests.test_suite
test_command=PYTHONPATH=. python -m subunit.run $LISTOPT $IDOPTION testtools.tests.test_suite
test_id_option=--load-list $IDFILE
test_list_option=--list

81
NEWS
View File

@@ -7,12 +7,58 @@ NEXT
Changes
-------
* addUnexpectedSuccess is translated to addFailure for test results that don't
know about addUnexpectedSuccess. (Jonathan Lange, #654474)
* The timestamps generated by ``TestResult`` objects when no timing data has
been received are now datetime-with-timezone, which allows them to be
sensibly serialised and transported. (Robert Collins, #692297)
Improvements
------------
* ``MultiTestResult`` now forwards the ``time`` API. (Robert Collins, #692294)
0.9.8
~~~~~
In this release we bring some very interesting improvements:
* new matchers for exceptions, sets, lists, dicts and more.
* experimental (works but the contract isn't supported) twisted reactor
support.
* The built in runner can now list tests and filter tests (the -l and
--load-list options).
Changes
-------
* addUnexpectedSuccess is translated to addFailure for test results that don't
know about addUnexpectedSuccess. Further, it fails the entire result for
all testtools TestResults (i.e. wasSuccessful() returns False after
addUnexpectedSuccess has been called). Note that when using a delegating
result such as ThreadsafeForwardingResult, MultiTestResult or
ExtendedToOriginalDecorator then the behaviour of addUnexpectedSuccess is
determined by the delegated to result(s).
(Jonathan Lange, Robert Collins, #654474, #683332)
* startTestRun will reset any errors on the result. That is, wasSuccessful()
will always return True immediately after startTestRun() is called. This
only applies to delegated test results (ThreadsafeForwardingResult,
MultiTestResult and ExtendedToOriginalDecorator) if the delegated to result
is a testtools test result - we cannot reliably reset the state of unknown
test result class instances. (Jonathan Lange, Robert Collins, #683332)
* Responsibility for running test cleanups has been moved to ``RunTest``.
This change does not affect public APIs and can be safely ignored by test
authors. (Jonathan Lange, #662647)
Improvements
------------
* ``assertIsInstance`` supports a custom error message to be supplied, which
is necessary when using ``assertDictEqual`` on Python 2.7 with a
``testtools.TestCase`` base class. (Jelmer Vernooij)
* Experimental support for running tests that return Deferreds.
(Jonathan Lange, Martin [gz])
@@ -22,6 +68,9 @@ Improvements
* Fix the runTest parameter of TestCase to actually work, rather than raising
a TypeError. (Jonathan Lange, #657760)
* New matcher ``EndsWith`` added to complement the existing ``StartsWith``
matcher. (Jonathan Lange, #669165)
* Non-release snapshots of testtools will now work with buildout.
(Jonathan Lange, #613734)
@@ -30,6 +79,14 @@ Improvements
* ``MatchesException`` added to the ``testtools.matchers`` module - matches
an exception class and parameters. (Robert Collins)
* ``MismatchesAll.describe`` no longer appends a trailing newline.
(Michael Hudson-Doyle, #686790)
* New ``KeysEqual`` matcher. (Jonathan Lange)
* New helpers for conditionally importing modules, ``try_import`` and
``try_imports``. (Jonathan Lange)
* ``Raises`` added to the ``testtools.matchers`` module - matches if the
supplied callable raises, and delegates to an optional matcher for validation
of the exception. (Robert Collins)
@@ -38,19 +95,29 @@ Improvements
supplied callable raises and delegates to ``MatchesException`` to validate
the exception. (Jonathan Lange)
* ``testools.TestCase.useFixture`` has been added to glue with fixtures nicely.
* Tests will now pass on Python 2.6.4 : an ``Exception`` change made only in
2.6.4 and reverted in Python 2.6.5 was causing test failures on that version.
(Martin [gz], #689858).
* ``testtools.TestCase.useFixture`` has been added to glue with fixtures nicely.
(Robert Collins)
* ``testtools.run`` now supports ``-l`` to list tests rather than executing
them. This is useful for integration with external test analysis/processing
tools like subunit and testrepository. (Robert Collins)
* ``testtools.run`` now supports ``--load-list``, which takes a file containing
test ids, one per line, and intersects those ids with the tests found. This
allows fine grained control of what tests are run even when the tests cannot
be named as objects to import (e.g. due to test parameterisation via
testscenarios). (Robert Collins)
* 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)
* New helpers for conditionally importing modules, ``try_import`` and
``try_imports``. (Jonathan Lange)
0.9.7

45
doc/hacking.rst
View File

@@ -119,28 +119,28 @@ permanently present at the top of the list.
Release tasks
-------------
#. Choose a version number, say X.Y.Z
#. Branch from trunk to testtools-X.Y.Z
#. In testtools-X.Y.Z, ensure __init__ has version X.Y.Z.
#. Replace NEXT in NEWS with the version number X.Y.Z, adjusting the reST.
#. Possibly write a blurb into NEWS.
#. Replace any additional references to NEXT with the version being released.
(should be none).
#. Commit the changes.
#. Tag the release, bzr tag testtools-X.Y.Z
#. Create a source distribution and upload to pypi ('make release').
#. Make sure all "Fix committed" bugs are in the 'next' milestone on Launchpad
#. Rename the 'next' milestone on Launchpad to 'X.Y.Z'
#. Create a release on the newly-renamed 'X.Y.Z' milestone
#. Upload the tarball and asc file to Launchpad
#. Merge the release branch testtools-X.Y.Z into trunk. Before the commit,
add a NEXT heading to the top of NEWS. Push trunk to Launchpad.
#. If a new series has been created (e.g. 0.10.0), make the series on Launchpad.
#. Make a new milestone for the *next release*.
#. During release we rename NEXT to $version.
#. We call new milestones NEXT.
1. Choose a version number, say X.Y.Z
1. Branch from trunk to testtools-X.Y.Z
1. In testtools-X.Y.Z, ensure __init__ has version X.Y.Z.
1. Replace NEXT in NEWS with the version number X.Y.Z, adjusting the reST.
1. Possibly write a blurb into NEWS.
1. Replace any additional references to NEXT with the version being
released. (should be none).
1. Commit the changes.
1. Tag the release, bzr tag testtools-X.Y.Z
1. Create a source distribution and upload to pypi ('make release').
1. Make sure all "Fix committed" bugs are in the 'next' milestone on
Launchpad
1. Rename the 'next' milestone on Launchpad to 'X.Y.Z'
1. Create a release on the newly-renamed 'X.Y.Z' milestone
1. Upload the tarball and asc file to Launchpad
1. Merge the release branch testtools-X.Y.Z into trunk. Before the commit,
add a NEXT heading to the top of NEWS and bump the version in __init__.py.
Push trunk to Launchpad
1. If a new series has been created (e.g. 0.10.0), make the series on Launchpad.
1. Make a new milestone for the *next release*.
1. During release we rename NEXT to $version.
1. We call new milestones NEXT.
.. _PEP 8: http://www.python.org/dev/peps/pep-0008/
.. _unittest: http://docs.python.org/library/unittest.html
@@ -148,3 +148,4 @@ Release tasks
.. _MIT license: http://www.opensource.org/licenses/mit-license.php
.. _Sphinx: http://sphinx.pocoo.org/
.. _restructuredtext: http://docutils.sourceforge.net/rst.html

4
testtools/__init__.py
View File

@@ -32,11 +32,11 @@ from testtools.matchers import (
Matcher,
)
from testtools.runtest import (
MultipleExceptions,
RunTest,
)
from testtools.testcase import (
ErrorHolder,
MultipleExceptions,
PlaceHolder,
TestCase,
clone_test_with_new_id,
@@ -69,4 +69,4 @@ from testtools.testsuite import (
# If the releaselevel is 'final', then the tarball will be major.minor.micro.
# Otherwise it is major.minor.micro~$(revno).
__version__ = (0, 9, 8, 'dev', 0)
__version__ = (0, 9, 9, 'dev', 0)

33
testtools/_spinner.py
View File

@@ -20,7 +20,10 @@ __all__ = [
import signal
from testtools.monkey import MonkeyPatcher
from twisted.internet import defer
from twisted.internet.base import DelayedCall
from twisted.internet.interfaces import IReactorThreads
from twisted.python.failure import Failure
from twisted.python.util import mergeFunctionMetadata
@@ -165,13 +168,20 @@ class Spinner(object):
# the ideal, and it actually works for many cases.
_OBLIGATORY_REACTOR_ITERATIONS = 0
def __init__(self, reactor):
def __init__(self, reactor, debug=False):
"""Construct a Spinner.
:param reactor: A Twisted reactor.
:param debug: Whether or not to enable Twisted's debugging. Defaults
to False.
"""
self._reactor = reactor
self._timeout_call = None
self._success = self._UNSET
self._failure = self._UNSET
self._saved_signals = []
self._junk = []
self._debug = debug
def _cancel_timeout(self):
if self._timeout_call:
@@ -222,7 +232,6 @@ class Spinner(object):
# we aren't going to bother.
junk.append(selectable)
if IReactorThreads.providedBy(self._reactor):
self._reactor.suggestThreadPoolSize(0)
if self._reactor.threadpool is not None:
self._reactor._stopThreadPool()
self._junk.extend(junk)
@@ -270,18 +279,24 @@ class Spinner(object):
:return: Whatever is at the end of the function's callback chain. If
it's an error, then raise that.
"""
debug = MonkeyPatcher()
if self._debug:
debug.add_patch(defer.Deferred, 'debug', True)
debug.add_patch(DelayedCall, 'debug', True)
debug.patch()
try:
junk = self.get_junk()
if junk:
raise StaleJunkError(junk)
self._save_signals()
self._timeout_call = self._reactor.callLater(
timeout, self._timed_out, function, timeout)
# Calling 'stop' on the reactor will make it impossible to re-start
# the reactor. Since the default signal handlers for TERM, BREAK and
# INT all call reactor.stop(), we'll patch it over with crash.
# XXX: It might be a better idea to either install custom signal
# handlers or to override the methods that are Twisted's signal
# handlers.
# Calling 'stop' on the reactor will make it impossible to
# re-start the reactor. Since the default signal handlers for
# TERM, BREAK and INT all call reactor.stop(), we'll patch it over
# with crash. XXX: It might be a better idea to either install
# custom signal handlers or to override the methods that are
# Twisted's signal handlers.
stop, self._reactor.stop = self._reactor.stop, self._reactor.crash
def run_function():
d = defer.maybeDeferred(function, *args, **kwargs)
@@ -297,3 +312,5 @@ class Spinner(object):
return self._get_result()
finally:
self._clean()
finally:
debug.restore()

28
testtools/compat.py
View File

@@ -65,6 +65,34 @@ else:
_u.__doc__ = __u_doc
if sys.version_info > (2, 5):
all = all
_error_repr = BaseException.__repr__
def isbaseexception(exception):
"""Return whether exception inherits from BaseException only"""
return (isinstance(exception, BaseException)
and not isinstance(exception, Exception))
else:
def all(iterable):
"""If contents of iterable all evaluate as boolean True"""
for obj in iterable:
if not obj:
return False
return True
def _error_repr(exception):
"""Format an exception instance as Python 2.5 and later do"""
return exception.__class__.__name__ + repr(exception.args)
def isbaseexception(exception):
"""Return whether exception would inherit from BaseException only
This approximates the hierarchy in Python 2.5 and later, compare the
difference between the diagrams at the bottom of the pages:
<http://docs.python.org/release/2.4.4/lib/module-exceptions.html>
<http://docs.python.org/release/2.5.4/lib/module-exceptions.html>
"""
return isinstance(exception, (KeyboardInterrupt, SystemExit))
def unicode_output_stream(stream):
"""Get wrapper for given stream that writes any unicode without exception

13
testtools/deferredruntest.py
View File

@@ -91,7 +91,8 @@ class AsynchronousDeferredRunTest(_DeferredRunTest):
This is highly experimental code. Use at your own risk.
"""
def __init__(self, case, handlers=None, reactor=None, timeout=0.005):
def __init__(self, case, handlers=None, reactor=None, timeout=0.005,
debug=False):
"""Construct an `AsynchronousDeferredRunTest`.
:param case: The `testtools.TestCase` to run.
@@ -102,22 +103,26 @@ class AsynchronousDeferredRunTest(_DeferredRunTest):
default reactor.
:param timeout: The maximum time allowed for running a test. The
default is 0.005s.
:param debug: Whether or not to enable Twisted's debugging. Use this
to get information about unhandled Deferreds and left-over
DelayedCalls. Defaults to False.
"""
super(AsynchronousDeferredRunTest, self).__init__(case, handlers)
if reactor is None:
from twisted.internet import reactor
self._reactor = reactor
self._timeout = timeout
self._debug = debug
@classmethod
def make_factory(cls, reactor=None, timeout=0.005):
def make_factory(cls, reactor=None, timeout=0.005, debug=False):
"""Make a factory that conforms to the RunTest factory interface."""
# 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 cls(case, handlers, reactor, timeout)
return cls(case, handlers, reactor, timeout, debug)
return AsynchronousDeferredRunTestFactory()
@defer.deferredGenerator
@@ -143,7 +148,7 @@ class AsynchronousDeferredRunTest(_DeferredRunTest):
def _make_spinner(self):
"""Make the `Spinner` to be used to run the tests."""
return Spinner(self._reactor)
return Spinner(self._reactor, debug=self._debug)
def _run_deferred(self):
"""Run the test, assuming everything in it is Deferred-returning.

72
testtools/matchers.py
View File

@@ -14,7 +14,6 @@ __metaclass__ = type
__all__ = [
'Annotate',
'DocTestMatches',
'DoesNotStartWith',
'Equals',
'Is',
'LessThan',
@@ -33,6 +32,8 @@ import operator
from pprint import pformat
import sys
from testtools.compat import classtypes, _error_repr, isbaseexception
class Matcher(object):
"""A pattern matcher.
@@ -179,6 +180,22 @@ class DoesNotStartWith(Mismatch):
self.matchee, self.expected)
class DoesNotEndWith(Mismatch):
def __init__(self, matchee, expected):
"""Create a DoesNotEndWith Mismatch.
:param matchee: the string that did not match.
:param expected: the string that `matchee` was expected to end with.
"""
self.matchee = matchee
self.expected = expected
def describe(self):
return "'%s' does not end with '%s'." % (
self.matchee, self.expected)
class _BinaryComparison(object):
"""Matcher that compares an object to another object."""
@@ -299,7 +316,7 @@ class MismatchesAll(Mismatch):
descriptions = ["Differences: ["]
for mismatch in self.mismatches:
descriptions.append(mismatch.describe())
descriptions.append("]\n")
descriptions.append("]")
return '\n'.join(descriptions)
@@ -344,25 +361,24 @@ class MatchesException(Matcher):
"""
Matcher.__init__(self)
self.expected = exception
def _expected_type(self):
if type(self.expected) is type:
return self.expected
return type(self.expected)
self._is_instance = type(self.expected) not in classtypes()
def match(self, other):
if type(other) != tuple:
return Mismatch('%r is not an exc_info tuple' % other)
if not issubclass(other[0], self._expected_type()):
return Mismatch('%r is not a %r' % (
other[0], self._expected_type()))
if (type(self.expected) is not type and
other[1].args != self.expected.args):
return Mismatch('%r has different arguments to %r.' % (
other[1], self.expected))
expected_class = self.expected
if self._is_instance:
expected_class = expected_class.__class__
if not issubclass(other[0], expected_class):
return Mismatch('%r is not a %r' % (other[0], expected_class))
if self._is_instance and other[1].args != self.expected.args:
return Mismatch('%s has different arguments to %s.' % (
_error_repr(other[1]), _error_repr(self.expected)))
def __str__(self):
return "MatchesException(%r)" % self.expected
if self._is_instance:
return "MatchesException(%s)" % _error_repr(self.expected)
return "MatchesException(%s)" % repr(self.expected)
class StartsWith(Matcher):
@@ -384,6 +400,25 @@ class StartsWith(Matcher):
return None
class EndsWith(Matcher):
"""Checks whether one string starts with another."""
def __init__(self, expected):
"""Create a EndsWith Matcher.
:param expected: the string that matchees should end with.
"""
self.expected = expected
def __str__(self):
return "Ends with '%s'." % self.expected
def match(self, matchee):
if not matchee.endswith(self.expected):
return DoesNotEndWith(matchee, self.expected)
return None
class KeysEqual(Matcher):
"""Checks whether a dict has particular keys."""
@@ -467,7 +502,6 @@ class Raises(Matcher):
# Catch all exceptions: Raises() should be able to match a
# KeyboardInterrupt or SystemExit.
except:
exc_info = sys.exc_info()
if self.exception_matcher:
mismatch = self.exception_matcher.match(sys.exc_info())
if not mismatch:
@@ -476,9 +510,9 @@ class Raises(Matcher):
mismatch = None
# The exception did not match, or no explicit matching logic was
# performed. If the exception is a non-user exception (that is, not
# a subclass of Exception) then propogate it.
if not issubclass(exc_info[0], Exception):
raise exc_info[0], exc_info[1], exc_info[2]
# a subclass of Exception on Python 2.5+) then propogate it.
if isbaseexception(sys.exc_info()[1]):
raise
return mismatch
def __str__(self):

78
testtools/run.py
View File

@@ -14,6 +14,7 @@ import sys
from testtools import TextTestResult
from testtools.compat import classtypes, istext, unicode_output_stream
from testtools.testsuite import iterate_tests
defaultTestLoader = unittest.defaultTestLoader
@@ -34,9 +35,12 @@ else:
class TestToolsTestRunner(object):
""" A thunk object to support unittest.TestProgram."""
def __init__(self, stdout):
self.stdout = stdout
def run(self, test):
"Run the given test case or test suite."
result = TextTestResult(unicode_output_stream(sys.stdout))
result = TextTestResult(unicode_output_stream(self.stdout))
result.startTestRun()
try:
return test.run(result)
@@ -58,6 +62,12 @@ class TestToolsTestRunner(object):
# removed.
# - A tweak has been added to detect 'python -m *.run' and use a
# better progName in that case.
# - self.module is more comprehensively set to None when being invoked from
# the commandline - __name__ is used as a sentinel value.
# - --list has been added which can list tests (should be upstreamed).
# - --load-list has been added which can reduce the tests used (should be
# upstreamed).
# - The limitation of using getopt is declared to the user.
FAILFAST = " -f, --failfast Stop on first failure\n"
CATCHBREAK = " -c, --catch Catch control-C and display results\n"
@@ -70,14 +80,17 @@ Options:
-h, --help Show this message
-v, --verbose Verbose output
-q, --quiet Minimal output
-l, --list List tests rather than executing them.
--load-list Specifies a file containing test ids, only tests matching
those ids are executed.
%(failfast)s%(catchbreak)s%(buffer)s
Examples:
%(progName)s test_module - run tests from test_module
%(progName)s module.TestClass - run tests from module.TestClass
%(progName)s module.Class.test_method - run specified test method
[tests] can be a list of any number of test modules, classes and test
methods.
All options must come before [tests]. [tests] can be a list of any number of
test modules, classes and test methods.
Alternative Usage: %(progName)s discover [options]
@@ -87,6 +100,9 @@ Options:
-p pattern Pattern to match test files ('test*.py' default)
-t directory Top level directory of project (default to
start directory)
-l, --list List tests rather than executing them.
--load-list Specifies a file containing test ids, only tests matching
those ids are executed.
For test discovery all test modules must be importable from the top
level directory of the project.
@@ -102,11 +118,13 @@ class TestProgram(object):
# defaults for testing
failfast = catchbreak = buffer = progName = None
def __init__(self, module='__main__', defaultTest=None, argv=None,
def __init__(self, module=__name__, defaultTest=None, argv=None,
testRunner=None, testLoader=defaultTestLoader,
exit=True, verbosity=1, failfast=None, catchbreak=None,
buffer=None):
if istext(module):
buffer=None, stdout=None):
if module == __name__:
self.module = None
elif istext(module):
self.module = __import__(module)
for part in module.split('.')[1:]:
self.module = getattr(self.module, part)
@@ -114,6 +132,8 @@ class TestProgram(object):
self.module = module
if argv is None:
argv = sys.argv
if stdout is None:
stdout = sys.stdout
self.exit = exit
self.failfast = failfast
@@ -121,6 +141,8 @@ class TestProgram(object):
self.verbosity = verbosity
self.buffer = buffer
self.defaultTest = defaultTest
self.listtests = False
self.load_list = None
self.testRunner = testRunner
self.testLoader = testLoader
progName = argv[0]
@@ -131,7 +153,27 @@ class TestProgram(object):
progName = os.path.basename(argv[0])
self.progName = progName
self.parseArgs(argv)
if self.load_list:
# TODO: preserve existing suites (like testresources does in
# OptimisingTestSuite.add, but with a standard protocol).
# This is needed because the load_tests hook allows arbitrary
# suites, even if that is rarely used.
source = file(self.load_list, 'rb')
try:
lines = source.readlines()
finally:
source.close()
test_ids = set(line.strip() for line in lines)
filtered = unittest.TestSuite()
for test in iterate_tests(self.test):
if test.id() in test_ids:
filtered.addTest(test)
self.test = filtered
if not self.listtests:
self.runTests()
else:
for test in iterate_tests(self.test):
stdout.write('%s\n' % test.id())
def usageExit(self, msg=None):
if msg:
@@ -153,9 +195,10 @@ class TestProgram(object):
return
import getopt
long_opts = ['help', 'verbose', 'quiet', 'failfast', 'catch', 'buffer']
long_opts = ['help', 'verbose', 'quiet', 'failfast', 'catch', 'buffer',
'list', 'load-list=']
try:
options, args = getopt.getopt(argv[1:], 'hHvqfcb', long_opts)
options, args = getopt.getopt(argv[1:], 'hHvqfcbl', long_opts)
for opt, value in options:
if opt in ('-h','-H','--help'):
self.usageExit()
@@ -175,14 +218,15 @@ class TestProgram(object):
if self.buffer is None:
self.buffer = True
# Should this raise an exception if -b is not valid?
if opt in ('-l', '--list'):
self.listtests = True
if opt == '--load-list':
self.load_list = value
if len(args) == 0 and self.defaultTest is None:
# createTests will load tests from self.module
self.testNames = None
elif len(args) > 0:
self.testNames = args
if __name__ == '__main__':
# to support python -m unittest ...
self.module = None
else:
self.testNames = (self.defaultTest,)
self.createTests()
@@ -225,6 +269,10 @@ class TestProgram(object):
help="Pattern to match tests ('test*.py' default)")
parser.add_option('-t', '--top-level-directory', dest='top', default=None,
help='Top level directory of project (defaults to start directory)')
parser.add_option('-l', '--list', dest='listtests', default=False,
help='List tests rather than running them.')
parser.add_option('--load-list', dest='load_list', default=None,
help='Specify a filename containing the test ids to use.')
options, args = parser.parse_args(argv)
if len(args) > 3:
@@ -241,6 +289,8 @@ class TestProgram(object):
self.catchbreak = options.catchbreak
if self.buffer is None:
self.buffer = options.buffer
self.listtests = options.listtests
self.load_list = options.load_list
if options.verbose:
self.verbosity = 2
@@ -274,7 +324,9 @@ class TestProgram(object):
sys.exit(not self.result.wasSuccessful())
################
def main(argv, stdout):
runner = TestToolsTestRunner(stdout)
program = TestProgram(argv=argv, testRunner=runner, stdout=stdout)
if __name__ == '__main__':
runner = TestToolsTestRunner()
program = TestProgram(argv=sys.argv, testRunner=runner)
main(sys.argv, sys.stdout)

73
testtools/runtest.py
View File

@@ -1,8 +1,9 @@
# Copyright (c) 2009 Jonathan M. Lange. See LICENSE for details.
# Copyright (c) 2009-2010 Jonathan M. Lange. See LICENSE for details.
"""Individual test case execution."""
__all__ = [
'MultipleExceptions',
'RunTest',
]
@@ -11,6 +12,13 @@ import sys
from testtools.testresult import ExtendedToOriginalDecorator
class MultipleExceptions(Exception):
"""Represents many exceptions raised from some operation.
:ivar args: The sys.exc_info() tuples for each exception.
"""
class RunTest(object):
"""An object to run a test.
@@ -24,15 +32,15 @@ class RunTest(object):
:ivar case: The test case that is to be run.
:ivar result: The result object a case is reporting to.
:ivar handlers: A list of (ExceptionClass->handler code) for exceptions
that should be caught if raised from the user code. Exceptions that
are caught are checked against this list in first to last order.
There is a catchall of Exception at the end of the list, so to add
a new exception to the list, insert it at the front (which ensures that
it will be checked before any existing base classes in the list. If you
add multiple exceptions some of which are subclasses of each other, add
the most specific exceptions last (so they come before their parent
classes in the list).
:ivar handlers: A list of (ExceptionClass, handler_function) for
exceptions that should be caught if raised from the user
code. Exceptions that are caught are checked against this list in
first to last order. There is a catch-all of `Exception` at the end
of the list, so to add a new exception to the list, insert it at the
front (which ensures that it will be checked before any existing base
classes in the list. If you add multiple exceptions some of which are
subclasses of each other, add the most specific exceptions last (so
they come before their parent classes in the list).
:ivar exception_caught: An object returned when _run_user catches an
exception.
:ivar _exceptions: A list of caught exceptions, used to do the single
@@ -107,9 +115,7 @@ class RunTest(object):
if self.exception_caught == self._run_user(self.case._run_setup,
self.result):
# Don't run the test method if we failed getting here.
e = self.case._runCleanups(self.result)
if e is not None:
self._exceptions.append(e)
self._run_cleanups(self.result)
return
# Run everything from here on in. If any of the methods raise an
# exception we'll have failed.
@@ -125,22 +131,42 @@ class RunTest(object):
failed = True
finally:
try:
e = self._run_user(self.case._runCleanups, self.result)
if e is not None:
self._exceptions.append(e)
if self.exception_caught == self._run_user(
self._run_cleanups, self.result):
failed = True
finally:
if not failed:
self.result.addSuccess(self.case,
details=self.case.getDetails())
def _run_user(self, fn, *args):
def _run_cleanups(self, result):
"""Run the cleanups that have been added with addCleanup.
See the docstring for addCleanup for more information.
:return: None if all cleanups ran without error,
`self.exception_caught` if there was an error.
"""
failing = False
while self.case._cleanups:
function, arguments, keywordArguments = self.case._cleanups.pop()
got_exception = self._run_user(
function, *arguments, **keywordArguments)
if got_exception == self.exception_caught:
failing = True
if failing:
return self.exception_caught
def _run_user(self, fn, *args, **kwargs):
"""Run a user supplied function.
Exceptions are processed by self.handlers.
Exceptions are processed by `_got_user_exception`.
:return: Either whatever 'fn' returns or `self.exception_caught` if
'fn' raised an exception.
"""
try:
return fn(*args)
return fn(*args, **kwargs)
except KeyboardInterrupt:
raise
except:
@@ -149,10 +175,19 @@ class RunTest(object):
def _got_user_exception(self, exc_info, tb_label='traceback'):
"""Called when user code raises an exception.
If 'exc_info' is a `MultipleExceptions`, then we recurse into it
unpacking the errors that it's made up from.
:param exc_info: A sys.exc_info() tuple for the user error.
:param tb_label: An optional string label for the error. If
not specified, will default to 'traceback'.
:return: `exception_caught` if we catch one of the exceptions that
have handlers in `self.handlers`, otherwise raise the error.
"""
if exc_info[0] is MultipleExceptions:
for sub_exc_info in exc_info[1].args:
self._got_user_exception(sub_exc_info, tb_label)
return self.exception_caught
try:
e = exc_info[1]
self.case.onException(exc_info, tb_label=tb_label)

46
testtools/testcase.py
View File

@@ -5,7 +5,6 @@
__metaclass__ = type
__all__ = [
'clone_test_with_new_id',
'MultipleExceptions',
'run_test_with',
'skip',
'skipIf',
@@ -92,13 +91,6 @@ def run_test_with(test_runner, **kwargs):
return decorator
class MultipleExceptions(Exception):
"""Represents many exceptions raised from some operation.
:ivar args: The sys.exc_info() tuples for each exception.
"""
class TestCase(unittest.TestCase):
"""Extensions to the basic TestCase.
@@ -224,35 +216,6 @@ class TestCase(unittest.TestCase):
className = ', '.join(klass.__name__ for klass in classOrIterable)
return className
def _runCleanups(self, result):
"""Run the cleanups that have been added with addCleanup.
See the docstring for addCleanup for more information.
:return: None if all cleanups ran without error, the most recently
raised exception from the cleanups otherwise.
"""
last_exception = None
while self._cleanups:
function, arguments, keywordArguments = self._cleanups.pop()
try:
function(*arguments, **keywordArguments)
except KeyboardInterrupt:
raise
except:
exceptions = [sys.exc_info()]
while exceptions:
try:
exc_info = exceptions.pop()
if exc_info[0] is MultipleExceptions:
exceptions.extend(exc_info[1].args)
continue
self._report_traceback(exc_info)
last_exception = exc_info[1]
finally:
del exc_info
return last_exception
def addCleanup(self, function, *arguments, **keywordArguments):
"""Add a cleanup function to be called after tearDown.
@@ -337,10 +300,11 @@ class TestCase(unittest.TestCase):
self.assertTrue(
needle not in haystack, '%r in %r' % (needle, haystack))
def assertIsInstance(self, obj, klass):
self.assertTrue(
isinstance(obj, klass),
'%r is not an instance of %s' % (obj, self._formatTypes(klass)))
def assertIsInstance(self, obj, klass, msg=None):
if msg is None:
msg = '%r is not an instance of %s' % (
obj, self._formatTypes(klass))
self.assertTrue(isinstance(obj, klass), msg)
def assertRaises(self, excClass, callableObj, *args, **kwargs):
"""Fail unless an exception of class excClass is thrown

18
testtools/testresult/doubles.py
View File

@@ -1,4 +1,4 @@
# Copyright (c) 2009 Jonathan M. Lange. See LICENSE for details.
# Copyright (c) 2009-2010 Jonathan M. Lange. See LICENSE for details.
"""Doubles of test result objects, useful for testing unittest code."""
@@ -15,15 +15,18 @@ class LoggingBase(object):
def __init__(self):
self._events = []
self.shouldStop = False
self._was_successful = True
class Python26TestResult(LoggingBase):
"""A precisely python 2.6 like test result, that logs."""
def addError(self, test, err):
self._was_successful = False
self._events.append(('addError', test, err))
def addFailure(self, test, err):
self._was_successful = False
self._events.append(('addFailure', test, err))
def addSuccess(self, test):
@@ -38,6 +41,9 @@ class Python26TestResult(LoggingBase):
def stopTest(self, test):
self._events.append(('stopTest', test))
def wasSuccessful(self):
return self._was_successful
class Python27TestResult(Python26TestResult):
"""A precisely python 2.7 like test result, that logs."""
@@ -62,9 +68,11 @@ class ExtendedTestResult(Python27TestResult):
"""A test result like the proposed extended unittest result API."""
def addError(self, test, err=None, details=None):
self._was_successful = False
self._events.append(('addError', test, err or details))
def addFailure(self, test, err=None, details=None):
self._was_successful = False
self._events.append(('addFailure', test, err or details))
def addExpectedFailure(self, test, err=None, details=None):
@@ -80,6 +88,7 @@ class ExtendedTestResult(Python27TestResult):
self._events.append(('addSuccess', test))
def addUnexpectedSuccess(self, test, details=None):
self._was_successful = False
if details is not None:
self._events.append(('addUnexpectedSuccess', test, details))
else:
@@ -88,8 +97,15 @@ class ExtendedTestResult(Python27TestResult):
def progress(self, offset, whence):
self._events.append(('progress', offset, whence))
def startTestRun(self):
super(ExtendedTestResult, self).startTestRun()
self._was_successful = True
def tags(self, new_tags, gone_tags):
self._events.append(('tags', new_tags, gone_tags))
def time(self, time):
self._events.append(('time', time))
def wasSuccessful(self):
return self._was_successful

79
testtools/testresult/real.py
View File

@@ -14,7 +14,26 @@ import datetime
import sys
import unittest
from testtools.compat import _format_exc_info, str_is_unicode, _u
from testtools.compat import all, _format_exc_info, str_is_unicode, _u
# From http://docs.python.org/library/datetime.html
_ZERO = datetime.timedelta(0)
# A UTC class.
class UTC(datetime.tzinfo):
"""UTC"""
def utcoffset(self, dt):
return _ZERO
def tzname(self, dt):
return "UTC"
def dst(self, dt):
return _ZERO
utc = UTC()
class TestResult(unittest.TestResult):
@@ -35,13 +54,11 @@ class TestResult(unittest.TestResult):
"""
def __init__(self):
super(TestResult, self).__init__()
self.skip_reasons = {}
self.__now = None
# -- Start: As per python 2.7 --
self.expectedFailures = []
self.unexpectedSuccesses = []
# -- End: As per python 2.7 --
# startTestRun resets all attributes, and older clients don't know to
# call startTestRun, so it is called once here.
# Because subclasses may reasonably not expect this, we call the
# specific version we want to run.
TestResult.startTestRun(self)
def addExpectedFailure(self, test, err=None, details=None):
"""Called when a test has failed in an expected manner.
@@ -108,6 +125,18 @@ class TestResult(unittest.TestResult):
"""Called when a test was expected to fail, but succeed."""
self.unexpectedSuccesses.append(test)
def wasSuccessful(self):
"""Has this result been successful so far?
If there have been any errors, failures or unexpected successes,
return False. Otherwise, return True.
Note: This differs from standard unittest in that we consider
unexpected successes to be equivalent to failures, rather than
successes.
"""
return not (self.errors or self.failures or self.unexpectedSuccesses)
if str_is_unicode:
# Python 3 and IronPython strings are unicode, use parent class method
_exc_info_to_unicode = unittest.TestResult._exc_info_to_string
@@ -139,15 +168,23 @@ class TestResult(unittest.TestResult):
time() method.
"""
if self.__now is None:
return datetime.datetime.now()
return datetime.datetime.now(utc)
else:
return self.__now
def startTestRun(self):
"""Called before a test run starts.
New in python 2.7
New in python 2.7. The testtools version resets the result to a
pristine condition ready for use in another test run.
"""
super(TestResult, self).__init__()
self.skip_reasons = {}
self.__now = None
# -- Start: As per python 2.7 --
self.expectedFailures = []
self.unexpectedSuccesses = []
# -- End: As per python 2.7 --
def stopTestRun(self):
"""Called after a test run completes
@@ -182,7 +219,7 @@ class MultiTestResult(TestResult):
def __init__(self, *results):
TestResult.__init__(self)
self._results = map(ExtendedToOriginalDecorator, results)
self._results = list(map(ExtendedToOriginalDecorator, results))
def _dispatch(self, message, *args, **kwargs):
return tuple(
@@ -220,9 +257,19 @@ class MultiTestResult(TestResult):
def stopTestRun(self):
return self._dispatch('stopTestRun')
def time(self, a_datetime):
return self._dispatch('time', a_datetime)
def done(self):
return self._dispatch('done')
def wasSuccessful(self):
"""Was this result successful?
Only returns True if every constituent result was successful.
"""
return all(self._dispatch('wasSuccessful'))
class TextTestResult(TestResult):
"""A TestResult which outputs activity to a text stream."""
@@ -258,6 +305,10 @@ class TextTestResult(TestResult):
stop = self._now()
self._show_list('ERROR', self.errors)
self._show_list('FAIL', self.failures)
for test in self.unexpectedSuccesses:
self.stream.write(
"%sUNEXPECTED SUCCESS: %s\n%s" % (
self.sep1, test.id(), self.sep2))
self.stream.write("Ran %d test%s in %.3fs\n\n" %
(self.testsRun, plural,
self._delta_to_float(stop - self.__start)))
@@ -267,7 +318,8 @@ class TextTestResult(TestResult):
self.stream.write("FAILED (")
details = []
details.append("failures=%d" % (
len(self.failures) + len(self.errors)))
sum(map(len, (
self.failures, self.errors, self.unexpectedSuccesses)))))
self.stream.write(", ".join(details))
self.stream.write(")\n")
super(TextTestResult, self).stopTestRun()
@@ -363,6 +415,9 @@ class ThreadsafeForwardingResult(TestResult):
self._test_start = self._now()
super(ThreadsafeForwardingResult, self).startTest(test)
def wasSuccessful(self):
return self.result.wasSuccessful()
class ExtendedToOriginalDecorator(object):
"""Permit new TestResult API code to degrade gracefully with old results.

33
testtools/tests/__init__.py
View File

@@ -10,47 +10,32 @@ def test_suite():
test_compat,
test_content,
test_content_type,
test_deferredruntest,
test_fixturesupport,
test_helpers,
test_matchers,
test_monkey,
test_run,
test_runtest,
test_spinner,
test_testtools,
test_testresult,
test_testsuite,
)
suites = []
modules = [
test_compat,
test_content,
test_content_type,
test_deferredruntest,
test_fixturesupport,
test_helpers,
test_matchers,
test_monkey,
test_runtest,
test_run,
test_spinner,
test_testresult,
test_testsuite,
test_testtools,
]
try:
# Tests that rely on Twisted.
from testtools.tests import (
test_deferredruntest,
test_spinner,
)
except ImportError:
pass
else:
modules.extend([test_deferredruntest, test_spinner])
try:
# Tests that rely on 'fixtures'.
from testtools.tests import (
test_fixturesupport,
)
except ImportError:
pass
else:
modules.extend([test_fixturesupport])
for module in modules:
suites.append(getattr(module, 'test_suite')())
suites = map(lambda x:x.test_suite(), modules)
return unittest.TestSuite(suites)

3
testtools/tests/test_compat.py
View File

@@ -19,6 +19,7 @@ from testtools.compat import (
)
from testtools.matchers import (
MatchesException,
Not,
Raises,
)
@@ -246,7 +247,7 @@ class TestUnicodeOutputStream(testtools.TestCase):
if newio:
self.expectFailure("Python 3 StringIO expects text not bytes",
self.assertThat, lambda: soutwrapper.write(self.uni),
Raises(MatchesException(TypeError)))
Not(Raises(MatchesException(TypeError))))
soutwrapper.write(self.uni)
self.assertEqual("pa???n", sout.getvalue())

81
testtools/tests/test_deferredruntest.py
View File

@@ -12,12 +12,7 @@ from testtools import (
from testtools.content import (
text_content,
)
from testtools.deferredruntest import (
assert_fails_with,
AsynchronousDeferredRunTest,
flush_logged_errors,
SynchronousDeferredRunTest,
)
from testtools.helpers import try_import
from testtools.tests.helpers import ExtendedTestResult
from testtools.matchers import (
Equals,
@@ -26,9 +21,20 @@ from testtools.matchers import (
Raises,
)
from testtools.runtest import RunTest
from testtools.tests.test_spinner import NeedsTwistedTestCase
from twisted.internet import defer
from twisted.python import failure, log
assert_fails_with = try_import('testtools.deferredruntest.assert_fails_with')
AsynchronousDeferredRunTest = try_import(
'testtools.deferredruntest.AsynchronousDeferredRunTest')
flush_logged_errors = try_import(
'testtools.deferredruntest.flush_logged_errors')
SynchronousDeferredRunTest = try_import(
'testtools.deferredruntest.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')
class X(object):
@@ -77,7 +83,7 @@ class X(object):
self.calls.append('test')
self.addCleanup(lambda: 1/0)
class TestIntegration(TestCase):
class TestIntegration(NeedsTwistedTestCase):
def assertResultsMatch(self, test, result):
events = list(result._events)
@@ -104,9 +110,9 @@ def make_integration_tests():
from unittest import TestSuite
from testtools import clone_test_with_new_id
runners = [
RunTest,
SynchronousDeferredRunTest,
AsynchronousDeferredRunTest,
('RunTest', RunTest),
('SynchronousDeferredRunTest', SynchronousDeferredRunTest),
('AsynchronousDeferredRunTest', AsynchronousDeferredRunTest),
]
tests = [
@@ -118,12 +124,12 @@ def make_integration_tests():
]
base_test = X.TestIntegration('test_runner')
integration_tests = []
for runner in runners:
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__,
runner_name,
test.__name__))
new_test.test_factory = test
new_test.runner = runner
@@ -131,7 +137,7 @@ def make_integration_tests():
return TestSuite(integration_tests)
class TestSynchronousDeferredRunTest(TestCase):
class TestSynchronousDeferredRunTest(NeedsTwistedTestCase):
def make_result(self):
return ExtendedTestResult()
@@ -185,7 +191,7 @@ class TestSynchronousDeferredRunTest(TestCase):
('stopTest', test)]))
class TestAsynchronousDeferredRunTest(TestCase):
class TestAsynchronousDeferredRunTest(NeedsTwistedTestCase):
def make_reactor(self):
from twisted.internet import reactor
@@ -486,6 +492,17 @@ class TestAsynchronousDeferredRunTest(TestCase):
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):
@@ -601,10 +618,40 @@ class TestAsynchronousDeferredRunTest(TestCase):
error = result._events[1][2]
self.assertThat(error, KeysEqual('traceback', 'twisted-log'))
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])
class TestAssertFailsWith(TestCase):
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):

12
testtools/tests/test_fixturesupport.py
View File

@@ -1,20 +1,26 @@
import unittest
import fixtures
from fixtures.tests.helpers import LoggingFixture
from testtools import (
TestCase,
content,
content_type,
)
from testtools.helpers import try_import
from testtools.tests.helpers import (
ExtendedTestResult,
)
fixtures = try_import('fixtures')
LoggingFixture = try_import('fixtures.tests.helpers.LoggingFixture')
class TestFixtureSupport(TestCase):
def setUp(self):
super(TestFixtureSupport, self).setUp()
if fixtures is None or LoggingFixture is None:
self.skipTest("Need fixtures")
def test_useFixture(self):
fixture = LoggingFixture()
class SimpleTest(TestCase):

53
testtools/tests/test_matchers.py
View File

@@ -13,7 +13,9 @@ from testtools.matchers import (
Annotate,
Equals,
DocTestMatches,
DoesNotEndWith,
DoesNotStartWith,
EndsWith,
KeysEqual,
Is,
LessThan,
@@ -181,8 +183,7 @@ class TestMatchesExceptionInstanceInterface(TestCase, TestMatchersInterface):
MatchesException(Exception('foo')))
]
describe_examples = [
("<type 'exceptions.Exception'> is not a "
"<type 'exceptions.ValueError'>",
("%r is not a %r" % (Exception, ValueError),
error_base_foo,
MatchesException(ValueError("foo"))),
("ValueError('bar',) has different arguments to ValueError('foo',).",
@@ -201,12 +202,11 @@ class TestMatchesExceptionTypeInterface(TestCase, TestMatchersInterface):
matches_mismatches = [error_base_foo]
str_examples = [
("MatchesException(<type 'exceptions.Exception'>)",
("MatchesException(%r)" % Exception,
MatchesException(Exception))
]
describe_examples = [
("<type 'exceptions.Exception'> is not a "
"<type 'exceptions.ValueError'>",
("%r is not a %r" % (Exception, ValueError),
error_base_foo,
MatchesException(ValueError)),
]
@@ -247,8 +247,7 @@ Expected:
Got:
3
]
""",
]""",
"3", MatchesAny(DocTestMatches("1"), DocTestMatches("2")))]
@@ -264,8 +263,7 @@ class TestMatchesAllInterface(TestCase, TestMatchersInterface):
describe_examples = [("""Differences: [
1 == 1
]
""",
]""",
1, MatchesAll(NotEquals(1), NotEquals(2)))]
@@ -362,7 +360,12 @@ class TestRaisesBaseTypes(TestCase):
# Exception, it is propogated.
match_keyb = Raises(MatchesException(KeyboardInterrupt))
def raise_keyb_from_match():
if sys.version_info > (2, 5):
matcher = Raises(MatchesException(Exception))
else:
# On Python 2.4 KeyboardInterrupt is a StandardError subclass
# but should propogate from less generic exception matchers
matcher = Raises(MatchesException(EnvironmentError))
matcher.match(self.raiser)
self.assertThat(raise_keyb_from_match, match_keyb)
@@ -411,6 +414,38 @@ class StartsWithTests(TestCase):
self.assertEqual("bar", mismatch.expected)
class DoesNotEndWithTests(TestCase):
def test_describe(self):
mismatch = DoesNotEndWith("fo", "bo")
self.assertEqual("'fo' does not end with 'bo'.", mismatch.describe())
class EndsWithTests(TestCase):
def test_str(self):
matcher = EndsWith("bar")
self.assertEqual("Ends with 'bar'.", str(matcher))
def test_match(self):
matcher = EndsWith("arf")
self.assertIs(None, matcher.match("barf"))
def test_mismatch_returns_does_not_end_with(self):
matcher = EndsWith("bar")
self.assertIsInstance(matcher.match("foo"), DoesNotEndWith)
def test_mismatch_sets_matchee(self):
matcher = EndsWith("bar")
mismatch = matcher.match("foo")
self.assertEqual("foo", mismatch.matchee)
def test_mismatch_sets_expected(self):
matcher = EndsWith("bar")
mismatch = matcher.match("foo")
self.assertEqual("bar", mismatch.expected)
def test_suite():
from unittest import TestLoader
return TestLoader().loadTestsFromName(__name__)

76
testtools/tests/test_run.py Normal file
View File

@@ -0,0 +1,76 @@
# Copyright (c) 2010 Testtools authors. See LICENSE for details.
"""Tests for the test runner logic."""
from testtools.helpers import try_import, try_imports
fixtures = try_import('fixtures')
StringIO = try_imports(['StringIO.StringIO', 'io.StringIO'])
import testtools
from testtools import TestCase, run
if fixtures:
class SampleTestFixture(fixtures.Fixture):
"""Creates testtools.runexample temporarily."""
def __init__(self):
self.package = fixtures.PythonPackage(
'runexample', [('__init__.py', """
from testtools import TestCase
class TestFoo(TestCase):
def test_bar(self):
pass
def test_quux(self):
pass
def test_suite():
from unittest import TestLoader
return TestLoader().loadTestsFromName(__name__)
""")])
def setUp(self):
super(SampleTestFixture, self).setUp()
self.useFixture(self.package)
testtools.__path__.append(self.package.base)
self.addCleanup(testtools.__path__.remove, self.package.base)
class TestRun(TestCase):
def test_run_list(self):
if fixtures is None:
self.skipTest("Need fixtures")
package = self.useFixture(SampleTestFixture())
out = StringIO()
run.main(['prog', '-l', 'testtools.runexample.test_suite'], out)
self.assertEqual("""testtools.runexample.TestFoo.test_bar
testtools.runexample.TestFoo.test_quux
""", out.getvalue())
def test_run_load_list(self):
if fixtures is None:
self.skipTest("Need fixtures")
package = self.useFixture(SampleTestFixture())
out = StringIO()
# We load two tests - one that exists and one that doesn't, and we
# should get the one that exists and neither the one that doesn't nor
# the unmentioned one that does.
tempdir = self.useFixture(fixtures.TempDir())
tempname = tempdir.path + '/tests.list'
f = open(tempname, 'wb')
try:
f.write("""
testtools.runexample.TestFoo.test_bar
testtools.runexample.missingtest
""")
finally:
f.close()
run.main(['prog', '-l', '--load-list', tempname,
'testtools.runexample.test_suite'], out)
self.assertEqual("""testtools.runexample.TestFoo.test_bar
""", out.getvalue())
def test_suite():
from unittest import TestLoader
return TestLoader().loadTestsFromName(__name__)

108
testtools/tests/test_spinner.py
View File

@@ -9,90 +9,91 @@ from testtools import (
skipIf,
TestCase,
)
from testtools.helpers import try_import
from testtools.matchers import (
Equals,
Is,
MatchesException,
Raises,
)
from testtools._spinner import (
DeferredNotFired,
extract_result,
NoResultError,
not_reentrant,
ReentryError,
Spinner,
StaleJunkError,
TimeoutError,
trap_unhandled_errors,
)
from twisted.internet import defer
from twisted.python.failure import Failure
_spinner = try_import('testtools._spinner')
defer = try_import('twisted.internet.defer')
Failure = try_import('twisted.python.failure.Failure')
class TestNotReentrant(TestCase):
class NeedsTwistedTestCase(TestCase):
def setUp(self):
super(NeedsTwistedTestCase, self).setUp()
if defer is None or Failure is None:
self.skipTest("Need Twisted to run")
class TestNotReentrant(NeedsTwistedTestCase):
def test_not_reentrant(self):
# A function decorated as not being re-entrant will raise a
# ReentryError if it is called while it is running.
# _spinner.ReentryError if it is called while it is running.
calls = []
@not_reentrant
@_spinner.not_reentrant
def log_something():
calls.append(None)
if len(calls) < 5:
log_something()
self.assertThat(log_something, Raises(MatchesException(ReentryError)))
self.assertThat(
log_something, Raises(MatchesException(_spinner.ReentryError)))
self.assertEqual(1, len(calls))
def test_deeper_stack(self):
calls = []
@not_reentrant
@_spinner.not_reentrant
def g():
calls.append(None)
if len(calls) < 5:
f()
@not_reentrant
@_spinner.not_reentrant
def f():
calls.append(None)
if len(calls) < 5:
g()
self.assertThat(f, Raises(MatchesException(ReentryError)))
self.assertThat(f, Raises(MatchesException(_spinner.ReentryError)))
self.assertEqual(2, len(calls))
class TestExtractResult(TestCase):
class TestExtractResult(NeedsTwistedTestCase):
def test_not_fired(self):
# extract_result raises DeferredNotFired if it's given a Deferred that
# has not fired.
self.assertThat(lambda:extract_result(defer.Deferred()),
Raises(MatchesException(DeferredNotFired)))
# _spinner.extract_result raises _spinner.DeferredNotFired if it's
# given a Deferred that has not fired.
self.assertThat(lambda:_spinner.extract_result(defer.Deferred()),
Raises(MatchesException(_spinner.DeferredNotFired)))
def test_success(self):
# extract_result returns the value of the Deferred if it has fired
# successfully.
# _spinner.extract_result returns the value of the Deferred if it has
# fired successfully.
marker = object()
d = defer.succeed(marker)
self.assertThat(extract_result(d), Equals(marker))
self.assertThat(_spinner.extract_result(d), Equals(marker))
def test_failure(self):
# extract_result raises the failure's exception if it's given a
# Deferred that is failing.
# _spinner.extract_result raises the failure's exception if it's given
# a Deferred that is failing.
try:
1/0
except ZeroDivisionError:
f = Failure()
d = defer.fail(f)
self.assertThat(lambda:extract_result(d),
self.assertThat(lambda:_spinner.extract_result(d),
Raises(MatchesException(ZeroDivisionError)))
class TestTrapUnhandledErrors(TestCase):
class TestTrapUnhandledErrors(NeedsTwistedTestCase):
def test_no_deferreds(self):
marker = object()
result, errors = trap_unhandled_errors(lambda: marker)
result, errors = _spinner.trap_unhandled_errors(lambda: marker)
self.assertEqual([], errors)
self.assertIs(marker, result)
@@ -105,12 +106,13 @@ class TestTrapUnhandledErrors(TestCase):
f = Failure()
failures.append(f)
defer.fail(f)
result, errors = trap_unhandled_errors(make_deferred_but_dont_handle)
result, errors = _spinner.trap_unhandled_errors(
make_deferred_but_dont_handle)
self.assertIs(None, result)
self.assertEqual(failures, [error.failResult for error in errors])
class TestRunInReactor(TestCase):
class TestRunInReactor(NeedsTwistedTestCase):
def make_reactor(self):
from twisted.internet import reactor
@@ -119,7 +121,7 @@ class TestRunInReactor(TestCase):
def make_spinner(self, reactor=None):
if reactor is None:
reactor = self.make_reactor()
return Spinner(reactor)
return _spinner.Spinner(reactor)
def make_timeout(self):
return 0.01
@@ -157,8 +159,8 @@ class TestRunInReactor(TestCase):
# to run_in_reactor.
spinner = self.make_spinner()
self.assertThat(lambda: spinner.run(
self.make_timeout(), spinner.run, self.make_timeout(), lambda: None),
Raises(MatchesException(ReentryError)))
self.make_timeout(), spinner.run, self.make_timeout(),
lambda: None), Raises(MatchesException(_spinner.ReentryError)))
def test_deferred_value_returned(self):
# If the given function returns a Deferred, run_in_reactor returns the
@@ -182,11 +184,12 @@ class TestRunInReactor(TestCase):
self.assertEqual(new_hdlrs, map(signal.getsignal, signals))
def test_timeout(self):
# If the function takes too long to run, we raise a TimeoutError.
# If the function takes too long to run, we raise a
# _spinner.TimeoutError.
timeout = self.make_timeout()
self.assertThat(
lambda:self.make_spinner().run(timeout, lambda: defer.Deferred()),
Raises(MatchesException(TimeoutError)))
Raises(MatchesException(_spinner.TimeoutError)))
def test_no_junk_by_default(self):
# If the reactor hasn't spun yet, then there cannot be any junk.
@@ -241,7 +244,14 @@ class TestRunInReactor(TestCase):
timeout = self.make_timeout()
spinner = self.make_spinner(reactor)
spinner.run(timeout, reactor.callInThread, time.sleep, timeout / 2.0)
self.assertThat(list(threading.enumerate()), Equals(current_threads))
# Python before 2.5 has a race condition with thread handling where
# join() does not remove threads from enumerate before returning - the
# thread being joined does the removal. This was fixed in Python 2.5
# but we still support 2.4, so we have to workaround the issue.
# http://bugs.python.org/issue1703448.
self.assertThat(
[thread for thread in threading.enumerate() if thread.isAlive()],
Equals(current_threads))
def test_leftover_junk_available(self):
# If 'run' is given a function that leaves the reactor dirty in some
@@ -263,7 +273,7 @@ class TestRunInReactor(TestCase):
timeout = self.make_timeout()
spinner.run(timeout, reactor.listenTCP, 0, ServerFactory())
self.assertThat(lambda: spinner.run(timeout, lambda: None),
Raises(MatchesException(StaleJunkError)))
Raises(MatchesException(_spinner.StaleJunkError)))
def test_clear_junk_clears_previous_junk(self):
# If 'run' is called and there's still junk in the spinner's junk
@@ -279,7 +289,7 @@ class TestRunInReactor(TestCase):
@skipIf(os.name != "posix", "Sending SIGINT with os.kill is posix only")
def test_sigint_raises_no_result_error(self):
# If we get a SIGINT during a run, we raise NoResultError.
# If we get a SIGINT during a run, we raise _spinner.NoResultError.
SIGINT = getattr(signal, 'SIGINT', None)
if not SIGINT:
self.skipTest("SIGINT not available")
@@ -288,19 +298,19 @@ class TestRunInReactor(TestCase):
timeout = self.make_timeout()
reactor.callLater(timeout, os.kill, os.getpid(), SIGINT)
self.assertThat(lambda:spinner.run(timeout * 5, defer.Deferred),
Raises(MatchesException(NoResultError)))
Raises(MatchesException(_spinner.NoResultError)))
self.assertEqual([], spinner._clean())
@skipIf(os.name != "posix", "Sending SIGINT with os.kill is posix only")
def test_sigint_raises_no_result_error_second_time(self):
# If we get a SIGINT during a run, we raise NoResultError. This test
# is exactly the same as test_sigint_raises_no_result_error, and
# exists to make sure we haven't futzed with state.
# If we get a SIGINT during a run, we raise _spinner.NoResultError.
# This test is exactly the same as test_sigint_raises_no_result_error,
# and exists to make sure we haven't futzed with state.
self.test_sigint_raises_no_result_error()
@skipIf(os.name != "posix", "Sending SIGINT with os.kill is posix only")
def test_fast_sigint_raises_no_result_error(self):
# If we get a SIGINT during a run, we raise NoResultError.
# If we get a SIGINT during a run, we raise _spinner.NoResultError.
SIGINT = getattr(signal, 'SIGINT', None)
if not SIGINT:
self.skipTest("SIGINT not available")
@@ -309,7 +319,7 @@ class TestRunInReactor(TestCase):
timeout = self.make_timeout()
reactor.callWhenRunning(os.kill, os.getpid(), SIGINT)
self.assertThat(lambda:spinner.run(timeout * 5, defer.Deferred),
Raises(MatchesException(NoResultError)))
Raises(MatchesException(_spinner.NoResultError)))
self.assertEqual([], spinner._clean())
@skipIf(os.name != "posix", "Sending SIGINT with os.kill is posix only")

218
testtools/tests/test_testresult.py
View File

@@ -45,12 +45,44 @@ from testtools.tests.helpers import (
ExtendedTestResult,
an_exc_info
)
from testtools.testresult.real import utc
StringIO = try_imports(['StringIO.StringIO', 'io.StringIO'])
class TestTestResultContract(TestCase):
"""Tests for the contract of TestResults."""
class Python26Contract(object):
def test_fresh_result_is_successful(self):
# A result is considered successful before any tests are run.
result = self.makeResult()
self.assertTrue(result.wasSuccessful())
def test_addError_is_failure(self):
# addError fails the test run.
result = self.makeResult()
result.startTest(self)
result.addError(self, an_exc_info)
result.stopTest(self)
self.assertFalse(result.wasSuccessful())
def test_addFailure_is_failure(self):
# addFailure fails the test run.
result = self.makeResult()
result.startTest(self)
result.addFailure(self, an_exc_info)
result.stopTest(self)
self.assertFalse(result.wasSuccessful())
def test_addSuccess_is_success(self):
# addSuccess does not fail the test run.
result = self.makeResult()
result.startTest(self)
result.addSuccess(self)
result.stopTest(self)
self.assertTrue(result.wasSuccessful())
class Python27Contract(Python26Contract):
def test_addExpectedFailure(self):
# Calling addExpectedFailure(test, exc_info) completes ok.
@@ -58,6 +90,52 @@ class TestTestResultContract(TestCase):
result.startTest(self)
result.addExpectedFailure(self, an_exc_info)
def test_addExpectedFailure_is_success(self):
# addExpectedFailure does not fail the test run.
result = self.makeResult()
result.startTest(self)
result.addExpectedFailure(self, an_exc_info)
result.stopTest(self)
self.assertTrue(result.wasSuccessful())
def test_addSkipped(self):
# Calling addSkip(test, reason) completes ok.
result = self.makeResult()
result.startTest(self)
result.addSkip(self, _u("Skipped for some reason"))
def test_addSkip_is_success(self):
# addSkip does not fail the test run.
result = self.makeResult()
result.startTest(self)
result.addSkip(self, _u("Skipped for some reason"))
result.stopTest(self)
self.assertTrue(result.wasSuccessful())
def test_addUnexpectedSuccess(self):
# Calling addUnexpectedSuccess(test) completes ok.
result = self.makeResult()
result.startTest(self)
result.addUnexpectedSuccess(self)
def test_addUnexpectedSuccess_was_successful(self):
# addUnexpectedSuccess does not fail the test run in Python 2.7.
result = self.makeResult()
result.startTest(self)
result.addUnexpectedSuccess(self)
result.stopTest(self)
self.assertTrue(result.wasSuccessful())
def test_startStopTestRun(self):
# Calling startTestRun completes ok.
result = self.makeResult()
result.startTestRun()
result.stopTestRun()
class DetailsContract(Python27Contract):
"""Tests for the contract of TestResults."""
def test_addExpectedFailure_details(self):
# Calling addExpectedFailure(test, details=xxx) completes ok.
result = self.makeResult()
@@ -76,24 +154,12 @@ class TestTestResultContract(TestCase):
result.startTest(self)
result.addFailure(self, details={})
def test_addSkipped(self):
# Calling addSkip(test, reason) completes ok.
result = self.makeResult()
result.startTest(self)
result.addSkip(self, _u("Skipped for some reason"))
def test_addSkipped_details(self):
# Calling addSkip(test, reason) completes ok.
result = self.makeResult()
result.startTest(self)
result.addSkip(self, details={})
def test_addUnexpectedSuccess(self):
# Calling addUnexpectedSuccess(test) completes ok.
result = self.makeResult()
result.startTest(self)
result.addUnexpectedSuccess(self)
def test_addUnexpectedSuccess_details(self):
# Calling addUnexpectedSuccess(test) completes ok.
result = self.makeResult()
@@ -106,32 +172,73 @@ class TestTestResultContract(TestCase):
result.startTest(self)
result.addSuccess(self, details={})
def test_startStopTestRun(self):
# Calling startTestRun completes ok.
class FallbackContract(DetailsContract):
"""When we fallback we take our policy choice to map calls.
For instance, we map unexpectedSuccess to an error code, not to success.
"""
def test_addUnexpectedSuccess_was_successful(self):
# addUnexpectedSuccess fails test run in testtools.
result = self.makeResult()
result.startTest(self)
result.addUnexpectedSuccess(self)
result.stopTest(self)
self.assertFalse(result.wasSuccessful())
class StartTestRunContract(FallbackContract):
"""Defines the contract for testtools policy choices.
That is things which are not simply extensions to unittest but choices we
have made differently.
"""
def test_startTestRun_resets_unexpected_success(self):
result = self.makeResult()
result.startTest(self)
result.addUnexpectedSuccess(self)
result.stopTest(self)
result.startTestRun()
result.stopTestRun()
self.assertTrue(result.wasSuccessful())
def test_startTestRun_resets_failure(self):
result = self.makeResult()
result.startTest(self)
result.addFailure(self, an_exc_info)
result.stopTest(self)
result.startTestRun()
self.assertTrue(result.wasSuccessful())
def test_startTestRun_resets_errors(self):
result = self.makeResult()
result.startTest(self)
result.addError(self, an_exc_info)
result.stopTest(self)
result.startTestRun()
self.assertTrue(result.wasSuccessful())
class TestTestResultContract(TestTestResultContract):
class TestTestResultContract(TestCase, StartTestRunContract):
def makeResult(self):
return TestResult()
class TestMultiTestresultContract(TestTestResultContract):
class TestMultiTestResultContract(TestCase, StartTestRunContract):
def makeResult(self):
return MultiTestResult(TestResult(), TestResult())
class TestTextTestResultContract(TestTestResultContract):
class TestTextTestResultContract(TestCase, StartTestRunContract):
def makeResult(self):
return TextTestResult(StringIO())
class TestThreadSafeForwardingResultContract(TestTestResultContract):
class TestThreadSafeForwardingResultContract(TestCase, StartTestRunContract):
def makeResult(self):
result_semaphore = threading.Semaphore(1)
@@ -139,6 +246,36 @@ class TestThreadSafeForwardingResultContract(TestTestResultContract):
return ThreadsafeForwardingResult(target, result_semaphore)
class TestExtendedTestResultContract(TestCase, StartTestRunContract):
def makeResult(self):
return ExtendedTestResult()
class TestPython26TestResultContract(TestCase, Python26Contract):
def makeResult(self):
return Python26TestResult()
class TestAdaptedPython26TestResultContract(TestCase, FallbackContract):
def makeResult(self):
return ExtendedToOriginalDecorator(Python26TestResult())
class TestPython27TestResultContract(TestCase, Python27Contract):
def makeResult(self):
return Python27TestResult()
class TestAdaptedPython27TestResultContract(TestCase, DetailsContract):
def makeResult(self):
return ExtendedToOriginalDecorator(Python27TestResult())
class TestTestResult(TestCase):
"""Tests for `TestResult`."""
@@ -169,10 +306,10 @@ class TestTestResult(TestCase):
self.addCleanup(restore)
class Module:
pass
now = datetime.datetime.now()
now = datetime.datetime.now(utc)
stubdatetime = Module()
stubdatetime.datetime = Module()
stubdatetime.datetime.now = lambda: now
stubdatetime.datetime.now = lambda tz: now
testresult.real.datetime = stubdatetime
# Calling _now() looks up the time.
self.assertEqual(now, result._now())
@@ -187,7 +324,7 @@ class TestTestResult(TestCase):
def test_now_datetime_time(self):
result = self.makeResult()
now = datetime.datetime.now()
now = datetime.datetime.now(utc)
result.time(now)
self.assertEqual(now, result._now())
@@ -288,6 +425,11 @@ class TestMultiTestResult(TestWithFakeExceptions):
result = multi_result.stopTestRun()
self.assertEqual(('foo', 'foo'), result)
def test_time(self):
# the time call is dispatched, not eaten by the base class
self.multiResult.time('foo')
self.assertResultLogsEqual([('time', 'foo')])
class TestTextTestResult(TestCase):
"""Tests for `TextTestResult`."""
@@ -308,6 +450,12 @@ class TestTextTestResult(TestCase):
self.fail("yo!")
return Test("failed")
def make_unexpectedly_successful_test(self):
class Test(TestCase):
def succeeded(self):
self.expectFailure("yo!", lambda: None)
return Test("succeeded")
def make_test(self):
class Test(TestCase):
def test(self):
@@ -359,7 +507,7 @@ class TestTextTestResult(TestCase):
def test_stopTestRun_current_time(self):
test = self.make_test()
now = datetime.datetime.now()
now = datetime.datetime.now(utc)
self.result.time(now)
self.result.startTestRun()
self.result.startTest(test)
@@ -393,9 +541,18 @@ class TestTextTestResult(TestCase):
self.assertThat(self.getvalue(),
DocTestMatches("...\n\nFAILED (failures=1)\n", doctest.ELLIPSIS))
def test_stopTestRun_not_successful_unexpected_success(self):
test = self.make_unexpectedly_successful_test()
self.result.startTestRun()
test.run(self.result)
self.result.stopTestRun()
self.assertThat(self.getvalue(),
DocTestMatches("...\n\nFAILED (failures=1)\n", doctest.ELLIPSIS))
def test_stopTestRun_shows_details(self):
self.result.startTestRun()
self.make_erroring_test().run(self.result)
self.make_unexpectedly_successful_test().run(self.result)
self.make_failing_test().run(self.result)
self.reset_output()
self.result.stopTestRun()
@@ -407,7 +564,7 @@ Text attachment: traceback
------------
Traceback (most recent call last):
File "...testtools...runtest.py", line ..., in _run_user...
return fn(*args)
return fn(*args, **kwargs)
File "...testtools...testcase.py", line ..., in _run_test_method
return self._get_test_method()()
File "...testtools...tests...test_testresult.py", line ..., in error
@@ -421,14 +578,17 @@ Text attachment: traceback
------------
Traceback (most recent call last):
File "...testtools...runtest.py", line ..., in _run_user...
return fn(*args)
return fn(*args, **kwargs)
File "...testtools...testcase.py", line ..., in _run_test_method
return self._get_test_method()()
File "...testtools...tests...test_testresult.py", line ..., in failed
self.fail("yo!")
AssertionError: yo!
------------
...""", doctest.ELLIPSIS))
======================================================================
UNEXPECTED SUCCESS: testtools.tests.test_testresult.Test.succeeded
----------------------------------------------------------------------
...""", doctest.ELLIPSIS | doctest.REPORT_NDIFF))
class TestThreadSafeForwardingResult(TestWithFakeExceptions):
@@ -1076,6 +1236,8 @@ class TestNonAsciiResults(TestCase):
"class UnprintableError(Exception):\n"
" def __str__(self):\n"
" raise RuntimeError\n"
" def __unicode__(self):\n"
" raise RuntimeError\n"
" def __repr__(self):\n"
" raise RuntimeError\n")
textoutput = self._test_external_case(

4
testtools/tests/test_testtools.py
View File

@@ -375,6 +375,10 @@ class TestAssertions(TestCase):
'42 is not an instance of %s' % self._formatTypes([Foo, Bar]),
self.assertIsInstance, 42, (Foo, Bar))
def test_assertIsInstance_overridden_message(self):
# assertIsInstance(obj, klass, msg) permits a custom message.
self.assertFails("foo", self.assertIsInstance, 42, str, "foo")
def test_assertIs(self):
# assertIs asserts that an object is identical to another object.
self.assertIs(None, None)