path: root/lib/python2.7/site-packages/Twisted-12.2.0-py2.7-linux-x86_64.egg/twisted/trial/test/test_util.py

# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.
#

"""
Tests for L{twisted.trial.util}
"""

import os

from zope.interface import implements

from twisted.internet.interfaces import IProcessTransport
from twisted.internet import defer
from twisted.internet.base import DelayedCall

from twisted.trial.unittest import SynchronousTestCase, TestCase
from twisted.trial import util
from twisted.trial.util import DirtyReactorAggregateError, _Janitor
from twisted.trial.test import packages, suppression



class TestMktemp(SynchronousTestCase):
    """
    Tests for L{TestCase.mktemp}, a helper function for creating temporary file
    or directory names.
    """
    def test_name(self):
        """
        The path name returned by C{mktemp} is directly beneath a directory
        which identifies the test method which created the name.
        """
        name = self.mktemp()
        dirs = os.path.dirname(name).split(os.sep)[:-1]
        self.assertEqual(
            dirs, ['twisted.trial.test.test_util', 'TestMktemp', 'test_name'])


    def test_unique(self):
        """
        Repeated calls to C{mktemp} return different values.
        """
        name = self.mktemp()
        self.assertNotEqual(name, self.mktemp())


    def test_created(self):
        """
        The directory part of the path name returned by C{mktemp} exists.
        """
        name = self.mktemp()
        dirname = os.path.dirname(name)
        self.assertTrue(os.path.exists(dirname))
        self.assertFalse(os.path.exists(name))


    def test_location(self):
        """
        The path returned by C{mktemp} is beneath the current working directory.
        """
        path = os.path.abspath(self.mktemp())
        self.assertTrue(path.startswith(os.getcwd()))



class TestIntrospection(SynchronousTestCase):
    def test_containers(self):
        """
        When pased a test case, L{util.getPythonContainers} returns a list
        including the test case and the module the test case is defined in.
        """
        parents = util.getPythonContainers(
            suppression.SynchronousTestSuppression2.testSuppressModule)
        expected = [suppression.SynchronousTestSuppression2, suppression]
        for a, b in zip(parents, expected):
            self.assertEqual(a, b)



class TestFindObject(packages.SysPathManglingTest):
    """
    Tests for L{twisted.trial.util.findObject}
    """

    def test_deprecation(self):
        """
        Calling L{findObject} results in a deprecation warning
        """
        util.findObject('')
        warningsShown = self.flushWarnings()
        self.assertEqual(len(warningsShown), 1)
        self.assertIdentical(warningsShown[0]['category'], DeprecationWarning)
        self.assertEqual(warningsShown[0]['message'],
                          "twisted.trial.util.findObject was deprecated "
                          "in Twisted 10.1.0: Please use "
                          "twisted.python.reflect.namedAny instead.")


    def test_importPackage(self):
        package1 = util.findObject('package')
        import package as package2
        self.assertEqual(package1, (True, package2))

    def test_importModule(self):
        test_sample2 = util.findObject('goodpackage.test_sample')
        from goodpackage import test_sample
        self.assertEqual((True, test_sample), test_sample2)

    def test_importError(self):
        self.failUnlessRaises(ZeroDivisionError,
                              util.findObject, 'package.test_bad_module')

    def test_sophisticatedImportError(self):
        self.failUnlessRaises(ImportError,
                              util.findObject, 'package2.test_module')

    def test_importNonexistentPackage(self):
        self.assertEqual(util.findObject('doesntexist')[0], False)

    def test_findNonexistentModule(self):
        self.assertEqual(util.findObject('package.doesntexist')[0], False)

    def test_findNonexistentObject(self):
        self.assertEqual(util.findObject(
            'goodpackage.test_sample.doesnt')[0], False)
        self.assertEqual(util.findObject(
            'goodpackage.test_sample.AlphabetTest.doesntexist')[0], False)

    def test_findObjectExist(self):
        alpha1 = util.findObject('goodpackage.test_sample.AlphabetTest')
        from goodpackage import test_sample
        self.assertEqual(alpha1, (True, test_sample.AlphabetTest))



class TestRunSequentially(TestCase):
    """
    Sometimes it is useful to be able to run an arbitrary list of callables,
    one after the other.

    When some of those callables can return Deferreds, things become complex.
    """

    def test_emptyList(self):
        """
        When asked to run an empty list of callables, runSequentially returns a
        successful Deferred that fires an empty list.
        """
        d = util._runSequentially([])
        d.addCallback(self.assertEqual, [])
        return d


    def test_singleSynchronousSuccess(self):
        """
        When given a callable that succeeds without returning a Deferred,
        include the return value in the results list, tagged with a SUCCESS
        flag.
        """
        d = util._runSequentially([lambda: None])
        d.addCallback(self.assertEqual, [(defer.SUCCESS, None)])
        return d


    def test_singleSynchronousFailure(self):
        """
        When given a callable that raises an exception, include a Failure for
        that exception in the results list, tagged with a FAILURE flag.
        """
        d = util._runSequentially([lambda: self.fail('foo')])
        def check(results):
            [(flag, fail)] = results
            fail.trap(self.failureException)
            self.assertEqual(fail.getErrorMessage(), 'foo')
            self.assertEqual(flag, defer.FAILURE)
        return d.addCallback(check)


    def test_singleAsynchronousSuccess(self):
        """
        When given a callable that returns a successful Deferred, include the
        result of the Deferred in the results list, tagged with a SUCCESS flag.
        """
        d = util._runSequentially([lambda: defer.succeed(None)])
        d.addCallback(self.assertEqual, [(defer.SUCCESS, None)])
        return d


    def test_singleAsynchronousFailure(self):
        """
        When given a callable that returns a failing Deferred, include the
        failure the results list, tagged with a FAILURE flag.
        """
        d = util._runSequentially([lambda: defer.fail(ValueError('foo'))])
        def check(results):
            [(flag, fail)] = results
            fail.trap(ValueError)
            self.assertEqual(fail.getErrorMessage(), 'foo')
            self.assertEqual(flag, defer.FAILURE)
        return d.addCallback(check)


    def test_callablesCalledInOrder(self):
        """
        Check that the callables are called in the given order, one after the
        other.
        """
        log = []
        deferreds = []

        def append(value):
            d = defer.Deferred()
            log.append(value)
            deferreds.append(d)
            return d

        d = util._runSequentially([lambda: append('foo'),
                                   lambda: append('bar')])

        # runSequentially should wait until the Deferred has fired before
        # running the second callable.
        self.assertEqual(log, ['foo'])
        deferreds[-1].callback(None)
        self.assertEqual(log, ['foo', 'bar'])

        # Because returning created Deferreds makes jml happy.
        deferreds[-1].callback(None)
        return d


    def test_continuesAfterError(self):
        """
        If one of the callables raises an error, then runSequentially continues
        to run the remaining callables.
        """
        d = util._runSequentially([lambda: self.fail('foo'), lambda: 'bar'])
        def check(results):
            [(flag1, fail), (flag2, result)] = results
            fail.trap(self.failureException)
            self.assertEqual(flag1, defer.FAILURE)
            self.assertEqual(fail.getErrorMessage(), 'foo')
            self.assertEqual(flag2, defer.SUCCESS)
            self.assertEqual(result, 'bar')
        return d.addCallback(check)


    def test_stopOnFirstError(self):
        """
        If the C{stopOnFirstError} option is passed to C{runSequentially}, then
        no further callables are called after the first exception is raised.
        """
        d = util._runSequentially([lambda: self.fail('foo'), lambda: 'bar'],
                                  stopOnFirstError=True)
        def check(results):
            [(flag1, fail)] = results
            fail.trap(self.failureException)
            self.assertEqual(flag1, defer.FAILURE)
            self.assertEqual(fail.getErrorMessage(), 'foo')
        return d.addCallback(check)


    def test_stripFlags(self):
        """
        If the C{stripFlags} option is passed to C{runSequentially} then the
        SUCCESS / FAILURE flags are stripped from the output. Instead, the
        Deferred fires a flat list of results containing only the results and
        failures.
        """
        d = util._runSequentially([lambda: self.fail('foo'), lambda: 'bar'],
                                  stripFlags=True)
        def check(results):
            [fail, result] = results
            fail.trap(self.failureException)
            self.assertEqual(fail.getErrorMessage(), 'foo')
            self.assertEqual(result, 'bar')
        return d.addCallback(check)
    test_stripFlags.todo = "YAGNI"



class DirtyReactorAggregateErrorTest(SynchronousTestCase):
    """
    Tests for the L{DirtyReactorAggregateError}.
    """

    def test_formatDelayedCall(self):
        """
        Delayed calls are formatted nicely.
        """
        error = DirtyReactorAggregateError(["Foo", "bar"])
        self.assertEqual(str(error),
                          """\
Reactor was unclean.
DelayedCalls: (set twisted.internet.base.DelayedCall.debug = True to debug)
Foo
bar""")


    def test_formatSelectables(self):
        """
        Selectables are formatted nicely.
        """
        error = DirtyReactorAggregateError([], ["selectable 1", "selectable 2"])
        self.assertEqual(str(error),
                          """\
Reactor was unclean.
Selectables:
selectable 1
selectable 2""")


    def test_formatDelayedCallsAndSelectables(self):
        """
        Both delayed calls and selectables can appear in the same error.
        """
        error = DirtyReactorAggregateError(["bleck", "Boozo"],
                                           ["Sel1", "Sel2"])
        self.assertEqual(str(error),
                          """\
Reactor was unclean.
DelayedCalls: (set twisted.internet.base.DelayedCall.debug = True to debug)
bleck
Boozo
Selectables:
Sel1
Sel2""")



class StubReactor(object):
    """
    A reactor stub which contains enough functionality to be used with the
    L{_Janitor}.

    @ivar iterations: A list of the arguments passed to L{iterate}.
    @ivar removeAllCalled: Number of times that L{removeAll} was called.
    @ivar selectables: The value that will be returned from L{removeAll}.
    @ivar delayedCalls: The value to return from L{getDelayedCalls}.
    """

    def __init__(self, delayedCalls, selectables=None):
        """
        @param delayedCalls: See L{StubReactor.delayedCalls}.
        @param selectables: See L{StubReactor.selectables}.
        """
        self.delayedCalls = delayedCalls
        self.iterations = []
        self.removeAllCalled = 0
        if not selectables:
            selectables = []
        self.selectables = selectables


    def iterate(self, timeout=None):
        """
        Increment C{self.iterations}.
        """
        self.iterations.append(timeout)


    def getDelayedCalls(self):
        """
        Return C{self.delayedCalls}.
        """
        return self.delayedCalls


    def removeAll(self):
        """
        Increment C{self.removeAllCalled} and return C{self.selectables}.
        """
        self.removeAllCalled += 1
        return self.selectables



class StubErrorReporter(object):
    """
    A subset of L{twisted.trial.itrial.IReporter} which records L{addError}
    calls.

    @ivar errors: List of two-tuples of (test, error) which were passed to
        L{addError}.
    """

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


    def addError(self, test, error):
        """
        Record parameters in C{self.errors}.
        """
        self.errors.append((test, error))



class JanitorTests(SynchronousTestCase):
    """
    Tests for L{_Janitor}!
    """

    def test_cleanPendingSpinsReactor(self):
        """
        During pending-call cleanup, the reactor will be spun twice with an
        instant timeout. This is not a requirement, it is only a test for
        current behavior. Hopefully Trial will eventually not do this kind of
        reactor stuff.
        """
        reactor = StubReactor([])
        jan = _Janitor(None, None, reactor=reactor)
        jan._cleanPending()
        self.assertEqual(reactor.iterations, [0, 0])


    def test_cleanPendingCancelsCalls(self):
        """
        During pending-call cleanup, the janitor cancels pending timed calls.
        """
        def func():
            return "Lulz"
        cancelled = []
        delayedCall = DelayedCall(300, func, (), {},
                                  cancelled.append, lambda x: None)
        reactor = StubReactor([delayedCall])
        jan = _Janitor(None, None, reactor=reactor)
        jan._cleanPending()
        self.assertEqual(cancelled, [delayedCall])


    def test_cleanPendingReturnsDelayedCallStrings(self):
        """
        The Janitor produces string representations of delayed calls from the
        delayed call cleanup method. It gets the string representations
        *before* cancelling the calls; this is important because cancelling the
        call removes critical debugging information from the string
        representation.
        """
        delayedCall = DelayedCall(300, lambda: None, (), {},
                                  lambda x: None, lambda x: None,
                                  seconds=lambda: 0)
        delayedCallString = str(delayedCall)
        reactor = StubReactor([delayedCall])
        jan = _Janitor(None, None, reactor=reactor)
        strings = jan._cleanPending()
        self.assertEqual(strings, [delayedCallString])


    def test_cleanReactorRemovesSelectables(self):
        """
        The Janitor will remove selectables during reactor cleanup.
        """
        reactor = StubReactor([])
        jan = _Janitor(None, None, reactor=reactor)
        jan._cleanReactor()
        self.assertEqual(reactor.removeAllCalled, 1)


    def test_cleanReactorKillsProcesses(self):
        """
        The Janitor will kill processes during reactor cleanup.
        """
        class StubProcessTransport(object):
            """
            A stub L{IProcessTransport} provider which records signals.
            @ivar signals: The signals passed to L{signalProcess}.
            """
            implements(IProcessTransport)

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

            def signalProcess(self, signal):
                """
                Append C{signal} to C{self.signals}.
                """
                self.signals.append(signal)

        pt = StubProcessTransport()
        reactor = StubReactor([], [pt])
        jan = _Janitor(None, None, reactor=reactor)
        jan._cleanReactor()
        self.assertEqual(pt.signals, ["KILL"])


    def test_cleanReactorReturnsSelectableStrings(self):
        """
        The Janitor returns string representations of the selectables that it
        cleaned up from the reactor cleanup method.
        """
        class Selectable(object):
            """
            A stub Selectable which only has an interesting string
            representation.
            """
            def __repr__(self):
                return "(SELECTABLE!)"

        reactor = StubReactor([], [Selectable()])
        jan = _Janitor(None, None, reactor=reactor)
        self.assertEqual(jan._cleanReactor(), ["(SELECTABLE!)"])


    def test_postCaseCleanupNoErrors(self):
        """
        The post-case cleanup method will return True and not call C{addError}
        on the result if there are no pending calls.
        """
        reactor = StubReactor([])
        test = object()
        reporter = StubErrorReporter()
        jan = _Janitor(test, reporter, reactor=reactor)
        self.assertTrue(jan.postCaseCleanup())
        self.assertEqual(reporter.errors, [])


    def test_postCaseCleanupWithErrors(self):
        """
        The post-case cleanup method will return False and call C{addError} on
        the result with a L{DirtyReactorAggregateError} Failure if there are
        pending calls.
        """
        delayedCall = DelayedCall(300, lambda: None, (), {},
                                  lambda x: None, lambda x: None,
                                  seconds=lambda: 0)
        delayedCallString = str(delayedCall)
        reactor = StubReactor([delayedCall], [])
        test = object()
        reporter = StubErrorReporter()
        jan = _Janitor(test, reporter, reactor=reactor)
        self.assertFalse(jan.postCaseCleanup())
        self.assertEqual(len(reporter.errors), 1)
        self.assertEqual(reporter.errors[0][1].value.delayedCalls,
                          [delayedCallString])


    def test_postClassCleanupNoErrors(self):
        """
        The post-class cleanup method will not call C{addError} on the result
        if there are no pending calls or selectables.
        """
        reactor = StubReactor([])
        test = object()
        reporter = StubErrorReporter()
        jan = _Janitor(test, reporter, reactor=reactor)
        jan.postClassCleanup()
        self.assertEqual(reporter.errors, [])


    def test_postClassCleanupWithPendingCallErrors(self):
        """
        The post-class cleanup method call C{addError} on the result with a
        L{DirtyReactorAggregateError} Failure if there are pending calls.
        """
        delayedCall = DelayedCall(300, lambda: None, (), {},
                                  lambda x: None, lambda x: None,
                                  seconds=lambda: 0)
        delayedCallString = str(delayedCall)
        reactor = StubReactor([delayedCall], [])
        test = object()
        reporter = StubErrorReporter()
        jan = _Janitor(test, reporter, reactor=reactor)
        jan.postClassCleanup()
        self.assertEqual(len(reporter.errors), 1)
        self.assertEqual(reporter.errors[0][1].value.delayedCalls,
                          [delayedCallString])


    def test_postClassCleanupWithSelectableErrors(self):
        """
        The post-class cleanup method call C{addError} on the result with a
        L{DirtyReactorAggregateError} Failure if there are selectables.
        """
        selectable = "SELECTABLE HERE"
        reactor = StubReactor([], [selectable])
        test = object()
        reporter = StubErrorReporter()
        jan = _Janitor(test, reporter, reactor=reactor)
        jan.postClassCleanup()
        self.assertEqual(len(reporter.errors), 1)
        self.assertEqual(reporter.errors[0][1].value.selectables,
                          [repr(selectable)])