Viewing file:  test_reporter.py (51.09 KB) -rw-r--r--
Select action/file-type:
 (+) |  (+) |  (+) | Code (+) | Session (+) |  (+) | SDB (+) |  (+) |  (+) |  (+) |  (+) |  (+) |
# Copyright (c) 2001-2010 Twisted Matrix Laboratories.
# See LICENSE for details.
#
# Maintainer: Jonathan Lange
"""
Tests for L{twisted.trial.reporter}.
"""
import errno, sys, os, re, StringIO
from twisted.internet.utils import suppressWarnings
from twisted.python import log
from twisted.python.failure import Failure
from twisted.trial import itrial, unittest, runner, reporter, util
from twisted.trial.reporter import UncleanWarningsReporterWrapper
from twisted.trial.test import erroneous
from twisted.trial.unittest import makeTodo, SkipTest, Todo
class BrokenStream(object):
"""
Stream-ish object that raises a signal interrupt error. We use this to make
sure that Trial still manages to write what it needs to write.
"""
written = False
flushed = False
def __init__(self, fObj):
self.fObj = fObj
def write(self, s):
if self.written:
return self.fObj.write(s)
self.written = True
raise IOError(errno.EINTR, "Interrupted write")
def flush(self):
if self.flushed:
return self.fObj.flush()
self.flushed = True
raise IOError(errno.EINTR, "Interrupted flush")
class StringTest(unittest.TestCase):
def stringComparison(self, expect, output):
output = filter(None, output)
self.failUnless(len(expect) <= len(output),
"Must have more observed than expected"
"lines %d < %d" % (len(output), len(expect)))
REGEX_PATTERN_TYPE = type(re.compile(''))
for line_number, (exp, out) in enumerate(zip(expect, output)):
if exp is None:
continue
elif isinstance(exp, str):
self.assertSubstring(exp, out, "Line %d: %r not in %r"
% (line_number, exp, out))
elif isinstance(exp, REGEX_PATTERN_TYPE):
self.failUnless(exp.match(out),
"Line %d: %r did not match string %r"
% (line_number, exp.pattern, out))
else:
raise TypeError("don't know what to do with object %r"
% (exp,))
class TestTestResult(unittest.TestCase):
def setUp(self):
self.result = reporter.TestResult()
def test_pyunitAddError(self):
# pyunit passes an exc_info tuple directly to addError
try:
raise RuntimeError('foo')
except RuntimeError, excValue:
self.result.addError(self, sys.exc_info())
failure = self.result.errors[0][1]
self.assertEqual(excValue, failure.value)
self.assertEqual(RuntimeError, failure.type)
def test_pyunitAddFailure(self):
# pyunit passes an exc_info tuple directly to addFailure
try:
raise self.failureException('foo')
except self.failureException, excValue:
self.result.addFailure(self, sys.exc_info())
failure = self.result.failures[0][1]
self.assertEqual(excValue, failure.value)
self.assertEqual(self.failureException, failure.type)
class TestReporterRealtime(TestTestResult):
def setUp(self):
output = StringIO.StringIO()
self.result = reporter.Reporter(output, realtime=True)
class TestErrorReporting(StringTest):
doubleSeparator = re.compile(r'^=+$')
def setUp(self):
self.loader = runner.TestLoader()
self.output = StringIO.StringIO()
self.result = reporter.Reporter(self.output)
def getOutput(self, suite):
result = self.getResult(suite)
result.done()
return self.output.getvalue()
def getResult(self, suite):
suite.run(self.result)
return self.result
def testFormatErroredMethod(self):
suite = self.loader.loadClass(erroneous.TestFailureInSetUp)
output = self.getOutput(suite).splitlines()
match = [self.doubleSeparator,
('[ERROR]: twisted.trial.test.erroneous.'
'TestFailureInSetUp.test_noop'),
'Traceback (most recent call last):',
re.compile(r'^\s+File .*erroneous\.py., line \d+, in setUp$'),
re.compile(r'^\s+raise FoolishError, '
r'.I am a broken setUp method.$'),
('twisted.trial.test.erroneous.FoolishError: '
'I am a broken setUp method')]
self.stringComparison(match, output)
def testFormatFailedMethod(self):
suite = self.loader.loadMethod(erroneous.TestRegularFail.test_fail)
output = self.getOutput(suite).splitlines()
match = [
self.doubleSeparator,
'[FAIL]: '
'twisted.trial.test.erroneous.TestRegularFail.test_fail',
'Traceback (most recent call last):',
re.compile(r'^\s+File .*erroneous\.py., line \d+, in test_fail$'),
re.compile(r'^\s+self\.fail\("I fail"\)$'),
'twisted.trial.unittest.FailTest: I fail'
]
self.stringComparison(match, output)
def testDoctestError(self):
from twisted.trial.test import erroneous
suite = unittest.decorate(
self.loader.loadDoctests(erroneous), itrial.ITestCase)
output = self.getOutput(suite)
path = 'twisted.trial.test.erroneous.unexpectedException'
for substring in ['1/0', 'ZeroDivisionError',
'Exception raised:', path]:
self.assertSubstring(substring, output)
self.failUnless(re.search('Fail(ed|ure in) example:', output),
"Couldn't match 'Failure in example: ' "
"or 'Failed example: '")
expect = [self.doubleSeparator,
re.compile(r'\[(ERROR|FAIL)\]: .*' + re.escape(path))]
self.stringComparison(expect, output.splitlines())
def testHiddenException(self):
"""
Check that errors in C{DelayedCall}s get reported, even if the
test already has a failure.
Only really necessary for testing the deprecated style of tests that
use iterate() directly. See
L{erroneous.DelayedCall.testHiddenException} for more details.
"""
test = erroneous.DelayedCall('testHiddenException')
output = self.getOutput(test).splitlines()
match = [
self.doubleSeparator,
'[FAIL]: '
'twisted.trial.test.erroneous.DelayedCall.testHiddenException',
'Traceback (most recent call last):',
re.compile(r'^\s+File .*erroneous\.py., line \d+, in '
'testHiddenException$'),
re.compile(r'^\s+self\.fail\("Deliberate failure to mask the '
'hidden exception"\)$'),
'twisted.trial.unittest.FailTest: '
'Deliberate failure to mask the hidden exception',
self.doubleSeparator,
'[ERROR]: '
'twisted.trial.test.erroneous.DelayedCall.testHiddenException',
'Traceback (most recent call last):',
re.compile(r'^\s+File .* in runUntilCurrent'),
re.compile(r'^\s+.*'),
re.compile('^\s+File .*erroneous\.py", line \d+, in go'),
re.compile('^\s+raise RuntimeError\(self.hiddenExceptionMsg\)'),
'exceptions.RuntimeError: something blew up',
]
self.stringComparison(match, output)
class TestUncleanWarningWrapperErrorReporting(TestErrorReporting):
"""
Tests that the L{UncleanWarningsReporterWrapper} can sufficiently proxy
IReporter failure and error reporting methods to a L{reporter.Reporter}.
"""
def setUp(self):
self.loader = runner.TestLoader()
self.output = StringIO.StringIO()
self.result = UncleanWarningsReporterWrapper(
reporter.Reporter(self.output))
class TracebackHandling(unittest.TestCase):
def getErrorFrames(self, test):
stream = StringIO.StringIO()
result = reporter.Reporter(stream)
test.run(result)
bads = result.failures + result.errors
assert len(bads) == 1
assert bads[0][0] == test
return result._trimFrames(bads[0][1].frames)
def checkFrames(self, observedFrames, expectedFrames):
for observed, expected in zip(observedFrames, expectedFrames):
self.assertEqual(observed[0], expected[0])
observedSegs = os.path.splitext(observed[1])[0].split(os.sep)
expectedSegs = expected[1].split('/')
self.assertEqual(observedSegs[-len(expectedSegs):],
expectedSegs)
self.assertEqual(len(observedFrames), len(expectedFrames))
def test_basic(self):
test = erroneous.TestRegularFail('test_fail')
frames = self.getErrorFrames(test)
self.checkFrames(frames,
[('test_fail', 'twisted/trial/test/erroneous')])
def test_subroutine(self):
test = erroneous.TestRegularFail('test_subfail')
frames = self.getErrorFrames(test)
self.checkFrames(frames,
[('test_subfail', 'twisted/trial/test/erroneous'),
('subroutine', 'twisted/trial/test/erroneous')])
def test_deferred(self):
test = erroneous.TestFailureInDeferredChain('test_fail')
frames = self.getErrorFrames(test)
self.checkFrames(frames,
[('_later', 'twisted/trial/test/erroneous')])
def test_noFrames(self):
result = reporter.Reporter(None)
self.assertEqual([], result._trimFrames([]))
def test_oneFrame(self):
result = reporter.Reporter(None)
self.assertEqual(['fake frame'], result._trimFrames(['fake frame']))
class FormatFailures(StringTest):
def setUp(self):
try:
raise RuntimeError('foo')
except RuntimeError:
self.f = Failure()
self.f.frames = [
['foo', 'foo/bar.py', 5, [('x', 5)], [('y', 'orange')]],
['qux', 'foo/bar.py', 10, [('a', 'two')], [('b', 'MCMXCIX')]]
]
self.stream = StringIO.StringIO()
self.result = reporter.Reporter(self.stream)
def test_formatDefault(self):
tb = self.result._formatFailureTraceback(self.f)
self.stringComparison([
'Traceback (most recent call last):',
' File "foo/bar.py", line 5, in foo',
re.compile(r'^\s*$'),
' File "foo/bar.py", line 10, in qux',
re.compile(r'^\s*$'),
'RuntimeError: foo'], tb.splitlines())
def test_formatString(self):
tb = '''
File "twisted/trial/unittest.py", line 256, in failUnlessSubstring
return self.failUnlessIn(substring, astring, msg)
exceptions.TypeError: iterable argument required
'''
expected = '''
File "twisted/trial/unittest.py", line 256, in failUnlessSubstring
return self.failUnlessIn(substring, astring, msg)
exceptions.TypeError: iterable argument required
'''
formatted = self.result._formatFailureTraceback(tb)
self.assertEqual(expected, formatted)
def test_mutation(self):
frames = self.f.frames[:]
tb = self.result._formatFailureTraceback(self.f)
self.assertEqual(self.f.frames, frames)
class PyunitTestNames(unittest.TestCase):
def setUp(self):
from twisted.trial.test import sample
self.stream = StringIO.StringIO()
self.test = sample.PyunitTest('test_foo')
def test_verboseReporter(self):
result = reporter.VerboseTextReporter(self.stream)
result.startTest(self.test)
output = self.stream.getvalue()
self.failUnlessEqual(
output, 'twisted.trial.test.sample.PyunitTest.test_foo ... ')
def test_treeReporter(self):
result = reporter.TreeReporter(self.stream)
result.startTest(self.test)
output = self.stream.getvalue()
output = output.splitlines()[-1].strip()
self.failUnlessEqual(output, result.getDescription(self.test) + ' ...')
def test_getDescription(self):
result = reporter.TreeReporter(self.stream)
output = result.getDescription(self.test)
self.failUnlessEqual(output, 'test_foo')
def test_minimalReporter(self):
"""
The summary of L{reporter.MinimalReporter} is a simple list of numbers,
indicating how many tests ran, how many failed etc.
The numbers represents:
* the run time of the tests
* the number of tests run, printed 2 times for legacy reasons
* the number of errors
* the number of failures
* the number of skips
"""
result = reporter.MinimalReporter(self.stream)
self.test.run(result)
result._printSummary()
output = self.stream.getvalue().strip().split(' ')
self.failUnlessEqual(output[1:], ['1', '1', '0', '0', '0'])
def test_minimalReporterTime(self):
"""
L{reporter.MinimalReporter} reports the time to run the tests as first
data in its output.
"""
times = [1.0, 1.2, 1.5, 1.9]
result = reporter.MinimalReporter(self.stream)
result._getTime = lambda: times.pop(0)
self.test.run(result)
result._printSummary()
output = self.stream.getvalue().strip().split(' ')
timer = output[0]
self.assertEquals(timer, "0.7")
def test_emptyMinimalReporter(self):
"""
The summary of L{reporter.MinimalReporter} is a list of zeroes when no
test is actually run.
"""
result = reporter.MinimalReporter(self.stream)
result._printSummary()
output = self.stream.getvalue().strip().split(' ')
self.failUnlessEqual(output, ['0', '0', '0', '0', '0', '0'])
class TestDirtyReactor(unittest.TestCase):
"""
The trial script has an option to treat L{DirtyReactorAggregateError}s as
warnings, as a migration tool for test authors. It causes a wrapper to be
placed around reporters that replaces L{DirtyReactorAggregatErrors} with
warnings.
"""
def setUp(self):
self.dirtyError = Failure(
util.DirtyReactorAggregateError(['foo'], ['bar']))
self.output = StringIO.StringIO()
self.test = TestDirtyReactor('test_errorByDefault')
def test_errorByDefault(self):
"""
L{DirtyReactorAggregateError}s are reported as errors with the default
Reporter.
"""
result = reporter.Reporter(stream=self.output)
result.addError(self.test, self.dirtyError)
self.assertEqual(len(result.errors), 1)
self.assertEqual(result.errors[0][1], self.dirtyError)
def test_warningsEnabled(self):
"""
L{DirtyReactorAggregateError}s are reported as warnings when using
the L{UncleanWarningsReporterWrapper}.
"""
result = UncleanWarningsReporterWrapper(
reporter.Reporter(stream=self.output))
self.assertWarns(UserWarning, self.dirtyError.getErrorMessage(),
reporter.__file__,
result.addError, self.test, self.dirtyError)
def test_warningsMaskErrors(self):
"""
L{DirtyReactorAggregateError}s are I{not} reported as errors if the
L{UncleanWarningsReporterWrapper} is used.
"""
result = UncleanWarningsReporterWrapper(
reporter.Reporter(stream=self.output))
self.assertWarns(UserWarning, self.dirtyError.getErrorMessage(),
reporter.__file__,
result.addError, self.test, self.dirtyError)
self.assertEquals(result._originalReporter.errors, [])
def test_dealsWithThreeTuples(self):
"""
Some annoying stuff can pass three-tuples to addError instead of
Failures (like PyUnit). The wrapper, of course, handles this case,
since it is a part of L{twisted.trial.itrial.IReporter}! But it does
not convert L{DirtyReactorAggregateError} to warnings in this case,
because nobody should be passing those in the form of three-tuples.
"""
result = UncleanWarningsReporterWrapper(
reporter.Reporter(stream=self.output))
result.addError(self.test,
(self.dirtyError.type, self.dirtyError.value, None))
self.assertEqual(len(result._originalReporter.errors), 1)
self.assertEquals(result._originalReporter.errors[0][1].type,
self.dirtyError.type)
self.assertEquals(result._originalReporter.errors[0][1].value,
self.dirtyError.value)
class TrialTestNames(unittest.TestCase):
def setUp(self):
from twisted.trial.test import sample
self.stream = StringIO.StringIO()
self.test = sample.FooTest('test_foo')
def test_verboseReporter(self):
result = reporter.VerboseTextReporter(self.stream)
result.startTest(self.test)
output = self.stream.getvalue()
self.failUnlessEqual(output, self.test.id() + ' ... ')
def test_treeReporter(self):
result = reporter.TreeReporter(self.stream)
result.startTest(self.test)
output = self.stream.getvalue()
output = output.splitlines()[-1].strip()
self.failUnlessEqual(output, result.getDescription(self.test) + ' ...')
def test_treeReporterWithDocstrings(self):
"""A docstring"""
result = reporter.TreeReporter(self.stream)
self.assertEqual(result.getDescription(self),
'test_treeReporterWithDocstrings')
def test_getDescription(self):
result = reporter.TreeReporter(self.stream)
output = result.getDescription(self.test)
self.failUnlessEqual(output, "test_foo")
class TestSkip(unittest.TestCase):
"""
Tests for L{reporter.Reporter}'s handling of skips.
"""
def setUp(self):
from twisted.trial.test import sample
self.stream = StringIO.StringIO()
self.result = reporter.Reporter(self.stream)
self.test = sample.FooTest('test_foo')
def _getSkips(self, result):
"""
Get the number of skips that happened to a reporter.
"""
return len(result.skips)
def test_accumulation(self):
self.result.addSkip(self.test, 'some reason')
self.assertEqual(self._getSkips(self.result), 1)
def test_success(self):
self.result.addSkip(self.test, 'some reason')
self.failUnlessEqual(True, self.result.wasSuccessful())
def test_summary(self):
"""
The summary of a successful run with skips indicates that the test
suite passed and includes the number of skips.
"""
self.result.addSkip(self.test, 'some reason')
self.result.done()
output = self.stream.getvalue().splitlines()[-1]
prefix = 'PASSED '
self.failUnless(output.startswith(prefix))
self.failUnlessEqual(output[len(prefix):].strip(), '(skips=1)')
def test_basicErrors(self):
"""
The output at the end of a test run with skips includes the reasons
for skipping those tests.
"""
self.result.addSkip(self.test, 'some reason')
self.result.done()
output = self.stream.getvalue().splitlines()[4]
self.failUnlessEqual(output.strip(), 'some reason')
def test_booleanSkip(self):
"""
Tests can be skipped without specifying a reason by setting the 'skip'
attribute to True. When this happens, the test output includes 'True'
as the reason.
"""
self.result.addSkip(self.test, True)
self.result.done()
output = self.stream.getvalue().splitlines()[4]
self.failUnlessEqual(output, 'True')
def test_exceptionSkip(self):
"""
Skips can be raised as errors. When this happens, the error is
included in the summary at the end of the test suite.
"""
try:
1/0
except Exception, e:
error = e
self.result.addSkip(self.test, error)
self.result.done()
output = '\n'.join(self.stream.getvalue().splitlines()[3:5]).strip()
self.failUnlessEqual(output, str(e))
class UncleanWarningSkipTest(TestSkip):
"""
Tests for skips on a L{reporter.Reporter} wrapped by an
L{UncleanWarningsReporterWrapper}.
"""
def setUp(self):
TestSkip.setUp(self)
self.result = UncleanWarningsReporterWrapper(self.result)
def _getSkips(self, result):
"""
Get the number of skips that happened to a reporter inside of an
unclean warnings reporter wrapper.
"""
return len(result._originalReporter.skips)
class TodoTest(unittest.TestCase):
"""
Tests for L{reporter.Reporter}'s handling of todos.
"""
def setUp(self):
from twisted.trial.test import sample
self.stream = StringIO.StringIO()
self.result = reporter.Reporter(self.stream)
self.test = sample.FooTest('test_foo')
def _getTodos(self, result):
"""
Get the number of todos that happened to a reporter.
"""
return len(result.expectedFailures)
def _getUnexpectedSuccesses(self, result):
"""
Get the number of unexpected successes that happened to a reporter.
"""
return len(result.unexpectedSuccesses)
def test_accumulation(self):
"""
L{reporter.Reporter} accumulates the expected failures that it
is notified of.
"""
self.result.addExpectedFailure(self.test, Failure(Exception()),
makeTodo('todo!'))
self.assertEqual(self._getTodos(self.result), 1)
def test_success(self):
"""
A test run is still successful even if there are expected failures.
"""
self.result.addExpectedFailure(self.test, Failure(Exception()),
makeTodo('todo!'))
self.assertEqual(True, self.result.wasSuccessful())
def test_unexpectedSuccess(self):
"""
A test which is marked as todo but succeeds will have an unexpected
success reported to its result. A test run is still successful even
when this happens.
"""
self.result.addUnexpectedSuccess(self.test, makeTodo("Heya!"))
self.assertEqual(True, self.result.wasSuccessful())
self.assertEqual(self._getUnexpectedSuccesses(self.result), 1)
def test_summary(self):
"""
The reporter's C{printSummary} method should print the number of
expected failures that occured.
"""
self.result.addExpectedFailure(self.test, Failure(Exception()),
makeTodo('some reason'))
self.result.done()
output = self.stream.getvalue().splitlines()[-1]
prefix = 'PASSED '
self.failUnless(output.startswith(prefix))
self.assertEqual(output[len(prefix):].strip(),
'(expectedFailures=1)')
def test_basicErrors(self):
"""
The reporter's L{printErrors} method should include the value of the
Todo.
"""
self.result.addExpectedFailure(self.test, Failure(Exception()),
makeTodo('some reason'))
self.result.done()
output = self.stream.getvalue().splitlines()[4].strip()
self.assertEqual(output, "Reason: 'some reason'")
def test_booleanTodo(self):
"""
Booleans CAN'T be used as the value of a todo. Maybe this sucks. This
is a test for current behavior, not a requirement.
"""
self.result.addExpectedFailure(self.test, Failure(Exception()),
makeTodo(True))
self.assertRaises(Exception, self.result.done)
def test_exceptionTodo(self):
"""
The exception for expected failures should be shown in the
C{printErrors} output.
"""
try:
1/0
except Exception, e:
error = e
self.result.addExpectedFailure(self.test, Failure(error),
makeTodo("todo!"))
self.result.done()
output = '\n'.join(self.stream.getvalue().splitlines()[3:]).strip()
self.assertTrue(str(e) in output)
class UncleanWarningTodoTest(TodoTest):
"""
Tests for L{UncleanWarningsReporterWrapper}'s handling of todos.
"""
def setUp(self):
TodoTest.setUp(self)
self.result = UncleanWarningsReporterWrapper(self.result)
def _getTodos(self, result):
"""
Get the number of todos that happened to a reporter inside of an
unclean warnings reporter wrapper.
"""
return len(result._originalReporter.expectedFailures)
def _getUnexpectedSuccesses(self, result):
"""
Get the number of unexpected successes that happened to a reporter
inside of an unclean warnings reporter wrapper.
"""
return len(result._originalReporter.unexpectedSuccesses)
class MockColorizer:
"""
Used by TestTreeReporter to make sure that output is colored correctly.
"""
def __init__(self, stream):
self.log = []
def write(self, text, color):
self.log.append((color, text))
class TestTreeReporter(unittest.TestCase):
def setUp(self):
from twisted.trial.test import sample
self.test = sample.FooTest('test_foo')
self.stream = StringIO.StringIO()
self.result = reporter.TreeReporter(self.stream)
self.result._colorizer = MockColorizer(self.stream)
self.log = self.result._colorizer.log
def makeError(self):
try:
1/0
except ZeroDivisionError:
f = Failure()
return f
def test_cleanupError(self):
"""
Run cleanupErrors and check that the output is correct, and colored
correctly.
"""
f = self.makeError()
self.result.cleanupErrors(f)
color, text = self.log[0]
self.assertEqual(color.strip(), self.result.ERROR)
self.assertEqual(text.strip(), 'cleanup errors')
color, text = self.log[1]
self.assertEqual(color.strip(), self.result.ERROR)
self.assertEqual(text.strip(), '[ERROR]')
test_cleanupError = suppressWarnings(
test_cleanupError,
util.suppress(category=reporter.BrokenTestCaseWarning),
util.suppress(category=DeprecationWarning))
def test_upDownError(self):
"""
Run upDownError and check that the output is correct and colored
correctly.
"""
self.result.upDownError("method", None, None, False)
color, text = self.log[0]
self.assertEqual(color.strip(), self.result.ERROR)
self.assertEqual(text.strip(), 'method')
test_upDownError = suppressWarnings(
test_upDownError,
util.suppress(category=DeprecationWarning,
message="upDownError is deprecated in Twisted 8.0."))
def test_summaryColoredSuccess(self):
"""
The summary in case of success should have a good count of successes
and be colored properly.
"""
self.result.addSuccess(self.test)
self.result.done()
self.assertEquals(self.log[1], (self.result.SUCCESS, 'PASSED'))
self.assertEquals(
self.stream.getvalue().splitlines()[-1].strip(), "(successes=1)")
def test_summaryColoredFailure(self):
"""
The summary in case of failure should have a good count of errors
and be colored properly.
"""
try:
raise RuntimeError('foo')
except RuntimeError, excValue:
self.result.addError(self, sys.exc_info())
self.result.done()
self.assertEquals(self.log[1], (self.result.FAILURE, 'FAILED'))
self.assertEquals(
self.stream.getvalue().splitlines()[-1].strip(), "(errors=1)")
def test_getPrelude(self):
"""
The tree needs to get the segments of the test ID that correspond
to the module and class that it belongs to.
"""
self.assertEqual(
['foo.bar', 'baz'],
self.result._getPreludeSegments('foo.bar.baz.qux'))
self.assertEqual(
['foo', 'bar'],
self.result._getPreludeSegments('foo.bar.baz'))
self.assertEqual(
['foo'],
self.result._getPreludeSegments('foo.bar'))
self.assertEqual([], self.result._getPreludeSegments('foo'))
class TestReporterInterface(unittest.TestCase):
"""
Tests for the bare interface of a trial reporter.
Subclass this test case and provide a different 'resultFactory' to test
that a particular reporter implementation will work with the rest of
Trial.
@cvar resultFactory: A callable that returns a reporter to be tested. The
callable must take the same parameters as L{reporter.Reporter}.
"""
resultFactory = reporter.Reporter
def setUp(self):
from twisted.trial.test import sample
self.test = sample.FooTest('test_foo')
self.stream = StringIO.StringIO()
self.publisher = log.LogPublisher()
self.result = self.resultFactory(self.stream, publisher=self.publisher)
def test_shouldStopInitiallyFalse(self):
"""
shouldStop is False to begin with.
"""
self.assertEquals(False, self.result.shouldStop)
def test_shouldStopTrueAfterStop(self):
"""
shouldStop becomes True soon as someone calls stop().
"""
self.result.stop()
self.assertEquals(True, self.result.shouldStop)
def test_wasSuccessfulInitiallyTrue(self):
"""
wasSuccessful() is True when there have been no results reported.
"""
self.assertEquals(True, self.result.wasSuccessful())
def test_wasSuccessfulTrueAfterSuccesses(self):
"""
wasSuccessful() is True when there have been only successes, False
otherwise.
"""
self.result.addSuccess(self.test)
self.assertEquals(True, self.result.wasSuccessful())
def test_wasSuccessfulFalseAfterErrors(self):
"""
wasSuccessful() becomes False after errors have been reported.
"""
try:
1 / 0
except ZeroDivisionError:
self.result.addError(self.test, sys.exc_info())
self.assertEquals(False, self.result.wasSuccessful())
def test_wasSuccessfulFalseAfterFailures(self):
"""
wasSuccessful() becomes False after failures have been reported.
"""
try:
self.fail("foo")
except self.failureException:
self.result.addFailure(self.test, sys.exc_info())
self.assertEquals(False, self.result.wasSuccessful())
class TestReporter(TestReporterInterface):
"""
Tests for the base L{reporter.Reporter} class.
"""
def setUp(self):
TestReporterInterface.setUp(self)
self._timer = 0
self.result._getTime = self._getTime
def _getTime(self):
self._timer += 1
return self._timer
def test_startStop(self):
self.result.startTest(self.test)
self.result.stopTest(self.test)
self.assertTrue(self.result._lastTime > 0)
self.assertEqual(self.result.testsRun, 1)
self.assertEqual(self.result.wasSuccessful(), True)
def test_brokenStream(self):
"""
Test that the reporter safely writes to its stream.
"""
result = self.resultFactory(stream=BrokenStream(self.stream))
result._writeln("Hello")
self.assertEqual(self.stream.getvalue(), 'Hello\n')
self.stream.truncate(0)
result._writeln("Hello %s!", 'World')
self.assertEqual(self.stream.getvalue(), 'Hello World!\n')
def test_printErrorsDeprecated(self):
"""
L{IReporter.printErrors} was deprecated in Twisted 8.0.
"""
def f():
self.result.printErrors()
self.assertWarns(
DeprecationWarning, "printErrors is deprecated in Twisted 8.0.",
__file__, f)
def test_printSummaryDeprecated(self):
"""
L{IReporter.printSummary} was deprecated in Twisted 8.0.
"""
def f():
self.result.printSummary()
self.assertWarns(
DeprecationWarning, "printSummary is deprecated in Twisted 8.0.",
__file__, f)
def test_writeDeprecated(self):
"""
L{IReporter.write} was deprecated in Twisted 8.0.
"""
def f():
self.result.write("")
self.assertWarns(
DeprecationWarning, "write is deprecated in Twisted 8.0.",
__file__, f)
def test_writelnDeprecated(self):
"""
L{IReporter.writeln} was deprecated in Twisted 8.0.
"""
def f():
self.result.writeln("")
self.assertWarns(
DeprecationWarning, "writeln is deprecated in Twisted 8.0.",
__file__, f)
def test_separatorDeprecated(self):
"""
L{IReporter.separator} was deprecated in Twisted 8.0.
"""
def f():
return self.result.separator
self.assertWarns(
DeprecationWarning, "separator is deprecated in Twisted 8.0.",
__file__, f)
def test_streamDeprecated(self):
"""
L{IReporter.stream} was deprecated in Twisted 8.0.
"""
def f():
return self.result.stream
self.assertWarns(
DeprecationWarning, "stream is deprecated in Twisted 8.0.",
__file__, f)
def test_upDownErrorDeprecated(self):
"""
L{IReporter.upDownError} was deprecated in Twisted 8.0.
"""
def f():
self.result.upDownError(None, None, None, None)
self.assertWarns(
DeprecationWarning, "upDownError is deprecated in Twisted 8.0.",
__file__, f)
def test_warning(self):
"""
L{reporter.Reporter} observes warnings emitted by the Twisted log
system and writes them to its output stream.
"""
message = RuntimeWarning("some warning text")
category = 'exceptions.RuntimeWarning'
filename = "path/to/some/file.py"
lineno = 71
self.publisher.msg(
warning=message, category=category,
filename=filename, lineno=lineno)
self.assertEqual(
self.stream.getvalue(),
"%s:%d: %s: %s\n" % (
filename, lineno, category.split('.')[-1], message))
def test_duplicateWarningSuppressed(self):
"""
A warning emitted twice within a single test is only written to the
stream once.
"""
# Emit the warning and assert that it shows up
self.test_warning()
# Emit the warning again and assert that the stream still only has one
# warning on it.
self.test_warning()
def test_warningEmittedForNewTest(self):
"""
A warning emitted again after a new test has started is written to the
stream again.
"""
test = self.__class__('test_warningEmittedForNewTest')
self.result.startTest(test)
# Clear whatever startTest wrote to the stream
self.stream.seek(0)
self.stream.truncate()
# Emit a warning (and incidentally, assert that it was emitted)
self.test_warning()
# Clean up from the first warning to simplify the rest of the
# assertions.
self.stream.seek(0)
self.stream.truncate()
# Stop the first test and start another one (it just happens to be the
# same one, but that doesn't matter)
self.result.stopTest(test)
self.result.startTest(test)
# Clean up the stopTest/startTest output
self.stream.seek(0)
self.stream.truncate()
# Emit the warning again and make sure it shows up
self.test_warning()
def test_stopObserving(self):
"""
L{reporter.Reporter} stops observing log events when its C{done} method
is called.
"""
self.result.done()
self.stream.seek(0)
self.stream.truncate()
self.publisher.msg(
warning=RuntimeWarning("some message"),
category='exceptions.RuntimeWarning',
filename="file/name.py", lineno=17)
self.assertEqual(self.stream.getvalue(), "")
class TestSafeStream(unittest.TestCase):
def test_safe(self):
"""
Test that L{reporter.SafeStream} successfully write to its original
stream even if an interrupt happens during the write.
"""
stream = StringIO.StringIO()
broken = BrokenStream(stream)
safe = reporter.SafeStream(broken)
safe.write("Hello")
self.assertEqual(stream.getvalue(), "Hello")
class TestSubunitReporter(TestReporterInterface):
"""
Tests for the subunit reporter.
This just tests that the subunit reporter implements the basic interface.
"""
resultFactory = reporter.SubunitReporter
def setUp(self):
if reporter.TestProtocolClient is None:
raise SkipTest(
"Subunit not installed, cannot test SubunitReporter")
TestReporterInterface.setUp(self)
def assertForwardsToSubunit(self, methodName, *args, **kwargs):
"""
Assert that 'methodName' on L{SubunitReporter} forwards to the
equivalent method on subunit.
Checks that the return value from subunit is returned from the
L{SubunitReporter} and that the reporter writes the same data to its
stream as subunit does to its own.
Assumes that the method on subunit has the same name as the method on
L{SubunitReporter}.
"""
stream = StringIO.StringIO()
subunitClient = reporter.TestProtocolClient(stream)
subunitReturn = getattr(subunitClient, methodName)(*args, **kwargs)
subunitOutput = stream.getvalue()
reporterReturn = getattr(self.result, methodName)(*args, **kwargs)
self.assertEquals(subunitReturn, reporterReturn)
self.assertEquals(subunitOutput, self.stream.getvalue())
def test_subunitWithoutAddExpectedFailureInstalled(self):
"""
Some versions of subunit don't have "addExpectedFailure". For these
versions, we report expected failures as successes.
"""
try:
addExpectedFailure = reporter.TestProtocolClient.addExpectedFailure
except AttributeError:
# Then we've actually got one of those old versions installed, and
# the test is immediately applicable.
pass
else:
del reporter.TestProtocolClient.addExpectedFailure
self.addCleanup(
setattr, reporter.TestProtocolClient, 'addExpectedFailure',
addExpectedFailure)
try:
1 / 0
except ZeroDivisionError:
self.result.addExpectedFailure(self.test, sys.exc_info(), "todo")
expectedFailureOutput = self.stream.getvalue()
self.stream.truncate(0)
self.result.addSuccess(self.test)
successOutput = self.stream.getvalue()
self.assertEquals(successOutput, expectedFailureOutput)
def test_subunitWithoutAddSkipInstalled(self):
"""
Some versions of subunit don't have "addSkip". For these versions, we
report skips as successes.
"""
try:
addSkip = reporter.TestProtocolClient.addSkip
except AttributeError:
# Then we've actually got one of those old versions installed, and
# the test is immediately applicable.
pass
else:
del reporter.TestProtocolClient.addSkip
self.addCleanup(
setattr, reporter.TestProtocolClient, 'addSkip', addSkip)
self.result.addSkip(self.test, "reason")
skipOutput = self.stream.getvalue()
self.stream.truncate(0)
self.result.addSuccess(self.test)
successOutput = self.stream.getvalue()
self.assertEquals(successOutput, skipOutput)
def test_addExpectedFailurePassedThrough(self):
"""
Some versions of subunit have "addExpectedFailure". For these
versions, when we call 'addExpectedFailure' on the test result, we
pass the error and test through to the subunit client.
"""
addExpectedFailureCalls = []
def addExpectedFailure(test, error):
addExpectedFailureCalls.append((test, error))
# Provide our own addExpectedFailure, whether or not the locally
# installed subunit has addExpectedFailure.
self.result._subunit.addExpectedFailure = addExpectedFailure
try:
1 / 0
except ZeroDivisionError:
exc_info = sys.exc_info()
self.result.addExpectedFailure(self.test, exc_info, 'todo')
self.assertEquals(addExpectedFailureCalls, [(self.test, exc_info)])
def test_addSkipSendsSubunitAddSkip(self):
"""
Some versions of subunit have "addSkip". For these versions, when we
call 'addSkip' on the test result, we pass the test and reason through
to the subunit client.
"""
addSkipCalls = []
def addSkip(test, reason):
addSkipCalls.append((test, reason))
# Provide our own addSkip, whether or not the locally-installed
# subunit has addSkip.
self.result._subunit.addSkip = addSkip
self.result.addSkip(self.test, 'reason')
self.assertEquals(addSkipCalls, [(self.test, 'reason')])
def test_doneDoesNothing(self):
"""
The subunit reporter doesn't need to print out a summary -- the stream
of results is everything. Thus, done() does nothing.
"""
self.result.done()
self.assertEquals('', self.stream.getvalue())
def test_startTestSendsSubunitStartTest(self):
"""
SubunitReporter.startTest() sends the subunit 'startTest' message.
"""
self.assertForwardsToSubunit('startTest', self.test)
def test_stopTestSendsSubunitStopTest(self):
"""
SubunitReporter.stopTest() sends the subunit 'stopTest' message.
"""
self.assertForwardsToSubunit('stopTest', self.test)
def test_addSuccessSendsSubunitAddSuccess(self):
"""
SubunitReporter.addSuccess() sends the subunit 'addSuccess' message.
"""
self.assertForwardsToSubunit('addSuccess', self.test)
def test_addErrorSendsSubunitAddError(self):
"""
SubunitReporter.addError() sends the subunit 'addError' message.
"""
try:
1 / 0
except ZeroDivisionError:
error = sys.exc_info()
self.assertForwardsToSubunit('addError', self.test, error)
def test_addFailureSendsSubunitAddFailure(self):
"""
SubunitReporter.addFailure() sends the subunit 'addFailure' message.
"""
try:
self.fail('hello')
except self.failureException:
failure = sys.exc_info()
self.assertForwardsToSubunit('addFailure', self.test, failure)
def test_addUnexpectedSuccessSendsSubunitAddSuccess(self):
"""
SubunitReporter.addFailure() sends the subunit 'addSuccess' message,
since subunit doesn't model unexpected success.
"""
stream = StringIO.StringIO()
subunitClient = reporter.TestProtocolClient(stream)
subunitClient.addSuccess(self.test)
subunitOutput = stream.getvalue()
self.result.addUnexpectedSuccess(self.test, 'todo')
self.assertEquals(subunitOutput, self.stream.getvalue())
def test_loadTimeErrors(self):
"""
Load-time errors are reported like normal errors.
"""
test = runner.TestLoader().loadByName('doesntexist')
test.run(self.result)
output = self.stream.getvalue()
# Just check that 'doesntexist' is in the output, rather than
# assembling the expected stack trace.
self.assertIn('doesntexist', output)
class TestSubunitReporterNotInstalled(unittest.TestCase):
"""
Test behaviour when the subunit reporter is not installed.
"""
def test_subunitNotInstalled(self):
"""
If subunit is not installed, TestProtocolClient will be None, and
SubunitReporter will raise an error when you try to construct it.
"""
stream = StringIO.StringIO()
self.patch(reporter, 'TestProtocolClient', None)
e = self.assertRaises(Exception, reporter.SubunitReporter, stream)
self.assertEquals("Subunit not available", str(e))
class TestTimingReporter(TestReporter):
resultFactory = reporter.TimingTextReporter
class LoggingReporter(reporter.Reporter):
"""
Simple reporter that stores the last test that was passed to it.
"""
def __init__(self, *args, **kwargs):
reporter.Reporter.__init__(self, *args, **kwargs)
self.test = None
def addError(self, test, error):
self.test = test
def addExpectedFailure(self, test, failure, todo):
self.test = test
def addFailure(self, test, failure):
self.test = test
def addSkip(self, test, skip):
self.test = test
def addUnexpectedSuccess(self, test, todo):
self.test = test
def startTest(self, test):
self.test = test
def stopTest(self, test):
self.test = test
class TestAdaptedReporter(unittest.TestCase):
"""
L{reporter._AdaptedReporter} is a reporter wrapper that wraps all of the
tests it receives before passing them on to the original reporter.
"""
def setUp(self):
self.wrappedResult = self.getWrappedResult()
def _testAdapter(self, test):
return test.id()
def assertWrapped(self, wrappedResult, test):
self.assertEqual(wrappedResult._originalReporter.test, self._testAdapter(test))
def getFailure(self, exceptionInstance):
"""
Return a L{Failure} from raising the given exception.
@param exceptionInstance: The exception to raise.
@return: L{Failure}
"""
try:
raise exceptionInstance
except:
return Failure()
def getWrappedResult(self):
result = LoggingReporter()
return reporter._AdaptedReporter(result, self._testAdapter)
def test_addError(self):
"""
C{addError} wraps its test with the provided adapter.
"""
self.wrappedResult.addError(self, self.getFailure(RuntimeError()))
self.assertWrapped(self.wrappedResult, self)
def test_addFailure(self):
"""
C{addFailure} wraps its test with the provided adapter.
"""
self.wrappedResult.addFailure(self, self.getFailure(AssertionError()))
self.assertWrapped(self.wrappedResult, self)
def test_addSkip(self):
"""
C{addSkip} wraps its test with the provided adapter.
"""
self.wrappedResult.addSkip(self, self.getFailure(SkipTest('no reason')))
self.assertWrapped(self.wrappedResult, self)
def test_startTest(self):
"""
C{startTest} wraps its test with the provided adapter.
"""
self.wrappedResult.startTest(self)
self.assertWrapped(self.wrappedResult, self)
def test_stopTest(self):
"""
C{stopTest} wraps its test with the provided adapter.
"""
self.wrappedResult.stopTest(self)
self.assertWrapped(self.wrappedResult, self)
def test_addExpectedFailure(self):
"""
C{addExpectedFailure} wraps its test with the provided adapter.
"""
self.wrappedResult.addExpectedFailure(
self, self.getFailure(RuntimeError()), Todo("no reason"))
self.assertWrapped(self.wrappedResult, self)
def test_addUnexpectedSuccess(self):
"""
C{addUnexpectedSuccess} wraps its test with the provided adapter.
"""
self.wrappedResult.addUnexpectedSuccess(self, Todo("no reason"))
self.assertWrapped(self.wrappedResult, self)
class FakeStream(object):
"""
A fake stream which C{isatty} method returns some predictable.
@ivar tty: returned value of C{isatty}.
@type tty: C{bool}
"""
def __init__(self, tty=True):
self.tty = tty
def isatty(self):
return self.tty
class AnsiColorizerTests(unittest.TestCase):
"""
Tests for L{reporter._AnsiColorizer}.
"""
def setUp(self):
self.savedModules = sys.modules.copy()
def tearDown(self):
sys.modules.clear()
sys.modules.update(self.savedModules)
def test_supportedStdOutTTY(self):
"""
L{reporter._AnsiColorizer.supported} returns C{False} if the given
stream is not a TTY.
"""
self.assertFalse(reporter._AnsiColorizer.supported(FakeStream(False)))
def test_supportedNoCurses(self):
"""
L{reporter._AnsiColorizer.supported} returns C{False} if the curses
module can't be imported.
"""
sys.modules['curses'] = None
self.assertFalse(reporter._AnsiColorizer.supported(FakeStream()))
def test_supportedSetupTerm(self):
"""
L{reporter._AnsiColorizer.supported} returns C{True} if
C{curses.tigetnum} returns more than 2 supported colors. It only tries
to call C{curses.setupterm} if C{curses.tigetnum} previously failed
with a C{curses.error}.
"""
class fakecurses(object):
error = RuntimeError
setUp = 0
def setupterm(self):
self.setUp += 1
def tigetnum(self, value):
if self.setUp:
return 3
else:
raise self.error()
sys.modules['curses'] = fakecurses()
self.assertTrue(reporter._AnsiColorizer.supported(FakeStream()))
self.assertTrue(reporter._AnsiColorizer.supported(FakeStream()))
self.assertEquals(sys.modules['curses'].setUp, 1)
def test_supportedTigetNumWrongError(self):
"""
L{reporter._AnsiColorizer.supported} returns C{False} and doesn't try
to call C{curses.setupterm} if C{curses.tigetnum} returns something
different than C{curses.error}.
"""
class fakecurses(object):
error = RuntimeError
def tigetnum(self, value):
raise ValueError()
sys.modules['curses'] = fakecurses()
self.assertFalse(reporter._AnsiColorizer.supported(FakeStream()))
def test_supportedTigetNumNotEnoughColor(self):
"""
L{reporter._AnsiColorizer.supported} returns C{False} if
C{curses.tigetnum} returns less than 2 supported colors.
"""
class fakecurses(object):
error = RuntimeError
def tigetnum(self, value):
return 1
sys.modules['curses'] = fakecurses()
self.assertFalse(reporter._AnsiColorizer.supported(FakeStream()))
def test_supportedTigetNumErrors(self):
"""
L{reporter._AnsiColorizer.supported} returns C{False} if
C{curses.tigetnum} raises an error, and calls C{curses.setupterm} once.
"""
class fakecurses(object):
error = RuntimeError
setUp = 0
def setupterm(self):
self.setUp += 1
def tigetnum(self, value):
raise self.error()
sys.modules['curses'] = fakecurses()
self.assertFalse(reporter._AnsiColorizer.supported(FakeStream()))
self.assertEquals(sys.modules['curses'].setUp, 1)
|