| Index: third_party/twisted_8_1/twisted/trial/test/test_runner.py
|
| diff --git a/third_party/twisted_8_1/twisted/trial/test/test_runner.py b/third_party/twisted_8_1/twisted/trial/test/test_runner.py
|
| deleted file mode 100644
|
| index 2addf8375d2bdf7080bacc908f1078836da23dd6..0000000000000000000000000000000000000000
|
| --- a/third_party/twisted_8_1/twisted/trial/test/test_runner.py
|
| +++ /dev/null
|
| @@ -1,676 +0,0 @@
|
| -# Copyright (c) 2005-2007 Twisted Matrix Laboratories.
|
| -# See LICENSE for details.
|
| -#
|
| -# Maintainer: Jonathan Lange <jml@twistedmatrix.com>
|
| -# Author: Robert Collins <robertc@robertcollins.net>
|
| -
|
| -
|
| -import StringIO
|
| -from zope.interface import implements
|
| -
|
| -from twisted.trial.itrial import IReporter, ITestCase
|
| -from twisted.trial import unittest, runner, reporter, util
|
| -from twisted.python import failure, log, reflect
|
| -from twisted.scripts import trial
|
| -from twisted.plugins import twisted_trial
|
| -from twisted import plugin
|
| -from twisted.internet import defer
|
| -
|
| -
|
| -pyunit = __import__('unittest')
|
| -
|
| -
|
| -class CapturingDebugger(object):
|
| -
|
| - def __init__(self):
|
| - self._calls = []
|
| -
|
| - def runcall(self, *args, **kwargs):
|
| - self._calls.append('runcall')
|
| - args[0](*args[1:], **kwargs)
|
| -
|
| -
|
| -
|
| -class CapturingReporter(object):
|
| - """
|
| - Reporter that keeps a log of all actions performed on it.
|
| - """
|
| -
|
| - implements(IReporter)
|
| -
|
| - stream = None
|
| - tbformat = None
|
| - args = None
|
| - separator = None
|
| - testsRun = None
|
| -
|
| - def __init__(self, *a, **kw):
|
| - """
|
| - Create a capturing reporter.
|
| - """
|
| - self._calls = []
|
| - self.shouldStop = False
|
| -
|
| -
|
| - def startTest(self, method):
|
| - """
|
| - Report the beginning of a run of a single test method
|
| - @param method: an object that is adaptable to ITestMethod
|
| - """
|
| - self._calls.append('startTest')
|
| -
|
| -
|
| - def stopTest(self, method):
|
| - """
|
| - Report the status of a single test method
|
| - @param method: an object that is adaptable to ITestMethod
|
| - """
|
| - self._calls.append('stopTest')
|
| -
|
| -
|
| - def cleanupErrors(self, errs):
|
| - """called when the reactor has been left in a 'dirty' state
|
| - @param errs: a list of L{twisted.python.failure.Failure}s
|
| - """
|
| - self._calls.append('cleanupError')
|
| -
|
| -
|
| - def addSuccess(self, test):
|
| - self._calls.append('addSuccess')
|
| -
|
| -
|
| - def done(self):
|
| - """
|
| - Do nothing. These tests don't care about done.
|
| - """
|
| -
|
| -
|
| -
|
| -class TestTrialRunner(unittest.TestCase):
|
| -
|
| - def setUp(self):
|
| - self.stream = StringIO.StringIO()
|
| - self.runner = runner.TrialRunner(CapturingReporter, stream=self.stream)
|
| - self.test = TestTrialRunner('test_empty')
|
| -
|
| - def test_empty(self):
|
| - """
|
| - Empty test method, used by the other tests.
|
| - """
|
| -
|
| - def tearDown(self):
|
| - self.runner._tearDownLogFile()
|
| -
|
| - def _getObservers(self):
|
| - return log.theLogPublisher.observers
|
| -
|
| - def test_addObservers(self):
|
| - """
|
| - Tests that the runner add logging observers during the run.
|
| - """
|
| - originalCount = len(self._getObservers())
|
| - self.runner.run(self.test)
|
| - newCount = len(self._getObservers())
|
| - self.failUnlessEqual(originalCount + 1, newCount)
|
| -
|
| - def test_addObservers_repeat(self):
|
| - self.runner.run(self.test)
|
| - count = len(self._getObservers())
|
| - self.runner.run(self.test)
|
| - newCount = len(self._getObservers())
|
| - self.failUnlessEqual(count, newCount)
|
| -
|
| - def test_logFileAlwaysActive(self):
|
| - """
|
| - Test that a new file is opened on each run.
|
| - """
|
| - oldSetUpLogging = self.runner._setUpLogging
|
| - l = []
|
| - def setUpLogging():
|
| - oldSetUpLogging()
|
| - l.append(self.runner._logFileObserver)
|
| - self.runner._setUpLogging = setUpLogging
|
| - self.runner.run(self.test)
|
| - self.runner.run(self.test)
|
| - self.failUnlessEqual(len(l), 2)
|
| - self.failIf(l[0] is l[1], "Should have created a new file observer")
|
| -
|
| - def test_logFileGetsClosed(self):
|
| - """
|
| - Test that file created is closed during the run.
|
| - """
|
| - oldSetUpLogging = self.runner._setUpLogging
|
| - l = []
|
| - def setUpLogging():
|
| - oldSetUpLogging()
|
| - l.append(self.runner._logFileObject)
|
| - self.runner._setUpLogging = setUpLogging
|
| - self.runner.run(self.test)
|
| - self.failUnlessEqual(len(l), 1)
|
| - self.failUnless(l[0].closed)
|
| -
|
| -
|
| -
|
| -class TrialRunnerWithUncleanWarningsReporter(TestTrialRunner):
|
| - """
|
| - Tests for the TrialRunner's interaction with an unclean-error suppressing
|
| - reporter.
|
| - """
|
| -
|
| - def setUp(self):
|
| - self.stream = StringIO.StringIO()
|
| - self.runner = runner.TrialRunner(CapturingReporter, stream=self.stream,
|
| - uncleanWarnings=True)
|
| - self.test = TestTrialRunner('test_empty')
|
| -
|
| -
|
| -
|
| -class DryRunMixin(object):
|
| -
|
| - suppress = [util.suppress(
|
| - category=DeprecationWarning,
|
| - message="Test visitors deprecated in Twisted 8.0")]
|
| -
|
| -
|
| - def setUp(self):
|
| - self.log = []
|
| - self.stream = StringIO.StringIO()
|
| - self.runner = runner.TrialRunner(CapturingReporter,
|
| - runner.TrialRunner.DRY_RUN,
|
| - stream=self.stream)
|
| - self.makeTestFixtures()
|
| -
|
| -
|
| - def makeTestFixtures(self):
|
| - """
|
| - Set C{self.test} and C{self.suite}, where C{self.suite} is an empty
|
| - TestSuite.
|
| - """
|
| -
|
| -
|
| - def test_empty(self):
|
| - """
|
| - If there are no tests, the reporter should not receive any events to
|
| - report.
|
| - """
|
| - result = self.runner.run(runner.TestSuite())
|
| - self.assertEqual(result._calls, [])
|
| -
|
| -
|
| - def test_singleCaseReporting(self):
|
| - """
|
| - If we are running a single test, check the reporter starts, passes and
|
| - then stops the test during a dry run.
|
| - """
|
| - result = self.runner.run(self.test)
|
| - self.assertEqual(result._calls, ['startTest', 'addSuccess', 'stopTest'])
|
| -
|
| -
|
| - def test_testsNotRun(self):
|
| - """
|
| - When we are doing a dry run, the tests should not actually be run.
|
| - """
|
| - self.runner.run(self.test)
|
| - self.assertEqual(self.log, [])
|
| -
|
| -
|
| -
|
| -class DryRunTest(DryRunMixin, unittest.TestCase):
|
| - """
|
| - Check that 'dry run' mode works well with Trial tests.
|
| - """
|
| - def makeTestFixtures(self):
|
| - class MockTest(unittest.TestCase):
|
| - def test_foo(test):
|
| - self.log.append('test_foo')
|
| - self.test = MockTest('test_foo')
|
| - self.suite = runner.TestSuite()
|
| -
|
| -
|
| -
|
| -class PyUnitDryRunTest(DryRunMixin, unittest.TestCase):
|
| - """
|
| - Check that 'dry run' mode works well with stdlib unittest tests.
|
| - """
|
| - def makeTestFixtures(self):
|
| - class PyunitCase(pyunit.TestCase):
|
| - def test_foo(self):
|
| - pass
|
| - self.test = PyunitCase('test_foo')
|
| - self.suite = pyunit.TestSuite()
|
| -
|
| -
|
| -
|
| -class TestRunner(unittest.TestCase):
|
| - def setUp(self):
|
| - self.config = trial.Options()
|
| - # whitebox hack a reporter in, because plugins are CACHED and will
|
| - # only reload if the FILE gets changed.
|
| -
|
| - parts = reflect.qual(CapturingReporter).split('.')
|
| - package = '.'.join(parts[:-1])
|
| - klass = parts[-1]
|
| - plugins = [twisted_trial._Reporter(
|
| - "Test Helper Reporter",
|
| - package,
|
| - description="Utility for unit testing.",
|
| - longOpt="capturing",
|
| - shortOpt=None,
|
| - klass=klass)]
|
| -
|
| -
|
| - # XXX There should really be a general way to hook the plugin system
|
| - # for tests.
|
| - def getPlugins(iface, *a, **kw):
|
| - self.assertEqual(iface, IReporter)
|
| - return plugins + list(self.original(iface, *a, **kw))
|
| -
|
| - self.original = plugin.getPlugins
|
| - plugin.getPlugins = getPlugins
|
| -
|
| - self.standardReport = ['startTest', 'addSuccess', 'stopTest',
|
| - 'startTest', 'addSuccess', 'stopTest',
|
| - 'startTest', 'addSuccess', 'stopTest',
|
| - 'startTest', 'addSuccess', 'stopTest',
|
| - 'startTest', 'addSuccess', 'stopTest',
|
| - 'startTest', 'addSuccess', 'stopTest',
|
| - 'startTest', 'addSuccess', 'stopTest']
|
| -
|
| -
|
| - def tearDown(self):
|
| - plugin.getPlugins = self.original
|
| -
|
| -
|
| - def parseOptions(self, args):
|
| - self.config.parseOptions(args)
|
| -
|
| -
|
| - def getRunner(self):
|
| - r = trial._makeRunner(self.config)
|
| - r.stream = StringIO.StringIO()
|
| - self.addCleanup(r._tearDownLogFile)
|
| - return r
|
| -
|
| -
|
| - def test_runner_can_get_reporter(self):
|
| - self.parseOptions([])
|
| - result = self.config['reporter']
|
| - my_runner = self.getRunner()
|
| - try:
|
| - self.assertEqual(result, my_runner._makeResult().__class__)
|
| - finally:
|
| - my_runner._tearDownLogFile()
|
| -
|
| -
|
| - def test_runner_get_result(self):
|
| - self.parseOptions([])
|
| - my_runner = self.getRunner()
|
| - result = my_runner._makeResult()
|
| - self.assertEqual(result.__class__, self.config['reporter'])
|
| -
|
| -
|
| - def test_uncleanWarningsOffByDefault(self):
|
| - """
|
| - By default Trial sets the 'uncleanWarnings' option on the runner to
|
| - False. This means that dirty reactor errors will be reported as
|
| - errors. See L{test_reporter.TestDirtyReactor}.
|
| - """
|
| - self.parseOptions([])
|
| - runner = self.getRunner()
|
| - self.assertNotIsInstance(runner._makeResult(),
|
| - reporter.UncleanWarningsReporterWrapper)
|
| -
|
| -
|
| - def test_getsUncleanWarnings(self):
|
| - """
|
| - Specifying '--unclean-warnings' on the trial command line will cause
|
| - reporters to be wrapped in a device which converts unclean errors to
|
| - warnings. See L{test_reporter.TestDirtyReactor} for implications.
|
| - """
|
| - self.parseOptions(['--unclean-warnings'])
|
| - runner = self.getRunner()
|
| - self.assertIsInstance(runner._makeResult(),
|
| - reporter.UncleanWarningsReporterWrapper)
|
| -
|
| -
|
| - def test_runner_working_directory(self):
|
| - self.parseOptions(['--temp-directory', 'some_path'])
|
| - runner = self.getRunner()
|
| - try:
|
| - self.assertEquals(runner.workingDirectory, 'some_path')
|
| - finally:
|
| - runner._tearDownLogFile()
|
| -
|
| -
|
| - def test_runner_normal(self):
|
| - self.parseOptions(['--temp-directory', self.mktemp(),
|
| - '--reporter', 'capturing',
|
| - 'twisted.trial.test.sample'])
|
| - my_runner = self.getRunner()
|
| - loader = runner.TestLoader()
|
| - suite = loader.loadByName('twisted.trial.test.sample', True)
|
| - result = my_runner.run(suite)
|
| - self.assertEqual(self.standardReport, result._calls)
|
| -
|
| -
|
| - def test_runner_debug(self):
|
| - self.parseOptions(['--reporter', 'capturing',
|
| - '--debug', 'twisted.trial.test.sample'])
|
| - my_runner = self.getRunner()
|
| - debugger = CapturingDebugger()
|
| - def get_debugger():
|
| - return debugger
|
| - my_runner._getDebugger = get_debugger
|
| - loader = runner.TestLoader()
|
| - suite = loader.loadByName('twisted.trial.test.sample', True)
|
| - result = my_runner.run(suite)
|
| - self.assertEqual(self.standardReport, result._calls)
|
| - self.assertEqual(['runcall'], debugger._calls)
|
| -
|
| -
|
| -
|
| -class TestTrialSuite(unittest.TestCase):
|
| -
|
| - def test_imports(self):
|
| - # FIXME, HTF do you test the reactor can be cleaned up ?!!!
|
| - from twisted.trial.runner import TrialSuite
|
| - # silence pyflakes warning
|
| - silencePyflakes = TrialSuite
|
| -
|
| -
|
| -
|
| -class TestUntilFailure(unittest.TestCase):
|
| - class FailAfter(unittest.TestCase):
|
| - """
|
| - A test case that fails when run 3 times in a row.
|
| - """
|
| - count = []
|
| - def test_foo(self):
|
| - self.count.append(None)
|
| - if len(self.count) == 3:
|
| - self.fail('Count reached 3')
|
| -
|
| - def setUp(self):
|
| - TestUntilFailure.FailAfter.count = []
|
| - self.test = TestUntilFailure.FailAfter('test_foo')
|
| - self.stream = StringIO.StringIO()
|
| - self.runner = runner.TrialRunner(reporter.Reporter, stream=self.stream)
|
| -
|
| - def test_runUntilFailure(self):
|
| - """
|
| - Test that the runUntilFailure method of the runner actually fail after
|
| - a few runs.
|
| - """
|
| - result = self.runner.runUntilFailure(self.test)
|
| - self.failUnlessEqual(result.testsRun, 1)
|
| - self.failIf(result.wasSuccessful())
|
| - self.assertEquals(self._getFailures(result), 1)
|
| -
|
| - def _getFailures(self, result):
|
| - """
|
| - Get the number of failures that were reported to a result.
|
| - """
|
| - return len(result.failures)
|
| -
|
| -
|
| -
|
| -class UncleanUntilFailureTests(TestUntilFailure):
|
| - """
|
| - Test that the run-until-failure feature works correctly with the unclean
|
| - error suppressor.
|
| - """
|
| -
|
| - def setUp(self):
|
| - TestUntilFailure.setUp(self)
|
| - self.runner = runner.TrialRunner(reporter.Reporter, stream=self.stream,
|
| - uncleanWarnings=True)
|
| -
|
| - def _getFailures(self, result):
|
| - """
|
| - Get the number of failures that were reported to a result that
|
| - is wrapped in an UncleanFailureWrapper.
|
| - """
|
| - return len(result._originalReporter.failures)
|
| -
|
| -
|
| -
|
| -class BreakingSuite(runner.TestSuite):
|
| - """
|
| - A L{TestSuite} that logs an error when it is run.
|
| - """
|
| -
|
| - def run(self, result):
|
| - try:
|
| - raise RuntimeError("error that occurs outside of a test")
|
| - except RuntimeError, e:
|
| - log.err(failure.Failure())
|
| -
|
| -
|
| -
|
| -class TestLoggedErrors(unittest.TestCase):
|
| - """
|
| - It is possible for an error generated by a test to be logged I{outside} of
|
| - any test. The log observers constructed by L{TestCase} won't catch these
|
| - errors. Here we try to generate such errors and ensure they are reported to
|
| - a L{TestResult} object.
|
| - """
|
| -
|
| - def tearDown(self):
|
| - self.flushLoggedErrors(RuntimeError)
|
| -
|
| -
|
| - def test_construct(self):
|
| - """
|
| - Check that we can construct a L{runner.LoggedSuite} and that it
|
| - starts empty.
|
| - """
|
| - suite = runner.LoggedSuite()
|
| - self.assertEqual(suite.countTestCases(), 0)
|
| -
|
| -
|
| - def test_capturesError(self):
|
| - """
|
| - Chek that a L{LoggedSuite} reports any logged errors to its result.
|
| - """
|
| - result = reporter.TestResult()
|
| - suite = runner.LoggedSuite([BreakingSuite()])
|
| - suite.run(result)
|
| - self.assertEqual(len(result.errors), 1)
|
| - self.assertEqual(result.errors[0][0].id(), runner.NOT_IN_TEST)
|
| - self.failUnless(result.errors[0][1].check(RuntimeError))
|
| -
|
| -
|
| -
|
| -class TestTestHolder(unittest.TestCase):
|
| -
|
| - def setUp(self):
|
| - self.description = "description"
|
| - self.holder = runner.TestHolder(self.description)
|
| -
|
| -
|
| - def test_holder(self):
|
| - """
|
| - Check that L{runner.TestHolder} takes a description as a parameter
|
| - and that this description is returned by the C{id} and
|
| - C{shortDescription} methods.
|
| - """
|
| - self.assertEqual(self.holder.id(), self.description)
|
| - self.assertEqual(self.holder.shortDescription(), self.description)
|
| -
|
| -
|
| - def test_holderImplementsITestCase(self):
|
| - """
|
| - L{runner.TestHolder} implements L{ITestCase}.
|
| - """
|
| - self.assertIdentical(self.holder, ITestCase(self.holder))
|
| -
|
| -
|
| -
|
| -class TestErrorHolder(TestTestHolder):
|
| - """
|
| - Test L{runner.ErrorHolder} shares behaviour with L{runner.TestHolder}.
|
| - """
|
| -
|
| - def setUp(self):
|
| - self.description = "description"
|
| - # make a real Failure so we can construct ErrorHolder()
|
| - try:
|
| - 1/0
|
| - except ZeroDivisionError:
|
| - error = failure.Failure()
|
| - self.holder = runner.ErrorHolder(self.description, error)
|
| -
|
| -
|
| -
|
| -class TestMalformedMethod(unittest.TestCase):
|
| - """
|
| - Test that trial manages when test methods don't have correct signatures.
|
| - """
|
| - class ContainMalformed(unittest.TestCase):
|
| - """
|
| - This TestCase holds malformed test methods that trial should handle.
|
| - """
|
| - def test_foo(self, blah):
|
| - pass
|
| - def test_bar():
|
| - pass
|
| - test_spam = defer.deferredGenerator(test_bar)
|
| -
|
| - def _test(self, method):
|
| - """
|
| - Wrapper for one of the test method of L{ContainMalformed}.
|
| - """
|
| - stream = StringIO.StringIO()
|
| - trialRunner = runner.TrialRunner(reporter.Reporter, stream=stream)
|
| - test = TestMalformedMethod.ContainMalformed(method)
|
| - result = trialRunner.run(test)
|
| - self.failUnlessEqual(result.testsRun, 1)
|
| - self.failIf(result.wasSuccessful())
|
| - self.failUnlessEqual(len(result.errors), 1)
|
| -
|
| - def test_extraArg(self):
|
| - """
|
| - Test when the method has extra (useless) arguments.
|
| - """
|
| - self._test('test_foo')
|
| -
|
| - def test_noArg(self):
|
| - """
|
| - Test when the method doesn't have even self as argument.
|
| - """
|
| - self._test('test_bar')
|
| -
|
| - def test_decorated(self):
|
| - """
|
| - Test a decorated method also fails.
|
| - """
|
| - self._test('test_spam')
|
| -
|
| -
|
| -
|
| -class DestructiveTestSuiteTestCase(unittest.TestCase):
|
| - """
|
| - Test for L{runner.DestructiveTestSuite}.
|
| - """
|
| -
|
| - def test_basic(self):
|
| - """
|
| - Thes destructive test suite should run the tests normally.
|
| - """
|
| - called = []
|
| - class MockTest(unittest.TestCase):
|
| - def test_foo(test):
|
| - called.append(True)
|
| - test = MockTest('test_foo')
|
| - result = reporter.TestResult()
|
| - suite = runner.DestructiveTestSuite([test])
|
| - self.assertEquals(called, [])
|
| - suite.run(result)
|
| - self.assertEquals(called, [True])
|
| - self.assertEquals(suite.countTestCases(), 0)
|
| -
|
| -
|
| - def test_shouldStop(self):
|
| - """
|
| - Test the C{shouldStop} management: raising a C{KeyboardInterrupt} must
|
| - interrupt the suite.
|
| - """
|
| - called = []
|
| - class MockTest(unittest.TestCase):
|
| - def test_foo1(test):
|
| - called.append(1)
|
| - def test_foo2(test):
|
| - raise KeyboardInterrupt()
|
| - def test_foo3(test):
|
| - called.append(2)
|
| - result = reporter.TestResult()
|
| - loader = runner.TestLoader()
|
| - loader.suiteFactory = runner.DestructiveTestSuite
|
| - suite = loader.loadClass(MockTest)
|
| - self.assertEquals(called, [])
|
| - suite.run(result)
|
| - self.assertEquals(called, [1])
|
| - # The last test shouldn't have been run
|
| - self.assertEquals(suite.countTestCases(), 1)
|
| -
|
| -
|
| - def test_cleanup(self):
|
| - """
|
| - Checks that the test suite cleanups its tests during the run, so that
|
| - it ends empty.
|
| - """
|
| - class MockTest(unittest.TestCase):
|
| - def test_foo(test):
|
| - pass
|
| - test = MockTest('test_foo')
|
| - result = reporter.TestResult()
|
| - suite = runner.DestructiveTestSuite([test])
|
| - self.assertEquals(suite.countTestCases(), 1)
|
| - suite.run(result)
|
| - self.assertEquals(suite.countTestCases(), 0)
|
| -
|
| -
|
| -
|
| -class TestRunnerDeprecation(unittest.TestCase):
|
| -
|
| - class FakeReporter(reporter.Reporter):
|
| - """
|
| - Fake reporter that does *not* implement done() but *does* implement
|
| - printErrors, separator, printSummary, stream, write and writeln
|
| - without deprecations.
|
| - """
|
| -
|
| - done = None
|
| - separator = None
|
| - stream = None
|
| -
|
| - def printErrors(self, *args):
|
| - pass
|
| -
|
| - def printSummary(self, *args):
|
| - pass
|
| -
|
| - def write(self, *args):
|
| - pass
|
| -
|
| - def writeln(self, *args):
|
| - pass
|
| -
|
| -
|
| - def test_reporterDeprecations(self):
|
| - """
|
| - The runner emits a warning if it is using a result that doesn't
|
| - implement 'done'.
|
| - """
|
| - trialRunner = runner.TrialRunner(None)
|
| - result = self.FakeReporter()
|
| - trialRunner._makeResult = lambda: result
|
| - def f():
|
| - # We have to use a pyunit test, otherwise we'll get deprecation
|
| - # warnings about using iterate() in a test.
|
| - trialRunner.run(pyunit.TestCase('id'))
|
| - self.assertWarns(
|
| - DeprecationWarning,
|
| - "%s should implement done() but doesn't. Falling back to "
|
| - "printErrors() and friends." % reflect.qual(result.__class__),
|
| - __file__, f)
|
|
|