1
# Copyright (c) 2001-2009 Twisted Matrix Laboratories.
2
# See LICENSE for details.
4
# Maintainer: Jonathan Lange
7
Tests for L{twisted.trial.reporter}.
11
import errno, sys, os, re, StringIO
13
from twisted.internet.utils import suppressWarnings
14
from twisted.python import log
15
from twisted.python.failure import Failure
16
from twisted.trial import itrial, unittest, runner, reporter, util
17
from twisted.trial.reporter import UncleanWarningsReporterWrapper
18
from twisted.trial.test import erroneous
19
from twisted.trial.unittest import makeTodo, SkipTest, Todo
22
class BrokenStream(object):
24
Stream-ish object that raises a signal interrupt error. We use this to make
25
sure that Trial still manages to write what it needs to write.
30
def __init__(self, fObj):
35
return self.fObj.write(s)
37
raise IOError(errno.EINTR, "Interrupted write")
41
return self.fObj.flush()
43
raise IOError(errno.EINTR, "Interrupted flush")
46
class StringTest(unittest.TestCase):
47
def stringComparison(self, expect, output):
48
output = filter(None, output)
49
self.failUnless(len(expect) <= len(output),
50
"Must have more observed than expected"
51
"lines %d < %d" % (len(output), len(expect)))
52
REGEX_PATTERN_TYPE = type(re.compile(''))
53
for line_number, (exp, out) in enumerate(zip(expect, output)):
56
elif isinstance(exp, str):
57
self.assertSubstring(exp, out, "Line %d: %r not in %r"
58
% (line_number, exp, out))
59
elif isinstance(exp, REGEX_PATTERN_TYPE):
60
self.failUnless(exp.match(out),
61
"Line %d: %r did not match string %r"
62
% (line_number, exp.pattern, out))
64
raise TypeError("don't know what to do with object %r"
68
class TestTestResult(unittest.TestCase):
70
self.result = reporter.TestResult()
72
def test_pyunitAddError(self):
73
# pyunit passes an exc_info tuple directly to addError
75
raise RuntimeError('foo')
76
except RuntimeError, excValue:
77
self.result.addError(self, sys.exc_info())
78
failure = self.result.errors[0][1]
79
self.assertEqual(excValue, failure.value)
80
self.assertEqual(RuntimeError, failure.type)
82
def test_pyunitAddFailure(self):
83
# pyunit passes an exc_info tuple directly to addFailure
85
raise self.failureException('foo')
86
except self.failureException, excValue:
87
self.result.addFailure(self, sys.exc_info())
88
failure = self.result.failures[0][1]
89
self.assertEqual(excValue, failure.value)
90
self.assertEqual(self.failureException, failure.type)
93
class TestReporterRealtime(TestTestResult):
95
output = StringIO.StringIO()
96
self.result = reporter.Reporter(output, realtime=True)
99
class TestErrorReporting(StringTest):
100
doubleSeparator = re.compile(r'^=+$')
103
self.loader = runner.TestLoader()
104
self.output = StringIO.StringIO()
105
self.result = reporter.Reporter(self.output)
107
def getOutput(self, suite):
108
result = self.getResult(suite)
110
return self.output.getvalue()
112
def getResult(self, suite):
113
suite.run(self.result)
116
def testFormatErroredMethod(self):
117
suite = self.loader.loadClass(erroneous.TestFailureInSetUp)
118
output = self.getOutput(suite).splitlines()
119
match = [self.doubleSeparator,
120
('[ERROR]: twisted.trial.test.erroneous.'
121
'TestFailureInSetUp.test_noop'),
122
'Traceback (most recent call last):',
123
re.compile(r'^\s+File .*erroneous\.py., line \d+, in setUp$'),
124
re.compile(r'^\s+raise FoolishError, '
125
r'.I am a broken setUp method.$'),
126
('twisted.trial.test.erroneous.FoolishError: '
127
'I am a broken setUp method')]
128
self.stringComparison(match, output)
130
def testFormatFailedMethod(self):
131
suite = self.loader.loadMethod(erroneous.TestRegularFail.test_fail)
132
output = self.getOutput(suite).splitlines()
134
self.doubleSeparator,
136
'twisted.trial.test.erroneous.TestRegularFail.test_fail',
137
'Traceback (most recent call last):',
138
re.compile(r'^\s+File .*erroneous\.py., line \d+, in test_fail$'),
139
re.compile(r'^\s+self\.fail\("I fail"\)$'),
140
'twisted.trial.unittest.FailTest: I fail'
142
self.stringComparison(match, output)
144
def testDoctestError(self):
145
from twisted.trial.test import erroneous
146
suite = unittest.decorate(
147
self.loader.loadDoctests(erroneous), itrial.ITestCase)
148
output = self.getOutput(suite)
149
path = 'twisted.trial.test.erroneous.unexpectedException'
150
for substring in ['1/0', 'ZeroDivisionError',
151
'Exception raised:', path]:
152
self.assertSubstring(substring, output)
153
self.failUnless(re.search('Fail(ed|ure in) example:', output),
154
"Couldn't match 'Failure in example: ' "
155
"or 'Failed example: '")
156
expect = [self.doubleSeparator,
157
re.compile(r'\[(ERROR|FAIL)\]: .*' + re.escape(path))]
158
self.stringComparison(expect, output.splitlines())
160
def testHiddenException(self):
162
Check that errors in C{DelayedCall}s get reported, even if the
163
test already has a failure.
165
Only really necessary for testing the deprecated style of tests that
166
use iterate() directly. See
167
L{erroneous.DelayedCall.testHiddenException} for more details.
169
test = erroneous.DelayedCall('testHiddenException')
170
output = self.getOutput(test).splitlines()
172
self.doubleSeparator,
174
'twisted.trial.test.erroneous.DelayedCall.testHiddenException',
175
'Traceback (most recent call last):',
176
re.compile(r'^\s+File .*erroneous\.py., line \d+, in '
177
'testHiddenException$'),
178
re.compile(r'^\s+self\.fail\("Deliberate failure to mask the '
179
'hidden exception"\)$'),
180
'twisted.trial.unittest.FailTest: '
181
'Deliberate failure to mask the hidden exception',
182
self.doubleSeparator,
184
'twisted.trial.test.erroneous.DelayedCall.testHiddenException',
185
'Traceback (most recent call last):',
186
re.compile(r'^\s+File .* in runUntilCurrent'),
187
re.compile(r'^\s+.*'),
188
re.compile('^\s+File .*erroneous\.py", line \d+, in go'),
189
re.compile('^\s+raise RuntimeError\(self.hiddenExceptionMsg\)'),
190
'exceptions.RuntimeError: something blew up',
193
self.stringComparison(match, output)
197
class TestUncleanWarningWrapperErrorReporting(TestErrorReporting):
199
Tests that the L{UncleanWarningsReporterWrapper} can sufficiently proxy
200
IReporter failure and error reporting methods to a L{reporter.Reporter}.
203
self.loader = runner.TestLoader()
204
self.output = StringIO.StringIO()
205
self.result = UncleanWarningsReporterWrapper(
206
reporter.Reporter(self.output))
210
class TracebackHandling(unittest.TestCase):
211
def getErrorFrames(self, test):
212
stream = StringIO.StringIO()
213
result = reporter.Reporter(stream)
215
bads = result.failures + result.errors
216
assert len(bads) == 1
217
assert bads[0][0] == test
218
return result._trimFrames(bads[0][1].frames)
220
def checkFrames(self, observedFrames, expectedFrames):
221
for observed, expected in zip(observedFrames, expectedFrames):
222
self.assertEqual(observed[0], expected[0])
223
observedSegs = os.path.splitext(observed[1])[0].split(os.sep)
224
expectedSegs = expected[1].split('/')
225
self.assertEqual(observedSegs[-len(expectedSegs):],
227
self.assertEqual(len(observedFrames), len(expectedFrames))
229
def test_basic(self):
230
test = erroneous.TestRegularFail('test_fail')
231
frames = self.getErrorFrames(test)
232
self.checkFrames(frames,
233
[('test_fail', 'twisted/trial/test/erroneous')])
235
def test_subroutine(self):
236
test = erroneous.TestRegularFail('test_subfail')
237
frames = self.getErrorFrames(test)
238
self.checkFrames(frames,
239
[('test_subfail', 'twisted/trial/test/erroneous'),
240
('subroutine', 'twisted/trial/test/erroneous')])
242
def test_deferred(self):
243
test = erroneous.TestFailureInDeferredChain('test_fail')
244
frames = self.getErrorFrames(test)
245
self.checkFrames(frames,
246
[('_later', 'twisted/trial/test/erroneous')])
248
def test_noFrames(self):
249
result = reporter.Reporter(None)
250
self.assertEqual([], result._trimFrames([]))
252
def test_oneFrame(self):
253
result = reporter.Reporter(None)
254
self.assertEqual(['fake frame'], result._trimFrames(['fake frame']))
257
class FormatFailures(StringTest):
260
raise RuntimeError('foo')
264
['foo', 'foo/bar.py', 5, [('x', 5)], [('y', 'orange')]],
265
['qux', 'foo/bar.py', 10, [('a', 'two')], [('b', 'MCMXCIX')]]
267
self.stream = StringIO.StringIO()
268
self.result = reporter.Reporter(self.stream)
270
def test_formatDefault(self):
271
tb = self.result._formatFailureTraceback(self.f)
272
self.stringComparison([
273
'Traceback (most recent call last):',
274
' File "foo/bar.py", line 5, in foo',
275
re.compile(r'^\s*$'),
276
' File "foo/bar.py", line 10, in qux',
277
re.compile(r'^\s*$'),
278
'RuntimeError: foo'], tb.splitlines())
280
def test_formatString(self):
282
File "twisted/trial/unittest.py", line 256, in failUnlessSubstring
283
return self.failUnlessIn(substring, astring, msg)
284
exceptions.TypeError: iterable argument required
288
File "twisted/trial/unittest.py", line 256, in failUnlessSubstring
289
return self.failUnlessIn(substring, astring, msg)
290
exceptions.TypeError: iterable argument required
292
formatted = self.result._formatFailureTraceback(tb)
293
self.assertEqual(expected, formatted)
295
def test_mutation(self):
296
frames = self.f.frames[:]
297
tb = self.result._formatFailureTraceback(self.f)
298
self.assertEqual(self.f.frames, frames)
301
class PyunitTestNames(unittest.TestCase):
303
from twisted.trial.test import sample
304
self.stream = StringIO.StringIO()
305
self.test = sample.PyunitTest('test_foo')
307
def test_verboseReporter(self):
308
result = reporter.VerboseTextReporter(self.stream)
309
result.startTest(self.test)
310
output = self.stream.getvalue()
311
self.failUnlessEqual(
312
output, 'twisted.trial.test.sample.PyunitTest.test_foo ... ')
314
def test_treeReporter(self):
315
result = reporter.TreeReporter(self.stream)
316
result.startTest(self.test)
317
output = self.stream.getvalue()
318
output = output.splitlines()[-1].strip()
319
self.failUnlessEqual(output, result.getDescription(self.test) + ' ...')
321
def test_getDescription(self):
322
result = reporter.TreeReporter(self.stream)
323
output = result.getDescription(self.test)
324
self.failUnlessEqual(output, 'test_foo')
327
def test_minimalReporter(self):
329
The summary of L{reporter.MinimalReporter} is a simple list of numbers,
330
indicating how many tests ran, how many failed etc.
332
The numbers represents:
333
* the run time of the tests
334
* the number of tests run, printed 2 times for legacy reasons
335
* the number of errors
336
* the number of failures
337
* the number of skips
339
result = reporter.MinimalReporter(self.stream)
340
self.test.run(result)
341
result._printSummary()
342
output = self.stream.getvalue().strip().split(' ')
343
self.failUnlessEqual(output[1:], ['1', '1', '0', '0', '0'])
346
def test_minimalReporterTime(self):
348
L{reporter.MinimalReporter} reports the time to run the tests as first
351
times = [1.0, 1.2, 1.5, 1.9]
352
result = reporter.MinimalReporter(self.stream)
353
result._getTime = lambda: times.pop(0)
354
self.test.run(result)
355
result._printSummary()
356
output = self.stream.getvalue().strip().split(' ')
358
self.assertEquals(timer, "0.7")
361
def test_emptyMinimalReporter(self):
363
The summary of L{reporter.MinimalReporter} is a list of zeroes when no
364
test is actually run.
366
result = reporter.MinimalReporter(self.stream)
367
result._printSummary()
368
output = self.stream.getvalue().strip().split(' ')
369
self.failUnlessEqual(output, ['0', '0', '0', '0', '0', '0'])
373
class TestDirtyReactor(unittest.TestCase):
375
The trial script has an option to treat L{DirtyReactorAggregateError}s as
376
warnings, as a migration tool for test authors. It causes a wrapper to be
377
placed around reporters that replaces L{DirtyReactorAggregatErrors} with
382
self.dirtyError = Failure(
383
util.DirtyReactorAggregateError(['foo'], ['bar']))
384
self.output = StringIO.StringIO()
385
self.test = TestDirtyReactor('test_errorByDefault')
388
def test_errorByDefault(self):
390
C{DirtyReactorAggregateError}s are reported as errors with the default
393
result = reporter.Reporter(stream=self.output)
394
result.addError(self.test, self.dirtyError)
395
self.assertEqual(len(result.errors), 1)
396
self.assertEqual(result.errors[0][1], self.dirtyError)
399
def test_warningsEnabled(self):
401
C{DirtyReactorErrors}s are reported as warnings when using the
402
L{UncleanWarningsReporterWrapper}.
404
result = UncleanWarningsReporterWrapper(
405
reporter.Reporter(stream=self.output))
406
self.assertWarns(UserWarning, self.dirtyError.getErrorMessage(),
408
result.addError, self.test, self.dirtyError)
411
def test_warningsMaskErrors(self):
413
C{DirtyReactorErrors}s are I{not} reported as errors if the
414
L{UncleanWarningsReporterWrapper} is used.
416
result = UncleanWarningsReporterWrapper(
417
reporter.Reporter(stream=self.output))
418
self.assertWarns(UserWarning, self.dirtyError.getErrorMessage(),
420
result.addError, self.test, self.dirtyError)
421
self.assertEquals(result._originalReporter.errors, [])
424
def test_dealsWithThreeTuples(self):
426
Some annoying stuff can pass three-tuples to addError instead of
427
Failures (like PyUnit). The wrapper, of course, handles this case,
428
since it is a part of L{twisted.trial.itrial.IReporter}! But it does
429
not convert L{DirtyReactorError} to warnings in this case, because
430
nobody should be passing those in the form of three-tuples.
432
result = UncleanWarningsReporterWrapper(
433
reporter.Reporter(stream=self.output))
434
result.addError(self.test,
435
(self.dirtyError.type, self.dirtyError.value, None))
436
self.assertEqual(len(result._originalReporter.errors), 1)
437
self.assertEquals(result._originalReporter.errors[0][1].type,
438
self.dirtyError.type)
439
self.assertEquals(result._originalReporter.errors[0][1].value,
440
self.dirtyError.value)
444
class TrialTestNames(unittest.TestCase):
447
from twisted.trial.test import sample
448
self.stream = StringIO.StringIO()
449
self.test = sample.FooTest('test_foo')
451
def test_verboseReporter(self):
452
result = reporter.VerboseTextReporter(self.stream)
453
result.startTest(self.test)
454
output = self.stream.getvalue()
455
self.failUnlessEqual(output, self.test.id() + ' ... ')
457
def test_treeReporter(self):
458
result = reporter.TreeReporter(self.stream)
459
result.startTest(self.test)
460
output = self.stream.getvalue()
461
output = output.splitlines()[-1].strip()
462
self.failUnlessEqual(output, result.getDescription(self.test) + ' ...')
464
def test_treeReporterWithDocstrings(self):
466
result = reporter.TreeReporter(self.stream)
467
self.assertEqual(result.getDescription(self),
468
'test_treeReporterWithDocstrings')
470
def test_getDescription(self):
471
result = reporter.TreeReporter(self.stream)
472
output = result.getDescription(self.test)
473
self.failUnlessEqual(output, "test_foo")
476
class TestSkip(unittest.TestCase):
478
Tests for L{reporter.Reporter}'s handling of skips.
481
from twisted.trial.test import sample
482
self.stream = StringIO.StringIO()
483
self.result = reporter.Reporter(self.stream)
484
self.test = sample.FooTest('test_foo')
486
def _getSkips(self, result):
488
Get the number of skips that happened to a reporter.
490
return len(result.skips)
492
def test_accumulation(self):
493
self.result.addSkip(self.test, 'some reason')
494
self.assertEqual(self._getSkips(self.result), 1)
496
def test_success(self):
497
self.result.addSkip(self.test, 'some reason')
498
self.failUnlessEqual(True, self.result.wasSuccessful())
501
def test_summary(self):
503
The summary of a successful run with skips indicates that the test
504
suite passed and includes the number of skips.
506
self.result.addSkip(self.test, 'some reason')
508
output = self.stream.getvalue().splitlines()[-1]
510
self.failUnless(output.startswith(prefix))
511
self.failUnlessEqual(output[len(prefix):].strip(), '(skips=1)')
514
def test_basicErrors(self):
516
The output at the end of a test run with skips includes the reasons
517
for skipping those tests.
519
self.result.addSkip(self.test, 'some reason')
521
output = self.stream.getvalue().splitlines()[4]
522
self.failUnlessEqual(output.strip(), 'some reason')
525
def test_booleanSkip(self):
527
Tests can be skipped without specifying a reason by setting the 'skip'
528
attribute to True. When this happens, the test output includes 'True'
531
self.result.addSkip(self.test, True)
533
output = self.stream.getvalue().splitlines()[4]
534
self.failUnlessEqual(output, 'True')
537
def test_exceptionSkip(self):
539
Skips can be raised as errors. When this happens, the error is
540
included in the summary at the end of the test suite.
546
self.result.addSkip(self.test, error)
548
output = '\n'.join(self.stream.getvalue().splitlines()[3:5]).strip()
549
self.failUnlessEqual(output, str(e))
552
class UncleanWarningSkipTest(TestSkip):
554
Tests for skips on a L{reporter.Reporter} wrapped by an
555
L{UncleanWarningsReporterWrapper}.
559
self.result = UncleanWarningsReporterWrapper(self.result)
561
def _getSkips(self, result):
563
Get the number of skips that happened to a reporter inside of an
564
unclean warnings reporter wrapper.
566
return len(result._originalReporter.skips)
570
class TodoTest(unittest.TestCase):
572
Tests for L{reporter.Reporter}'s handling of todos.
576
from twisted.trial.test import sample
577
self.stream = StringIO.StringIO()
578
self.result = reporter.Reporter(self.stream)
579
self.test = sample.FooTest('test_foo')
582
def _getTodos(self, result):
584
Get the number of todos that happened to a reporter.
586
return len(result.expectedFailures)
589
def _getUnexpectedSuccesses(self, result):
591
Get the number of unexpected successes that happened to a reporter.
593
return len(result.unexpectedSuccesses)
596
def test_accumulation(self):
598
L{reporter.Reporter} accumulates the expected failures that it
601
self.result.addExpectedFailure(self.test, Failure(Exception()),
603
self.assertEqual(self._getTodos(self.result), 1)
606
def test_success(self):
608
A test run is still successful even if there are expected failures.
610
self.result.addExpectedFailure(self.test, Failure(Exception()),
612
self.assertEqual(True, self.result.wasSuccessful())
615
def test_unexpectedSuccess(self):
617
A test which is marked as todo but succeeds will have an unexpected
618
success reported to its result. A test run is still successful even
621
self.result.addUnexpectedSuccess(self.test, makeTodo("Heya!"))
622
self.assertEqual(True, self.result.wasSuccessful())
623
self.assertEqual(self._getUnexpectedSuccesses(self.result), 1)
626
def test_summary(self):
628
The reporter's C{printSummary} method should print the number of
629
expected failures that occured.
631
self.result.addExpectedFailure(self.test, Failure(Exception()),
632
makeTodo('some reason'))
634
output = self.stream.getvalue().splitlines()[-1]
636
self.failUnless(output.startswith(prefix))
637
self.assertEqual(output[len(prefix):].strip(),
638
'(expectedFailures=1)')
641
def test_basicErrors(self):
643
The reporter's L{printErrors} method should include the value of the
646
self.result.addExpectedFailure(self.test, Failure(Exception()),
647
makeTodo('some reason'))
649
output = self.stream.getvalue().splitlines()[4].strip()
650
self.assertEqual(output, "Reason: 'some reason'")
653
def test_booleanTodo(self):
655
Booleans CAN'T be used as the value of a todo. Maybe this sucks. This
656
is a test for current behavior, not a requirement.
658
self.result.addExpectedFailure(self.test, Failure(Exception()),
660
self.assertRaises(Exception, self.result.done)
663
def test_exceptionTodo(self):
665
The exception for expected failures should be shown in the
666
C{printErrors} output.
672
self.result.addExpectedFailure(self.test, Failure(error),
675
output = '\n'.join(self.stream.getvalue().splitlines()[3:]).strip()
676
self.assertTrue(str(e) in output)
680
class UncleanWarningTodoTest(TodoTest):
682
Tests for L{UncleanWarningsReporterWrapper}'s handling of todos.
687
self.result = UncleanWarningsReporterWrapper(self.result)
690
def _getTodos(self, result):
692
Get the number of todos that happened to a reporter inside of an
693
unclean warnings reporter wrapper.
695
return len(result._originalReporter.expectedFailures)
698
def _getUnexpectedSuccesses(self, result):
700
Get the number of unexpected successes that happened to a reporter
701
inside of an unclean warnings reporter wrapper.
703
return len(result._originalReporter.unexpectedSuccesses)
709
Used by TestTreeReporter to make sure that output is colored correctly.
712
def __init__(self, stream):
716
def write(self, text, color):
717
self.log.append((color, text))
721
class TestTreeReporter(unittest.TestCase):
723
from twisted.trial.test import sample
724
self.test = sample.FooTest('test_foo')
725
self.stream = StringIO.StringIO()
726
self.result = reporter.TreeReporter(self.stream)
727
self.result._colorizer = MockColorizer(self.stream)
728
self.log = self.result._colorizer.log
733
except ZeroDivisionError:
737
def test_cleanupError(self):
739
Run cleanupErrors and check that the output is correct, and colored
743
self.result.cleanupErrors(f)
744
color, text = self.log[0]
745
self.assertEqual(color.strip(), self.result.ERROR)
746
self.assertEqual(text.strip(), 'cleanup errors')
747
color, text = self.log[1]
748
self.assertEqual(color.strip(), self.result.ERROR)
749
self.assertEqual(text.strip(), '[ERROR]')
750
test_cleanupError = suppressWarnings(
752
util.suppress(category=reporter.BrokenTestCaseWarning),
753
util.suppress(category=DeprecationWarning))
756
def test_upDownError(self):
758
Run upDownError and check that the output is correct and colored
761
self.result.upDownError("method", None, None, False)
762
color, text = self.log[0]
763
self.assertEqual(color.strip(), self.result.ERROR)
764
self.assertEqual(text.strip(), 'method')
765
test_upDownError = suppressWarnings(
767
util.suppress(category=DeprecationWarning,
768
message="upDownError is deprecated in Twisted 8.0."))
771
def test_summaryColoredSuccess(self):
773
The summary in case of success should have a good count of successes
774
and be colored properly.
776
self.result.addSuccess(self.test)
778
self.assertEquals(self.log[1], (self.result.SUCCESS, 'PASSED'))
780
self.stream.getvalue().splitlines()[-1].strip(), "(successes=1)")
783
def test_summaryColoredFailure(self):
785
The summary in case of failure should have a good count of errors
786
and be colored properly.
789
raise RuntimeError('foo')
790
except RuntimeError, excValue:
791
self.result.addError(self, sys.exc_info())
793
self.assertEquals(self.log[1], (self.result.FAILURE, 'FAILED'))
795
self.stream.getvalue().splitlines()[-1].strip(), "(errors=1)")
798
def test_getPrelude(self):
800
The tree needs to get the segments of the test ID that correspond
801
to the module and class that it belongs to.
805
self.result._getPreludeSegments('foo.bar.baz.qux'))
808
self.result._getPreludeSegments('foo.bar.baz'))
811
self.result._getPreludeSegments('foo.bar'))
812
self.assertEqual([], self.result._getPreludeSegments('foo'))
816
class TestReporterInterface(unittest.TestCase):
818
Tests for the bare interface of a trial reporter.
820
Subclass this test case and provide a different 'resultFactory' to test
821
that a particular reporter implementation will work with the rest of
824
@cvar resultFactory: A callable that returns a reporter to be tested. The
825
callable must take the same parameters as L{reporter.Reporter}.
828
resultFactory = reporter.Reporter
831
from twisted.trial.test import sample
832
self.test = sample.FooTest('test_foo')
833
self.stream = StringIO.StringIO()
834
self.publisher = log.LogPublisher()
835
self.result = self.resultFactory(self.stream, publisher=self.publisher)
838
def test_shouldStopInitiallyFalse(self):
840
shouldStop is False to begin with.
842
self.assertEquals(False, self.result.shouldStop)
845
def test_shouldStopTrueAfterStop(self):
847
shouldStop becomes True soon as someone calls stop().
850
self.assertEquals(True, self.result.shouldStop)
853
def test_wasSuccessfulInitiallyTrue(self):
855
wasSuccessful() is True when there have been no results reported.
857
self.assertEquals(True, self.result.wasSuccessful())
860
def test_wasSuccessfulTrueAfterSuccesses(self):
862
wasSuccessful() is True when there have been only successes, False
865
self.result.addSuccess(self.test)
866
self.assertEquals(True, self.result.wasSuccessful())
869
def test_wasSuccessfulFalseAfterErrors(self):
871
wasSuccessful() becomes False after errors have been reported.
875
except ZeroDivisionError:
876
self.result.addError(self.test, sys.exc_info())
877
self.assertEquals(False, self.result.wasSuccessful())
880
def test_wasSuccessfulFalseAfterFailures(self):
882
wasSuccessful() becomes False after failures have been reported.
886
except self.failureException:
887
self.result.addFailure(self.test, sys.exc_info())
888
self.assertEquals(False, self.result.wasSuccessful())
892
class TestReporter(TestReporterInterface):
894
Tests for the base L{reporter.Reporter} class.
898
TestReporterInterface.setUp(self)
900
self.result._getTime = self._getTime
908
def test_startStop(self):
909
self.result.startTest(self.test)
910
self.result.stopTest(self.test)
911
self.assertTrue(self.result._lastTime > 0)
912
self.assertEqual(self.result.testsRun, 1)
913
self.assertEqual(self.result.wasSuccessful(), True)
916
def test_brokenStream(self):
918
Test that the reporter safely writes to its stream.
920
result = self.resultFactory(stream=BrokenStream(self.stream))
921
result._writeln("Hello")
922
self.assertEqual(self.stream.getvalue(), 'Hello\n')
923
self.stream.truncate(0)
924
result._writeln("Hello %s!", 'World')
925
self.assertEqual(self.stream.getvalue(), 'Hello World!\n')
928
def test_printErrorsDeprecated(self):
930
L{IReporter.printErrors} was deprecated in Twisted 8.0.
933
self.result.printErrors()
935
DeprecationWarning, "printErrors is deprecated in Twisted 8.0.",
939
def test_printSummaryDeprecated(self):
941
L{IReporter.printSummary} was deprecated in Twisted 8.0.
944
self.result.printSummary()
946
DeprecationWarning, "printSummary is deprecated in Twisted 8.0.",
950
def test_writeDeprecated(self):
952
L{IReporter.write} was deprecated in Twisted 8.0.
955
self.result.write("")
957
DeprecationWarning, "write is deprecated in Twisted 8.0.",
961
def test_writelnDeprecated(self):
963
L{IReporter.writeln} was deprecated in Twisted 8.0.
966
self.result.writeln("")
968
DeprecationWarning, "writeln is deprecated in Twisted 8.0.",
972
def test_separatorDeprecated(self):
974
L{IReporter.separator} was deprecated in Twisted 8.0.
977
return self.result.separator
979
DeprecationWarning, "separator is deprecated in Twisted 8.0.",
983
def test_streamDeprecated(self):
985
L{IReporter.stream} was deprecated in Twisted 8.0.
988
return self.result.stream
990
DeprecationWarning, "stream is deprecated in Twisted 8.0.",
994
def test_upDownErrorDeprecated(self):
996
L{IReporter.upDownError} was deprecated in Twisted 8.0.
999
self.result.upDownError(None, None, None, None)
1001
DeprecationWarning, "upDownError is deprecated in Twisted 8.0.",
1005
def test_warning(self):
1007
L{reporter.Reporter} observes warnings emitted by the Twisted log
1008
system and writes them to its output stream.
1010
message = RuntimeWarning("some warning text")
1011
category = 'exceptions.RuntimeWarning'
1012
filename = "path/to/some/file.py"
1015
warning=message, category=category,
1016
filename=filename, lineno=lineno)
1018
self.stream.getvalue(),
1019
"%s:%d: %s: %s\n" % (
1020
filename, lineno, category.split('.')[-1], message))
1023
def test_duplicateWarningSuppressed(self):
1025
A warning emitted twice within a single test is only written to the
1028
# Emit the warning and assert that it shows up
1030
# Emit the warning again and assert that the stream still only has one
1035
def test_warningEmittedForNewTest(self):
1037
A warning emitted again after a new test has started is written to the
1040
test = self.__class__('test_warningEmittedForNewTest')
1041
self.result.startTest(test)
1043
# Clear whatever startTest wrote to the stream
1045
self.stream.truncate()
1047
# Emit a warning (and incidentally, assert that it was emitted)
1050
# Clean up from the first warning to simplify the rest of the
1053
self.stream.truncate()
1055
# Stop the first test and start another one (it just happens to be the
1056
# same one, but that doesn't matter)
1057
self.result.stopTest(test)
1058
self.result.startTest(test)
1060
# Clean up the stopTest/startTest output
1062
self.stream.truncate()
1064
# Emit the warning again and make sure it shows up
1068
def test_stopObserving(self):
1070
L{reporter.Reporter} stops observing log events when its C{done} method
1075
self.stream.truncate()
1077
warning=RuntimeWarning("some message"),
1078
category='exceptions.RuntimeWarning',
1079
filename="file/name.py", lineno=17)
1080
self.assertEqual(self.stream.getvalue(), "")
1084
class TestSafeStream(unittest.TestCase):
1085
def test_safe(self):
1087
Test that L{reporter.SafeStream} successfully write to its original
1088
stream even if an interrupt happens during the write.
1090
stream = StringIO.StringIO()
1091
broken = BrokenStream(stream)
1092
safe = reporter.SafeStream(broken)
1094
self.assertEqual(stream.getvalue(), "Hello")
1098
class TestSubunitReporter(TestReporterInterface):
1100
Tests for the subunit reporter.
1102
This just tests that the subunit reporter implements the basic interface.
1105
resultFactory = reporter.SubunitReporter
1109
if reporter.TestProtocolClient is None:
1111
"Subunit not installed, cannot test SubunitReporter")
1112
TestReporterInterface.setUp(self)
1115
def assertForwardsToSubunit(self, methodName, *args, **kwargs):
1117
Assert that 'methodName' on L{SubunitReporter} forwards to the
1118
equivalent method on subunit.
1120
Checks that the return value from subunit is returned from the
1121
L{SubunitReporter} and that the reporter writes the same data to its
1122
stream as subunit does to its own.
1124
Assumes that the method on subunit has the same name as the method on
1127
stream = StringIO.StringIO()
1128
subunitClient = reporter.TestProtocolClient(stream)
1129
subunitReturn = getattr(subunitClient, methodName)(*args, **kwargs)
1130
subunitOutput = stream.getvalue()
1131
reporterReturn = getattr(self.result, methodName)(*args, **kwargs)
1132
self.assertEquals(subunitReturn, reporterReturn)
1133
self.assertEquals(subunitOutput, self.stream.getvalue())
1136
def test_subunitWithoutAddExpectedFailureInstalled(self):
1138
Some versions of subunit don't have "addExpectedFailure". For these
1139
versions, we report expected failures as successes.
1142
addExpectedFailure = reporter.TestProtocolClient.addExpectedFailure
1143
except AttributeError:
1144
# Then we've actually got one of those old versions installed, and
1145
# the test is immediately applicable.
1148
del reporter.TestProtocolClient.addExpectedFailure
1150
setattr, reporter.TestProtocolClient, 'addExpectedFailure',
1154
except ZeroDivisionError:
1155
self.result.addExpectedFailure(self.test, sys.exc_info(), "todo")
1156
expectedFailureOutput = self.stream.getvalue()
1157
self.stream.truncate(0)
1158
self.result.addSuccess(self.test)
1159
successOutput = self.stream.getvalue()
1160
self.assertEquals(successOutput, expectedFailureOutput)
1163
def test_subunitWithoutAddSkipInstalled(self):
1165
Some versions of subunit don't have "addSkip". For these versions, we
1166
report skips as successes.
1169
addSkip = reporter.TestProtocolClient.addSkip
1170
except AttributeError:
1171
# Then we've actually got one of those old versions installed, and
1172
# the test is immediately applicable.
1175
del reporter.TestProtocolClient.addSkip
1177
setattr, reporter.TestProtocolClient, 'addSkip', addSkip)
1178
self.result.addSkip(self.test, "reason")
1179
skipOutput = self.stream.getvalue()
1180
self.stream.truncate(0)
1181
self.result.addSuccess(self.test)
1182
successOutput = self.stream.getvalue()
1183
self.assertEquals(successOutput, skipOutput)
1186
def test_addExpectedFailurePassedThrough(self):
1188
Some versions of subunit have "addExpectedFailure". For these
1189
versions, when we call 'addExpectedFailure' on the test result, we
1190
pass the error and test through to the subunit client.
1192
addExpectedFailureCalls = []
1193
def addExpectedFailure(test, error):
1194
addExpectedFailureCalls.append((test, error))
1196
# Provide our own addExpectedFailure, whether or not the locally
1197
# installed subunit has addExpectedFailure.
1198
self.result._subunit.addExpectedFailure = addExpectedFailure
1201
except ZeroDivisionError:
1202
exc_info = sys.exc_info()
1203
self.result.addExpectedFailure(self.test, exc_info, 'todo')
1204
self.assertEquals(addExpectedFailureCalls, [(self.test, exc_info)])
1207
def test_addSkipSendsSubunitAddSkip(self):
1209
Some versions of subunit have "addSkip". For these versions, when we
1210
call 'addSkip' on the test result, we pass the test and reason through
1211
to the subunit client.
1214
def addSkip(test, reason):
1215
addSkipCalls.append((test, reason))
1217
# Provide our own addSkip, whether or not the locally-installed
1218
# subunit has addSkip.
1219
self.result._subunit.addSkip = addSkip
1220
self.result.addSkip(self.test, 'reason')
1221
self.assertEquals(addSkipCalls, [(self.test, 'reason')])
1224
def test_doneDoesNothing(self):
1226
The subunit reporter doesn't need to print out a summary -- the stream
1227
of results is everything. Thus, done() does nothing.
1230
self.assertEquals('', self.stream.getvalue())
1233
def test_startTestSendsSubunitStartTest(self):
1235
SubunitReporter.startTest() sends the subunit 'startTest' message.
1237
self.assertForwardsToSubunit('startTest', self.test)
1240
def test_stopTestSendsSubunitStopTest(self):
1242
SubunitReporter.stopTest() sends the subunit 'stopTest' message.
1244
self.assertForwardsToSubunit('stopTest', self.test)
1247
def test_addSuccessSendsSubunitAddSuccess(self):
1249
SubunitReporter.addSuccess() sends the subunit 'addSuccess' message.
1251
self.assertForwardsToSubunit('addSuccess', self.test)
1254
def test_addErrorSendsSubunitAddError(self):
1256
SubunitReporter.addError() sends the subunit 'addError' message.
1260
except ZeroDivisionError:
1261
error = sys.exc_info()
1262
self.assertForwardsToSubunit('addError', self.test, error)
1265
def test_addFailureSendsSubunitAddFailure(self):
1267
SubunitReporter.addFailure() sends the subunit 'addFailure' message.
1271
except self.failureException:
1272
failure = sys.exc_info()
1273
self.assertForwardsToSubunit('addFailure', self.test, failure)
1276
def test_addUnexpectedSuccessSendsSubunitAddSuccess(self):
1278
SubunitReporter.addFailure() sends the subunit 'addSuccess' message,
1279
since subunit doesn't model unexpected success.
1281
stream = StringIO.StringIO()
1282
subunitClient = reporter.TestProtocolClient(stream)
1283
subunitClient.addSuccess(self.test)
1284
subunitOutput = stream.getvalue()
1285
self.result.addUnexpectedSuccess(self.test, 'todo')
1286
self.assertEquals(subunitOutput, self.stream.getvalue())
1290
class TestSubunitReporterNotInstalled(unittest.TestCase):
1292
Test behaviour when the subunit reporter is not installed.
1295
def test_subunitNotInstalled(self):
1297
If subunit is not installed, TestProtocolClient will be None, and
1298
SubunitReporter will raise an error when you try to construct it.
1300
stream = StringIO.StringIO()
1301
self.patch(reporter, 'TestProtocolClient', None)
1302
e = self.assertRaises(Exception, reporter.SubunitReporter, stream)
1303
self.assertEquals("Subunit not available", str(e))
1307
class TestTimingReporter(TestReporter):
1308
resultFactory = reporter.TimingTextReporter
1312
class LoggingReporter(reporter.Reporter):
1314
Simple reporter that stores the last test that was passed to it.
1317
def __init__(self, *args, **kwargs):
1318
reporter.Reporter.__init__(self, *args, **kwargs)
1321
def addError(self, test, error):
1324
def addExpectedFailure(self, test, failure, todo):
1327
def addFailure(self, test, failure):
1330
def addSkip(self, test, skip):
1333
def addUnexpectedSuccess(self, test, todo):
1336
def startTest(self, test):
1339
def stopTest(self, test):
1344
class TestAdaptedReporter(unittest.TestCase):
1346
L{reporter._AdaptedReporter} is a reporter wrapper that wraps all of the
1347
tests it receives before passing them on to the original reporter.
1351
self.wrappedResult = self.getWrappedResult()
1354
def _testAdapter(self, test):
1358
def assertWrapped(self, wrappedResult, test):
1359
self.assertEqual(wrappedResult._originalReporter.test, self._testAdapter(test))
1362
def getFailure(self, exceptionInstance):
1364
Return a L{Failure} from raising the given exception.
1366
@param exceptionInstance: The exception to raise.
1370
raise exceptionInstance
1375
def getWrappedResult(self):
1376
result = LoggingReporter()
1377
return reporter._AdaptedReporter(result, self._testAdapter)
1380
def test_addError(self):
1382
C{addError} wraps its test with the provided adapter.
1384
self.wrappedResult.addError(self, self.getFailure(RuntimeError()))
1385
self.assertWrapped(self.wrappedResult, self)
1388
def test_addFailure(self):
1390
C{addFailure} wraps its test with the provided adapter.
1392
self.wrappedResult.addFailure(self, self.getFailure(AssertionError()))
1393
self.assertWrapped(self.wrappedResult, self)
1396
def test_addSkip(self):
1398
C{addSkip} wraps its test with the provided adapter.
1400
self.wrappedResult.addSkip(self, self.getFailure(SkipTest('no reason')))
1401
self.assertWrapped(self.wrappedResult, self)
1404
def test_startTest(self):
1406
C{startTest} wraps its test with the provided adapter.
1408
self.wrappedResult.startTest(self)
1409
self.assertWrapped(self.wrappedResult, self)
1412
def test_stopTest(self):
1414
C{stopTest} wraps its test with the provided adapter.
1416
self.wrappedResult.stopTest(self)
1417
self.assertWrapped(self.wrappedResult, self)
1420
def test_addExpectedFailure(self):
1422
C{addExpectedFailure} wraps its test with the provided adapter.
1424
self.wrappedResult.addExpectedFailure(
1425
self, self.getFailure(RuntimeError()), Todo("no reason"))
1426
self.assertWrapped(self.wrappedResult, self)
1429
def test_addUnexpectedSuccess(self):
1431
C{addUnexpectedSuccess} wraps its test with the provided adapter.
1433
self.wrappedResult.addUnexpectedSuccess(self, Todo("no reason"))
1434
self.assertWrapped(self.wrappedResult, self)
1438
class FakeStream(object):
1440
A fake stream which C{isatty} method returns some predictable.
1442
@ivar tty: returned value of C{isatty}.
1446
def __init__(self, tty=True):
1455
class AnsiColorizerTests(unittest.TestCase):
1457
Tests for L{reporter._AnsiColorizer}.
1461
self.savedModules = sys.modules.copy()
1466
sys.modules.update(self.savedModules)
1469
def test_supportedStdOutTTY(self):
1471
L{reporter._AnsiColorizer.supported} returns C{False} if the given
1472
stream is not a TTY.
1474
self.assertFalse(reporter._AnsiColorizer.supported(FakeStream(False)))
1477
def test_supportedNoCurses(self):
1479
L{reporter._AnsiColorizer.supported} returns C{False} if the curses
1480
module can't be imported.
1482
sys.modules['curses'] = None
1483
self.assertFalse(reporter._AnsiColorizer.supported(FakeStream()))
1486
def test_supportedSetupTerm(self):
1488
L{reporter._AnsiColorizer.supported} returns C{True} if
1489
C{curses.tigetnum} returns more than 2 supported colors. It only tries
1490
to call C{curses.setupterm} if C{curses.tigetnum} previously failed
1491
with a C{curses.error}.
1493
class fakecurses(object):
1494
error = RuntimeError
1497
def setupterm(self):
1500
def tigetnum(self, value):
1506
sys.modules['curses'] = fakecurses()
1507
self.assertTrue(reporter._AnsiColorizer.supported(FakeStream()))
1508
self.assertTrue(reporter._AnsiColorizer.supported(FakeStream()))
1510
self.assertEquals(sys.modules['curses'].setUp, 1)
1513
def test_supportedTigetNumWrongError(self):
1515
L{reporter._AnsiColorizer.supported} returns C{False} and doesn't try
1516
to call C{curses.setupterm} if C{curses.tigetnum} returns something
1517
different than C{curses.error}.
1519
class fakecurses(object):
1520
error = RuntimeError
1522
def tigetnum(self, value):
1525
sys.modules['curses'] = fakecurses()
1526
self.assertFalse(reporter._AnsiColorizer.supported(FakeStream()))
1529
def test_supportedTigetNumNotEnoughColor(self):
1531
L{reporter._AnsiColorizer.supported} returns C{False} if
1532
C{curses.tigetnum} returns less than 2 supported colors.
1534
class fakecurses(object):
1535
error = RuntimeError
1537
def tigetnum(self, value):
1540
sys.modules['curses'] = fakecurses()
1541
self.assertFalse(reporter._AnsiColorizer.supported(FakeStream()))
1544
def test_supportedTigetNumErrors(self):
1546
L{reporter._AnsiColorizer.supported} returns C{False} if
1547
C{curses.tigetnum} raises an error, and calls C{curses.setupterm} once.
1549
class fakecurses(object):
1550
error = RuntimeError
1553
def setupterm(self):
1556
def tigetnum(self, value):
1559
sys.modules['curses'] = fakecurses()
1560
self.assertFalse(reporter._AnsiColorizer.supported(FakeStream()))
1561
self.assertEquals(sys.modules['curses'].setUp, 1)
added
removed
vendor/Twisted-10.0.0/twisted/trial/test/test_reporter.py
