1
# Copyright (c) 2005-2010 Twisted Matrix Laboratories.
2
# See LICENSE for details.
4
# Maintainer: Jonathan Lange
5
# Author: Robert Collins
8
import StringIO, os, sys
9
from zope.interface import implements
11
from twisted.trial.itrial import IReporter, ITestCase
12
from twisted.trial import unittest, runner, reporter, util
13
from twisted.python import failure, log, reflect, filepath
14
from twisted.scripts import trial
15
from twisted.plugins import twisted_trial
16
from twisted import plugin
17
from twisted.internet import defer
20
pyunit = __import__('unittest')
23
class CapturingDebugger(object):
28
def runcall(self, *args, **kwargs):
29
self._calls.append('runcall')
30
args[0](*args[1:], **kwargs)
34
class CapturingReporter(object):
36
Reporter that keeps a log of all actions performed on it.
47
def __init__(self, stream=None, tbformat=None, rterrors=None,
50
Create a capturing reporter.
53
self.shouldStop = False
55
self._tbformat = tbformat
56
self._rterrors = rterrors
57
self._publisher = publisher
60
def startTest(self, method):
62
Report the beginning of a run of a single test method
63
@param method: an object that is adaptable to ITestMethod
65
self._calls.append('startTest')
68
def stopTest(self, method):
70
Report the status of a single test method
71
@param method: an object that is adaptable to ITestMethod
73
self._calls.append('stopTest')
76
def cleanupErrors(self, errs):
77
"""called when the reactor has been left in a 'dirty' state
78
@param errs: a list of L{twisted.python.failure.Failure}s
80
self._calls.append('cleanupError')
83
def addSuccess(self, test):
84
self._calls.append('addSuccess')
89
Do nothing. These tests don't care about done.
94
class TrialRunnerTestsMixin:
96
Mixin defining tests for L{runner.TrialRunner}.
99
self.runner._tearDownLogFile()
102
def test_empty(self):
104
Empty test method, used by the other tests.
108
def _getObservers(self):
109
return log.theLogPublisher.observers
112
def test_addObservers(self):
114
Any log system observers L{TrialRunner.run} adds are removed by the
117
originalCount = len(self._getObservers())
118
self.runner.run(self.test)
119
newCount = len(self._getObservers())
120
self.assertEqual(newCount, originalCount)
123
def test_logFileAlwaysActive(self):
125
Test that a new file is opened on each run.
127
oldSetUpLogFile = self.runner._setUpLogFile
131
l.append(self.runner._logFileObserver)
132
self.runner._setUpLogFile = setUpLogFile
133
self.runner.run(self.test)
134
self.runner.run(self.test)
135
self.failUnlessEqual(len(l), 2)
136
self.failIf(l[0] is l[1], "Should have created a new file observer")
139
def test_logFileGetsClosed(self):
141
Test that file created is closed during the run.
143
oldSetUpLogFile = self.runner._setUpLogFile
147
l.append(self.runner._logFileObject)
148
self.runner._setUpLogFile = setUpLogFile
149
self.runner.run(self.test)
150
self.failUnlessEqual(len(l), 1)
151
self.failUnless(l[0].closed)
155
class TestTrialRunner(TrialRunnerTestsMixin, unittest.TestCase):
157
Tests for L{runner.TrialRunner} with the feature to turn unclean errors
158
into warnings disabled.
161
self.stream = StringIO.StringIO()
162
self.runner = runner.TrialRunner(CapturingReporter, stream=self.stream)
163
self.test = TestTrialRunner('test_empty')
166
def test_publisher(self):
168
The reporter constructed by L{runner.TrialRunner} is passed
169
L{twisted.python.log} as the value for the C{publisher} parameter.
171
result = self.runner._makeResult()
172
self.assertIdentical(result._publisher, log)
176
class TrialRunnerWithUncleanWarningsReporter(TrialRunnerTestsMixin,
179
Tests for the TrialRunner's interaction with an unclean-error suppressing
184
self.stream = StringIO.StringIO()
185
self.runner = runner.TrialRunner(CapturingReporter, stream=self.stream,
186
uncleanWarnings=True)
187
self.test = TestTrialRunner('test_empty')
191
class DryRunMixin(object):
193
suppress = [util.suppress(
194
category=DeprecationWarning,
195
message="Test visitors deprecated in Twisted 8.0")]
200
self.stream = StringIO.StringIO()
201
self.runner = runner.TrialRunner(CapturingReporter,
202
runner.TrialRunner.DRY_RUN,
204
self.makeTestFixtures()
207
def makeTestFixtures(self):
209
Set C{self.test} and C{self.suite}, where C{self.suite} is an empty
214
def test_empty(self):
216
If there are no tests, the reporter should not receive any events to
219
result = self.runner.run(runner.TestSuite())
220
self.assertEqual(result._calls, [])
223
def test_singleCaseReporting(self):
225
If we are running a single test, check the reporter starts, passes and
226
then stops the test during a dry run.
228
result = self.runner.run(self.test)
229
self.assertEqual(result._calls, ['startTest', 'addSuccess', 'stopTest'])
232
def test_testsNotRun(self):
234
When we are doing a dry run, the tests should not actually be run.
236
self.runner.run(self.test)
237
self.assertEqual(self.log, [])
241
class DryRunTest(DryRunMixin, unittest.TestCase):
243
Check that 'dry run' mode works well with Trial tests.
245
def makeTestFixtures(self):
246
class MockTest(unittest.TestCase):
248
self.log.append('test_foo')
249
self.test = MockTest('test_foo')
250
self.suite = runner.TestSuite()
254
class PyUnitDryRunTest(DryRunMixin, unittest.TestCase):
256
Check that 'dry run' mode works well with stdlib unittest tests.
258
def makeTestFixtures(self):
259
class PyunitCase(pyunit.TestCase):
262
self.test = PyunitCase('test_foo')
263
self.suite = pyunit.TestSuite()
267
class TestRunner(unittest.TestCase):
269
self.config = trial.Options()
270
# whitebox hack a reporter in, because plugins are CACHED and will
271
# only reload if the FILE gets changed.
273
parts = reflect.qual(CapturingReporter).split('.')
274
package = '.'.join(parts[:-1])
276
plugins = [twisted_trial._Reporter(
277
"Test Helper Reporter",
279
description="Utility for unit testing.",
285
# XXX There should really be a general way to hook the plugin system
287
def getPlugins(iface, *a, **kw):
288
self.assertEqual(iface, IReporter)
289
return plugins + list(self.original(iface, *a, **kw))
291
self.original = plugin.getPlugins
292
plugin.getPlugins = getPlugins
294
self.standardReport = ['startTest', 'addSuccess', 'stopTest',
295
'startTest', 'addSuccess', 'stopTest',
296
'startTest', 'addSuccess', 'stopTest',
297
'startTest', 'addSuccess', 'stopTest',
298
'startTest', 'addSuccess', 'stopTest',
299
'startTest', 'addSuccess', 'stopTest',
300
'startTest', 'addSuccess', 'stopTest']
304
plugin.getPlugins = self.original
307
def parseOptions(self, args):
308
self.config.parseOptions(args)
312
r = trial._makeRunner(self.config)
313
r.stream = StringIO.StringIO()
314
# XXX The runner should always take care of cleaning this up itself.
315
# It's not clear why this is necessary. The runner always tears down
317
self.addCleanup(r._tearDownLogFile)
318
# XXX The runner should always take care of cleaning this up itself as
319
# well. It's necessary because TrialRunner._setUpTestdir might raise
320
# an exception preventing Reporter.done from being run, leaving the
321
# observer added by Reporter.__init__ still present in the system.
322
# Something better needs to happen inside
323
# TrialRunner._runWithoutDecoration to remove the need for this cludge.
324
r._log = log.LogPublisher()
328
def test_runner_can_get_reporter(self):
329
self.parseOptions([])
330
result = self.config['reporter']
331
runner = self.getRunner()
332
self.assertEqual(result, runner._makeResult().__class__)
335
def test_runner_get_result(self):
336
self.parseOptions([])
337
runner = self.getRunner()
338
result = runner._makeResult()
339
self.assertEqual(result.__class__, self.config['reporter'])
342
def test_uncleanWarningsOffByDefault(self):
344
By default Trial sets the 'uncleanWarnings' option on the runner to
345
False. This means that dirty reactor errors will be reported as
346
errors. See L{test_reporter.TestDirtyReactor}.
348
self.parseOptions([])
349
runner = self.getRunner()
350
self.assertNotIsInstance(runner._makeResult(),
351
reporter.UncleanWarningsReporterWrapper)
354
def test_getsUncleanWarnings(self):
356
Specifying '--unclean-warnings' on the trial command line will cause
357
reporters to be wrapped in a device which converts unclean errors to
358
warnings. See L{test_reporter.TestDirtyReactor} for implications.
360
self.parseOptions(['--unclean-warnings'])
361
runner = self.getRunner()
362
self.assertIsInstance(runner._makeResult(),
363
reporter.UncleanWarningsReporterWrapper)
366
def test_runner_working_directory(self):
367
self.parseOptions(['--temp-directory', 'some_path'])
368
runner = self.getRunner()
369
self.assertEquals(runner.workingDirectory, 'some_path')
372
def test_concurrentImplicitWorkingDirectory(self):
374
If no working directory is explicitly specified and the default
375
working directory is in use by another runner, L{TrialRunner.run}
376
selects a different default working directory to use.
378
self.parseOptions([])
380
initialDirectory = os.getcwd()
381
self.addCleanup(os.chdir, initialDirectory)
383
firstRunner = self.getRunner()
384
secondRunner = self.getRunner()
388
class ConcurrentCase(unittest.TestCase):
389
def test_first(self):
391
Start a second test run which will have a default working
392
directory which is the same as the working directory of the
393
test run already in progress.
395
# Change the working directory to the value it had before this
396
# test suite was started.
397
where['concurrent'] = subsequentDirectory = os.getcwd()
398
os.chdir(initialDirectory)
399
self.addCleanup(os.chdir, subsequentDirectory)
401
secondRunner.run(ConcurrentCase('test_second'))
403
def test_second(self):
405
Record the working directory for later analysis.
407
where['record'] = os.getcwd()
409
result = firstRunner.run(ConcurrentCase('test_first'))
410
bad = result.errors + result.failures
415
'concurrent': os.path.join(initialDirectory, '_trial_temp'),
416
'record': os.path.join(initialDirectory, '_trial_temp-1')})
419
def test_concurrentExplicitWorkingDirectory(self):
421
If a working directory which is already in use is explicitly specified,
422
L{TrialRunner.run} raises L{_WorkingDirectoryBusy}.
424
self.parseOptions(['--temp-directory', os.path.abspath(self.mktemp())])
426
initialDirectory = os.getcwd()
427
self.addCleanup(os.chdir, initialDirectory)
429
firstRunner = self.getRunner()
430
secondRunner = self.getRunner()
432
class ConcurrentCase(unittest.TestCase):
433
def test_concurrent(self):
435
Try to start another runner in the same working directory and
436
assert that it raises L{_WorkingDirectoryBusy}.
439
runner._WorkingDirectoryBusy,
440
secondRunner.run, ConcurrentCase('test_failure'))
442
def test_failure(self):
444
Should not be called, always fails.
446
self.fail("test_failure should never be called.")
448
result = firstRunner.run(ConcurrentCase('test_concurrent'))
449
bad = result.errors + result.failures
454
def test_runner_normal(self):
455
self.parseOptions(['--temp-directory', self.mktemp(),
456
'--reporter', 'capturing',
457
'twisted.trial.test.sample'])
458
my_runner = self.getRunner()
459
loader = runner.TestLoader()
460
suite = loader.loadByName('twisted.trial.test.sample', True)
461
result = my_runner.run(suite)
462
self.assertEqual(self.standardReport, result._calls)
465
def test_runner_debug(self):
466
self.parseOptions(['--reporter', 'capturing',
467
'--debug', 'twisted.trial.test.sample'])
468
my_runner = self.getRunner()
469
debugger = CapturingDebugger()
472
my_runner._getDebugger = get_debugger
473
loader = runner.TestLoader()
474
suite = loader.loadByName('twisted.trial.test.sample', True)
475
result = my_runner.run(suite)
476
self.assertEqual(self.standardReport, result._calls)
477
self.assertEqual(['runcall'], debugger._calls)
480
def test_removeSafelyNoTrialMarker(self):
482
If a path doesn't contain a node named C{"_trial_marker"}, that path is
483
not removed by L{runner._removeSafely} and a L{runner._NoTrialMarker}
484
exception is raised instead.
486
directory = self.mktemp()
488
dirPath = filepath.FilePath(directory)
490
self.parseOptions([])
491
myRunner = self.getRunner()
492
self.assertRaises(runner._NoTrialMarker,
493
myRunner._removeSafely, dirPath)
496
def test_removeSafelyRemoveFailsMoveSucceeds(self):
498
If an L{OSError} is raised while removing a path in
499
L{runner._removeSafely}, an attempt is made to move the path to a new
504
Raise an C{OSError} to emulate the branch of L{runner._removeSafely}
505
in which path removal fails.
509
# Patch stdout so we can check the print statements in _removeSafely
510
out = StringIO.StringIO()
511
stdout = self.patch(sys, 'stdout', out)
513
# Set up a trial directory with a _trial_marker
514
directory = self.mktemp()
516
dirPath = filepath.FilePath(directory)
517
dirPath.child('_trial_marker').touch()
518
# Ensure that path.remove() raises an OSError
519
dirPath.remove = dummyRemove
521
self.parseOptions([])
522
myRunner = self.getRunner()
523
myRunner._removeSafely(dirPath)
524
self.assertIn("could not remove FilePath", out.getvalue())
527
def test_removeSafelyRemoveFailsMoveFails(self):
529
If an L{OSError} is raised while removing a path in
530
L{runner._removeSafely}, an attempt is made to move the path to a new
531
name. If that attempt fails, the L{OSError} is re-raised.
535
Raise an C{OSError} to emulate the branch of L{runner._removeSafely}
536
in which path removal fails.
538
raise OSError("path removal failed")
540
def dummyMoveTo(path):
542
Raise an C{OSError} to emulate the branch of L{runner._removeSafely}
543
in which path movement fails.
545
raise OSError("path movement failed")
547
# Patch stdout so we can check the print statements in _removeSafely
548
out = StringIO.StringIO()
549
stdout = self.patch(sys, 'stdout', out)
551
# Set up a trial directory with a _trial_marker
552
directory = self.mktemp()
554
dirPath = filepath.FilePath(directory)
555
dirPath.child('_trial_marker').touch()
557
# Ensure that path.remove() and path.moveTo() both raise OSErrors
558
dirPath.remove = dummyRemove
559
dirPath.moveTo = dummyMoveTo
561
self.parseOptions([])
562
myRunner = self.getRunner()
563
error = self.assertRaises(OSError, myRunner._removeSafely, dirPath)
564
self.assertEquals(str(error), "path movement failed")
565
self.assertIn("could not remove FilePath", out.getvalue())
569
class TestTrialSuite(unittest.TestCase):
571
def test_imports(self):
572
# FIXME, HTF do you test the reactor can be cleaned up ?!!!
573
from twisted.trial.runner import TrialSuite
574
# silence pyflakes warning
575
silencePyflakes = TrialSuite
580
class TestUntilFailure(unittest.TestCase):
581
class FailAfter(unittest.TestCase):
583
A test case that fails when run 3 times in a row.
587
self.count.append(None)
588
if len(self.count) == 3:
589
self.fail('Count reached 3')
593
TestUntilFailure.FailAfter.count = []
594
self.test = TestUntilFailure.FailAfter('test_foo')
595
self.stream = StringIO.StringIO()
596
self.runner = runner.TrialRunner(reporter.Reporter, stream=self.stream)
599
def test_runUntilFailure(self):
601
Test that the runUntilFailure method of the runner actually fail after
604
result = self.runner.runUntilFailure(self.test)
605
self.failUnlessEqual(result.testsRun, 1)
606
self.failIf(result.wasSuccessful())
607
self.assertEquals(self._getFailures(result), 1)
610
def _getFailures(self, result):
612
Get the number of failures that were reported to a result.
614
return len(result.failures)
617
def test_runUntilFailureDecorate(self):
619
C{runUntilFailure} doesn't decorate the tests uselessly: it does it one
620
time when run starts, but not at each turn.
623
def decorate(test, interface):
624
decorated.append((test, interface))
626
self.patch(unittest, "decorate", decorate)
627
result = self.runner.runUntilFailure(self.test)
628
self.failUnlessEqual(result.testsRun, 1)
630
self.assertEquals(len(decorated), 1)
631
self.assertEquals(decorated, [(self.test, ITestCase)])
634
def test_runUntilFailureForceGCDecorate(self):
636
C{runUntilFailure} applies the force-gc decoration after the standard
637
L{ITestCase} decoration, but only one time.
640
def decorate(test, interface):
641
decorated.append((test, interface))
643
self.patch(unittest, "decorate", decorate)
644
self.runner._forceGarbageCollection = True
645
result = self.runner.runUntilFailure(self.test)
646
self.failUnlessEqual(result.testsRun, 1)
648
self.assertEquals(len(decorated), 2)
649
self.assertEquals(decorated,
650
[(self.test, ITestCase),
651
(self.test, unittest._ForceGarbageCollectionDecorator)])
655
class UncleanUntilFailureTests(TestUntilFailure):
657
Test that the run-until-failure feature works correctly with the unclean
662
TestUntilFailure.setUp(self)
663
self.runner = runner.TrialRunner(reporter.Reporter, stream=self.stream,
664
uncleanWarnings=True)
666
def _getFailures(self, result):
668
Get the number of failures that were reported to a result that
669
is wrapped in an UncleanFailureWrapper.
671
return len(result._originalReporter.failures)
675
class BreakingSuite(runner.TestSuite):
677
A L{TestSuite} that logs an error when it is run.
680
def run(self, result):
682
raise RuntimeError("error that occurs outside of a test")
683
except RuntimeError, e:
684
log.err(failure.Failure())
688
class TestLoggedErrors(unittest.TestCase):
690
It is possible for an error generated by a test to be logged I{outside} of
691
any test. The log observers constructed by L{TestCase} won't catch these
692
errors. Here we try to generate such errors and ensure they are reported to
693
a L{TestResult} object.
697
self.flushLoggedErrors(RuntimeError)
700
def test_construct(self):
702
Check that we can construct a L{runner.LoggedSuite} and that it
705
suite = runner.LoggedSuite()
706
self.assertEqual(suite.countTestCases(), 0)
709
def test_capturesError(self):
711
Chek that a L{LoggedSuite} reports any logged errors to its result.
713
result = reporter.TestResult()
714
suite = runner.LoggedSuite([BreakingSuite()])
716
self.assertEqual(len(result.errors), 1)
717
self.assertEqual(result.errors[0][0].id(), runner.NOT_IN_TEST)
718
self.failUnless(result.errors[0][1].check(RuntimeError))
722
class TestTestHolder(unittest.TestCase):
725
self.description = "description"
726
self.holder = runner.TestHolder(self.description)
729
def test_holder(self):
731
Check that L{runner.TestHolder} takes a description as a parameter
732
and that this description is returned by the C{id} and
733
C{shortDescription} methods.
735
self.assertEqual(self.holder.id(), self.description)
736
self.assertEqual(self.holder.shortDescription(), self.description)
739
def test_holderImplementsITestCase(self):
741
L{runner.TestHolder} implements L{ITestCase}.
743
self.assertIdentical(self.holder, ITestCase(self.holder))
747
class TestErrorHolder(TestTestHolder):
749
Test L{runner.ErrorHolder} shares behaviour with L{runner.TestHolder}.
753
self.description = "description"
754
# make a real Failure so we can construct ErrorHolder()
757
except ZeroDivisionError:
758
error = failure.Failure()
759
self.holder = runner.ErrorHolder(self.description, error)
763
class TestMalformedMethod(unittest.TestCase):
765
Test that trial manages when test methods don't have correct signatures.
767
class ContainMalformed(unittest.TestCase):
769
This TestCase holds malformed test methods that trial should handle.
771
def test_foo(self, blah):
775
test_spam = defer.deferredGenerator(test_bar)
777
def _test(self, method):
779
Wrapper for one of the test method of L{ContainMalformed}.
781
stream = StringIO.StringIO()
782
trialRunner = runner.TrialRunner(reporter.Reporter, stream=stream)
783
test = TestMalformedMethod.ContainMalformed(method)
784
result = trialRunner.run(test)
785
self.failUnlessEqual(result.testsRun, 1)
786
self.failIf(result.wasSuccessful())
787
self.failUnlessEqual(len(result.errors), 1)
789
def test_extraArg(self):
791
Test when the method has extra (useless) arguments.
793
self._test('test_foo')
795
def test_noArg(self):
797
Test when the method doesn't have even self as argument.
799
self._test('test_bar')
801
def test_decorated(self):
803
Test a decorated method also fails.
805
self._test('test_spam')
809
class DestructiveTestSuiteTestCase(unittest.TestCase):
811
Test for L{runner.DestructiveTestSuite}.
814
def test_basic(self):
816
Thes destructive test suite should run the tests normally.
819
class MockTest(unittest.TestCase):
822
test = MockTest('test_foo')
823
result = reporter.TestResult()
824
suite = runner.DestructiveTestSuite([test])
825
self.assertEquals(called, [])
827
self.assertEquals(called, [True])
828
self.assertEquals(suite.countTestCases(), 0)
831
def test_shouldStop(self):
833
Test the C{shouldStop} management: raising a C{KeyboardInterrupt} must
837
class MockTest(unittest.TestCase):
841
raise KeyboardInterrupt()
844
result = reporter.TestResult()
845
loader = runner.TestLoader()
846
loader.suiteFactory = runner.DestructiveTestSuite
847
suite = loader.loadClass(MockTest)
848
self.assertEquals(called, [])
850
self.assertEquals(called, [1])
851
# The last test shouldn't have been run
852
self.assertEquals(suite.countTestCases(), 1)
855
def test_cleanup(self):
857
Checks that the test suite cleanups its tests during the run, so that
860
class MockTest(unittest.TestCase):
863
test = MockTest('test_foo')
864
result = reporter.TestResult()
865
suite = runner.DestructiveTestSuite([test])
866
self.assertEquals(suite.countTestCases(), 1)
868
self.assertEquals(suite.countTestCases(), 0)
872
class TestRunnerDeprecation(unittest.TestCase):
874
class FakeReporter(reporter.Reporter):
876
Fake reporter that does *not* implement done() but *does* implement
877
printErrors, separator, printSummary, stream, write and writeln
878
without deprecations.
885
def printErrors(self, *args):
888
def printSummary(self, *args):
891
def write(self, *args):
894
def writeln(self, *args):
898
def test_reporterDeprecations(self):
900
The runner emits a warning if it is using a result that doesn't
903
trialRunner = runner.TrialRunner(None)
904
result = self.FakeReporter()
905
trialRunner._makeResult = lambda: result
907
# We have to use a pyunit test, otherwise we'll get deprecation
908
# warnings about using iterate() in a test.
909
trialRunner.run(pyunit.TestCase('id'))
912
"%s should implement done() but doesn't. Falling back to "
913
"printErrors() and friends." % reflect.qual(result.__class__),
added
removed
vendor/Twisted-10.0.0/twisted/trial/test/test_runner.py
