REDROOM

Navigation

Breadcrumbs

/ / lib / python3 / dist-packages / twisted / trial / test

Create

Create File | Create Folder | Upload Files | Run Command

Edit File Size: 31.50 KB

//lib/python3/dist-packages/twisted/trial/test/test_runner.py

Text Base64

# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.
#
# Maintainer: Jonathan Lange
# Author: Robert Collins

import os
import pdb
import sys
import unittest as pyunit
from io import StringIO
from typing import List

from zope.interface import implementer
from zope.interface.verify import verifyObject

from twisted import plugin
from twisted.internet import defer
from twisted.plugins import twisted_trial
from twisted.python import failure, log, reflect
from twisted.python.filepath import FilePath
from twisted.python.reflect import namedAny
from twisted.scripts import trial
from twisted.trial import reporter, runner, unittest, util
from twisted.trial._asyncrunner import _ForceGarbageCollectionDecorator
from twisted.trial.itrial import IReporter, ITestCase

class CapturingDebugger:
    def __init__(self):
        self._calls = []

def runcall(self, *args, **kwargs):
        self._calls.append("runcall")
        args[0](*args[1:], **kwargs)

@implementer(IReporter)
class CapturingReporter:
    """
    Reporter that keeps a log of all actions performed on it.
    """

stream = None
    tbformat = None
    args = None
    separator = None
    testsRun = None

def __init__(self, stream=None, tbformat=None, rterrors=None, publisher=None):
        """
        Create a capturing reporter.
        """
        self._calls = []
        self.shouldStop = False
        self._stream = stream
        self._tbformat = tbformat
        self._rterrors = rterrors
        self._publisher = publisher

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.
        """

def addError(self, test, error):
        """
        Do nothing. These tests don't care about addError.
        """

def addFailure(self, test, failure):
        """
        Do nothing. These tests don't care about addFailure.
        """

def addExpectedFailure(self, test, failure, todo=None):
        """
        Do nothing. These tests don't care about addExpectedFailure.
        """

def addUnexpectedSuccess(self, test, todo=None):
        """
        Do nothing. These tests don't care about addUnexpectedSuccess.
        """

def addSkip(self, test, reason):
        """
        Do nothing. These tests don't care about addSkip.
        """

def wasSuccessful(self):
        """
        Do nothing. These tests don't care about wasSuccessful.
        """

class TrialRunnerTestsMixin:
    """
    Mixin defining tests for L{runner.TrialRunner}.
    """

def test_empty(self):
        """
        Empty test method, used by the other tests.
        """

def _getObservers(self):
        return log.theLogPublisher.observers

def test_addObservers(self):
        """
        Any log system observers L{TrialRunner.run} adds are removed by the
        time it returns.
        """
        originalCount = len(self._getObservers())
        self.runner.run(self.test)
        newCount = len(self._getObservers())
        self.assertEqual(newCount, originalCount)

def test_logFileAlwaysActive(self):
        """
        Test that a new file is opened on each run.
        """
        logPath = FilePath(self.runner.workingDirectory).child(self.runner.logfile)

for i in range(2):
            self.runner.run(self.test)
            logPath.restat()
            self.assertTrue(logPath.exists())
            # We can demonstrate it is re-opened on the next iteration by
            # deleting the current version.
            logPath.remove()

class TrialRunnerTests(TrialRunnerTestsMixin, unittest.SynchronousTestCase):
    """
    Tests for L{runner.TrialRunner} with the feature to turn unclean errors
    into warnings disabled.
    """

def setUp(self):
        self.stream = StringIO()
        self.runner = runner.TrialRunner(CapturingReporter, stream=self.stream)
        self.test = TrialRunnerTests("test_empty")

def test_publisher(self):
        """
        The reporter constructed by L{runner.TrialRunner} is passed
        L{twisted.python.log} as the value for the C{publisher} parameter.
        """
        result = self.runner._makeResult()
        self.assertIdentical(result._publisher, log)

class TrialRunnerWithUncleanWarningsReporterTests(
    TrialRunnerTestsMixin, unittest.SynchronousTestCase
):
    """
    Tests for the TrialRunner's interaction with an unclean-error suppressing
    reporter.
    """

def setUp(self):
        self.stream = StringIO()
        self.runner = runner.TrialRunner(
            CapturingReporter, stream=self.stream, uncleanWarnings=True
        )
        self.test = TrialRunnerTests("test_empty")

class DryRunMixin:
    """
    Mixin for testing that 'dry run' mode works with various
    L{pyunit.TestCase} subclasses.
    """

def setUp(self):
        self.log = []
        self.stream = 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 SynchronousDryRunTests(DryRunMixin, unittest.SynchronousTestCase):
    """
    Check that 'dry run' mode works well with trial's L{SynchronousTestCase}.
    """

def makeTestFixtures(self):
        class PyunitCase(unittest.SynchronousTestCase):
            def test_foo(self):
                pass

self.test = PyunitCase("test_foo")
        self.suite = pyunit.TestSuite()

class DryRunTests(DryRunMixin, unittest.SynchronousTestCase):
    """
    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 PyUnitDryRunTests(DryRunMixin, unittest.SynchronousTestCase):
    """
    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 RunnerTests(unittest.SynchronousTestCase):
    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

# twisted.trial.test.sample has 10 test cases in it so the "standard"
        # report expected from running it will have 10 repetitions of this
        # sequence.
        self.standardReport = ["startTest", "addSuccess", "stopTest"] * 10

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()
        return r

def test_runner_can_get_reporter(self):
        self.parseOptions([])
        result = self.config["reporter"]
        runner = self.getRunner()
        self.assertEqual(result, runner._makeResult().__class__)

def test_runner_get_result(self):
        self.parseOptions([])
        runner = self.getRunner()
        result = 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.DirtyReactorTests}.
        """
        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.DirtyReactorTests} 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()
        self.assertEqual(runner.workingDirectory, "some_path")

def test_concurrentImplicitWorkingDirectory(self):
        """
        If no working directory is explicitly specified and the default
        working directory is in use by another runner, L{TrialRunner.run}
        selects a different default working directory to use.
        """
        self.parseOptions([])

# Make sure we end up with the same working directory after this test
        # as we had before it.
        self.addCleanup(os.chdir, os.getcwd())

# Make a new directory and change into it.  This isolates us from state
        # that other tests might have dumped into this process's temp
        # directory.
        runDirectory = FilePath(self.mktemp())
        runDirectory.makedirs()
        os.chdir(runDirectory.path)

firstRunner = self.getRunner()
        secondRunner = self.getRunner()

where = {}

class ConcurrentCase(unittest.SynchronousTestCase):
            def test_first(self):
                """
                Start a second test run which will have a default working
                directory which is the same as the working directory of the
                test run already in progress.
                """
                # Change the working directory to the value it had before this
                # test suite was started.
                where["concurrent"] = subsequentDirectory = os.getcwd()
                os.chdir(runDirectory.path)
                self.addCleanup(os.chdir, subsequentDirectory)

secondRunner.run(ConcurrentCase("test_second"))

def test_second(self):
                """
                Record the working directory for later analysis.
                """
                where["record"] = os.getcwd()

result = firstRunner.run(ConcurrentCase("test_first"))
        bad = result.errors + result.failures
        if bad:
            self.fail(bad[0][1])
        self.assertEqual(
            where,
            {
                "concurrent": runDirectory.child("_trial_temp").path,
                "record": runDirectory.child("_trial_temp-1").path,
            },
        )

def test_concurrentExplicitWorkingDirectory(self):
        """
        If a working directory which is already in use is explicitly specified,
        L{TrialRunner.run} raises L{_WorkingDirectoryBusy}.
        """
        self.parseOptions(["--temp-directory", os.path.abspath(self.mktemp())])

initialDirectory = os.getcwd()
        self.addCleanup(os.chdir, initialDirectory)

firstRunner = self.getRunner()
        secondRunner = self.getRunner()

class ConcurrentCase(unittest.SynchronousTestCase):
            def test_concurrent(self):
                """
                Try to start another runner in the same working directory and
                assert that it raises L{_WorkingDirectoryBusy}.
                """
                self.assertRaises(
                    util._WorkingDirectoryBusy,
                    secondRunner.run,
                    ConcurrentCase("test_failure"),
                )

def test_failure(self):
                """
                Should not be called, always fails.
                """
                self.fail("test_failure should never be called.")

result = firstRunner.run(ConcurrentCase("test_concurrent"))
        bad = result.errors + result.failures
        if bad:
            self.fail(bad[0][1])

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 runSampleSuite(self, my_runner):
        loader = runner.TestLoader()
        suite = loader.loadByName("twisted.trial.test.sample", True)
        return my_runner.run(suite)

def test_runnerDebug(self):
        """
        Trial uses its debugger if the `--debug` option is passed.
        """
        self.parseOptions(
            ["--reporter", "capturing", "--debug", "twisted.trial.test.sample"]
        )
        my_runner = self.getRunner()
        debugger = my_runner.debugger = CapturingDebugger()
        result = self.runSampleSuite(my_runner)
        self.assertEqual(self.standardReport, result._calls)
        self.assertEqual(["runcall"], debugger._calls)

def test_runnerDebuggerDefaultsToPdb(self):
        """
        Trial uses pdb if no debugger is specified by `--debugger`
        """
        self.parseOptions(["--debug", "twisted.trial.test.sample"])
        pdbrcFile = FilePath("pdbrc")
        pdbrcFile.touch()

self.runcall_called = False

def runcall(pdb, suite, result):
            self.runcall_called = True

self.patch(pdb.Pdb, "runcall", runcall)

self.runSampleSuite(self.getRunner())

self.assertTrue(self.runcall_called)

def test_runnerDebuggerWithExplicitlyPassedPdb(self):
        """
        Trial uses pdb if pdb is passed explicitly to the `--debugger` arg.
        """
        self.parseOptions(
            [
                "--reporter",
                "capturing",
                "--debugger",
                "pdb",
                "--debug",
                "twisted.trial.test.sample",
            ]
        )

self.runcall_called = False

def runcall(pdb, suite, result):
            self.runcall_called = True

self.patch(pdb.Pdb, "runcall", runcall)

self.runSampleSuite(self.getRunner())

self.assertTrue(self.runcall_called)

cdebugger = CapturingDebugger()

def test_runnerDebugger(self):
        """
        Trial uses specified debugger if the debugger is available.
        """
        self.parseOptions(
            [
                "--reporter",
                "capturing",
                "--debugger",
                "twisted.trial.test.test_runner.RunnerTests.cdebugger",
                "--debug",
                "twisted.trial.test.sample",
            ]
        )
        my_runner = self.getRunner()
        result = self.runSampleSuite(my_runner)
        self.assertEqual(self.standardReport, result._calls)
        self.assertEqual(["runcall"], my_runner.debugger._calls)

def test_exitfirst(self):
        """
        If trial was passed the C{--exitfirst} option, the constructed test
        result object is wrapped with L{reporter._ExitWrapper}.
        """
        self.parseOptions(["--exitfirst"])
        runner = self.getRunner()
        result = runner._makeResult()
        self.assertIsInstance(result, reporter._ExitWrapper)

class TrialSuiteTests(unittest.SynchronousTestCase):
    def test_imports(self):
        # FIXME, HTF do you test the reactor can be cleaned up ?!!!
        namedAny("twisted.trial.runner.TrialSuite")

class UntilFailureTests(unittest.SynchronousTestCase):
    class FailAfter(pyunit.TestCase):
        """
        A test case that fails when run 3 times in a row.
        """

count: List[None] = []

def test_foo(self):
            self.count.append(None)
            if len(self.count) == 3:
                self.fail("Count reached 3")

def setUp(self):
        UntilFailureTests.FailAfter.count = []
        self.test = UntilFailureTests.FailAfter("test_foo")
        self.stream = 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.assertEqual(result.testsRun, 1)
        self.assertFalse(result.wasSuccessful())
        self.assertEqual(self._getFailures(result), 1)

def _getFailures(self, result):
        """
        Get the number of failures that were reported to a result.
        """
        return len(result.failures)

def test_runUntilFailureDecorate(self):
        """
        C{runUntilFailure} doesn't decorate the tests uselessly: it does it one
        time when run starts, but not at each turn.
        """
        decorated = []

def decorate(test, interface):
            decorated.append((test, interface))
            return test

self.patch(unittest, "decorate", decorate)
        result = self.runner.runUntilFailure(self.test)
        self.assertEqual(result.testsRun, 1)

self.assertEqual(len(decorated), 1)
        self.assertEqual(decorated, [(self.test, ITestCase)])

def test_runUntilFailureForceGCDecorate(self):
        """
        C{runUntilFailure} applies the force-gc decoration after the standard
        L{ITestCase} decoration, but only one time.
        """
        decorated = []

def decorate(test, interface):
            decorated.append((test, interface))
            return test

self.patch(unittest, "decorate", decorate)
        self.runner._forceGarbageCollection = True
        result = self.runner.runUntilFailure(self.test)
        self.assertEqual(result.testsRun, 1)

self.assertEqual(len(decorated), 2)
        self.assertEqual(
            decorated,
            [(self.test, ITestCase), (self.test, _ForceGarbageCollectionDecorator)],
        )

class UncleanUntilFailureTests(UntilFailureTests):
    """
    Test that the run-until-failure feature works correctly with the unclean
    error suppressor.
    """

def setUp(self):
        UntilFailureTests.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:
            log.err(failure.Failure())

class LoggedErrorsTests(unittest.SynchronousTestCase):
    """
    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.assertTrue(result.errors[0][1].check(RuntimeError))

class TestHolderTests(unittest.SynchronousTestCase):
    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))
        self.assertTrue(
            verifyObject(ITestCase, self.holder),
            "%r claims to provide %r but does not do so correctly."
            % (self.holder, ITestCase),
        )

def test_runsWithStandardResult(self):
        """
        A L{runner.TestHolder} can run against the standard Python
        C{TestResult}.
        """
        result = pyunit.TestResult()
        self.holder.run(result)
        self.assertTrue(result.wasSuccessful())
        self.assertEqual(1, result.testsRun)

class ErrorHolderTestsMixin:
    """
    This mixin defines test methods which can be applied to a
    L{runner.ErrorHolder} constructed with either a L{Failure} or a
    C{exc_info}-style tuple.

Subclass this and implement C{setUp} to create C{self.holder} referring to a
    L{runner.ErrorHolder} instance and C{self.error} referring to a L{Failure}
    which the holder holds.
    """

exceptionForTests = ZeroDivisionError("integer division or modulo by zero")

class TestResultStub:
        """
        Stub for L{TestResult}.
        """

def __init__(self):
            self.errors = []

def startTest(self, test):
            pass

def stopTest(self, test):
            pass

def addError(self, test, error):
            self.errors.append((test, error))

def test_runsWithStandardResult(self):
        """
        A L{runner.ErrorHolder} can run against the standard Python
        C{TestResult}.
        """
        result = pyunit.TestResult()
        self.holder.run(result)
        self.assertFalse(result.wasSuccessful())
        self.assertEqual(1, result.testsRun)

def test_run(self):
        """
        L{runner.ErrorHolder} adds an error to the result when run.
        """
        self.holder.run(self.result)
        self.assertEqual(
            self.result.errors,
            [(self.holder, (self.error.type, self.error.value, self.error.tb))],
        )

def test_call(self):
        """
        L{runner.ErrorHolder} adds an error to the result when called.
        """
        self.holder(self.result)
        self.assertEqual(
            self.result.errors,
            [(self.holder, (self.error.type, self.error.value, self.error.tb))],
        )

def test_countTestCases(self):
        """
        L{runner.ErrorHolder.countTestCases} always returns 0.
        """
        self.assertEqual(self.holder.countTestCases(), 0)

def test_repr(self):
        """
        L{runner.ErrorHolder.__repr__} returns a string describing the error it
        holds.
        """
        expected = "<ErrorHolder description='description' " "error={}>".format(
            repr(self.holder.error[1])
        )

self.assertEqual(repr(self.holder), expected)

class FailureHoldingErrorHolderTests(ErrorHolderTestsMixin, TestHolderTests):
    """
    Tests for L{runner.ErrorHolder} behaving similarly to L{runner.TestHolder}
    when constructed with a L{Failure} representing its error.
    """

def setUp(self):
        self.description = "description"
        # make a real Failure so we can construct ErrorHolder()
        try:
            raise self.exceptionForTests
        except ZeroDivisionError:
            self.error = failure.Failure()
        self.holder = runner.ErrorHolder(self.description, self.error)
        self.result = self.TestResultStub()

class ExcInfoHoldingErrorHolderTests(ErrorHolderTestsMixin, TestHolderTests):
    """
    Tests for L{runner.ErrorHolder} behaving similarly to L{runner.TestHolder}
    when constructed with a C{exc_info}-style tuple representing its error.
    """

def setUp(self):
        self.description = "description"
        # make a real Failure so we can construct ErrorHolder()
        try:
            raise self.exceptionForTests
        except ZeroDivisionError:
            exceptionInfo = sys.exc_info()
            self.error = failure.Failure()
        self.holder = runner.ErrorHolder(self.description, exceptionInfo)
        self.result = self.TestResultStub()

class MalformedMethodTests(unittest.SynchronousTestCase):
    """
    Test that trial manages when test methods don't have correct signatures.
    """

class ContainMalformed(pyunit.TestCase):
        """
        This TestCase holds malformed test methods that trial should handle.
        """

def test_foo(self, blah):
            pass

def test_bar():  # type: ignore[misc]
            pass

test_spam = defer.inlineCallbacks(test_bar)

def _test(self, method):
        """
        Wrapper for one of the test method of L{ContainMalformed}.
        """
        stream = StringIO()
        trialRunner = runner.TrialRunner(reporter.Reporter, stream=stream)
        test = MalformedMethodTests.ContainMalformed(method)
        result = trialRunner.run(test)
        self.assertEqual(result.testsRun, 1)
        self.assertFalse(result.wasSuccessful())
        self.assertEqual(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 DestructiveTestSuiteTests(unittest.SynchronousTestCase):
    """
    Test for L{runner.DestructiveTestSuite}.
    """

def test_basic(self):
        """
        Thes destructive test suite should run the tests normally.
        """
        called = []

class MockTest(pyunit.TestCase):
            def test_foo(test):
                called.append(True)

test = MockTest("test_foo")
        result = reporter.TestResult()
        suite = runner.DestructiveTestSuite([test])
        self.assertEqual(called, [])
        suite.run(result)
        self.assertEqual(called, [True])
        self.assertEqual(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.assertEqual(called, [])
        suite.run(result)
        self.assertEqual(called, [1])
        # The last test shouldn't have been run
        self.assertEqual(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(pyunit.TestCase):
            def test_foo(test):
                pass

test = MockTest("test_foo")
        result = reporter.TestResult()
        suite = runner.DestructiveTestSuite([test])
        self.assertEqual(suite.countTestCases(), 1)
        suite.run(result)
        self.assertEqual(suite.countTestCases(), 0)

class RunnerDeprecationTests(unittest.SynchronousTestCase):
    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

class QualifiedNameWalkerTests(unittest.SynchronousTestCase):
    """
    Tests for L{twisted.trial.runner._qualNameWalker}.
    """

def test_walksDownPath(self):
        """
        C{_qualNameWalker} is a generator that, when given a Python qualified
        name, yields that name, and then the parent of that name, and so forth,
        along with a list of the tried components, in a 2-tuple.
        """
        walkerResults = list(runner._qualNameWalker("walker.texas.ranger"))

self.assertEqual(
            walkerResults,
            [
                ("walker.texas.ranger", []),
                ("walker.texas", ["ranger"]),
                ("walker", ["texas", "ranger"]),
            ],
        )

class TrialMainDoesNothingTests(unittest.SynchronousTestCase):
    """
    Importing L{twisted.trial.__main__} will not run the script
    unless it is actually C{__main__}.
    """

def test_importDoesNothing(self):
        """
        If we import L{twisted.trial.__main__}, it should do nothing.
        """
        # We shouldn't suddenly drop into a script when we import this!
        __import__("twisted.trial.__main__")

Exit & Reset Text mode: syntax highlighting auto-detects file type.

Directory Contents

Dirs: 1 × Files: 39

Sort:

Enable drag-to-move

Name	Size	Perms	Modified	Actions
__pycache__ DIR	-	drwxr-xr-x	2026-01-08 12:56:24	Edit Download recursive recursive
detests.py	5.63 KB	lrw-r--r--	2024-08-27 10:30:39	Edit Download recursive recursive
erroneous.py	6.44 KB	lrw-r--r--	2024-08-27 10:30:39	Edit Download recursive recursive
matchers.py	2.88 KB	lrw-r--r--	2024-08-27 10:30:39	Edit Download recursive recursive
mockcustomsuite.py	544 B	lrw-r--r--	2024-08-27 10:30:39	Edit Download recursive recursive
mockcustomsuite2.py	541 B	lrw-r--r--	2024-08-27 10:30:39	Edit Download recursive recursive
mockcustomsuite3.py	684 B	lrw-r--r--	2024-08-27 10:30:39	Edit Download recursive recursive
mockdoctest.py	2.36 KB	lrw-r--r--	2024-08-27 10:30:39	Edit Download recursive recursive
moduleself.py	178 B	lrw-r--r--	2024-08-27 10:30:39	Edit Download recursive recursive
moduletest.py	302 B	lrw-r--r--	2024-08-27 10:30:39	Edit Download recursive recursive
novars.py	182 B	lrw-r--r--	2024-08-27 10:30:39	Edit Download recursive recursive
ordertests.py	912 B	lrw-r--r--	2024-08-27 10:30:39	Edit Download recursive recursive
packages.py	4.54 KB	lrw-r--r--	2024-08-27 10:30:39	Edit Download recursive recursive
pyunitcases.py	3.10 KB	lrw-r--r--	2024-08-27 10:30:39	Edit Download recursive recursive
sample.py	2.13 KB	lrw-r--r--	2024-08-27 10:30:39	Edit Download recursive recursive
scripttest.py	457 B	lrwxr-xr-x	2024-08-27 10:30:39	Edit Download recursive recursive
skipping.py	5.99 KB	lrw-r--r--	2024-08-27 10:30:39	Edit Download recursive recursive
suppression.py	2.44 KB	lrw-r--r--	2024-08-27 10:30:39	Edit Download recursive recursive
test_assertions.py	59.39 KB	lrw-r--r--	2024-08-27 10:30:39	Edit Download recursive recursive
test_asyncassertions.py	2.49 KB	lrw-r--r--	2024-08-27 10:30:39	Edit Download recursive recursive
test_deferred.py	9.63 KB	lrw-r--r--	2024-08-27 10:30:39	Edit Download recursive recursive
test_doctest.py	1.76 KB	lrw-r--r--	2024-08-27 10:30:39	Edit Download recursive recursive
test_keyboard.py	3.96 KB	lrw-r--r--	2024-08-27 10:30:39	Edit Download recursive recursive
test_loader.py	23.95 KB	lrw-r--r--	2024-08-27 10:30:39	Edit Download recursive recursive
test_log.py	7.85 KB	lrw-r--r--	2024-08-27 10:30:39	Edit Download recursive recursive
test_matchers.py	2.97 KB	lrw-r--r--	2024-08-27 10:30:39	Edit Download recursive recursive
test_output.py	5.26 KB	lrw-r--r--	2024-08-27 10:30:39	Edit Download recursive recursive
test_plugins.py	1.43 KB	lrw-r--r--	2024-08-27 10:30:39	Edit Download recursive recursive
test_pyunitcompat.py	7.88 KB	lrw-r--r--	2024-08-27 10:30:39	Edit Download recursive recursive
test_reporter.py	56.44 KB	lrw-r--r--	2024-08-27 10:30:39	Edit Download recursive recursive
test_runner.py	31.50 KB	lrw-r--r--	2024-08-27 10:30:39	Edit Download recursive recursive
test_script.py	32.41 KB	lrw-r--r--	2024-08-27 10:30:39	Edit Download recursive recursive
test_skip.py	2.69 KB	lrw-r--r--	2024-08-27 10:30:39	Edit Download recursive recursive
test_suppression.py	5.77 KB	lrw-r--r--	2024-08-27 10:30:39	Edit Download recursive recursive
test_testcase.py	1.94 KB	lrw-r--r--	2024-08-27 10:30:39	Edit Download recursive recursive
test_tests.py	48.83 KB	lrw-r--r--	2024-08-27 10:30:39	Edit Download recursive recursive
test_util.py	21.65 KB	lrw-r--r--	2024-08-27 10:30:39	Edit Download recursive recursive
test_warning.py	18.40 KB	lrw-r--r--	2024-08-27 10:30:39	Edit Download recursive recursive
weird.py	675 B	lrw-r--r--	2024-08-27 10:30:39	Edit Download recursive recursive
__init__.py	1.68 KB	lrw-r--r--	2024-08-27 10:30:39	Edit Download recursive recursive