python类directlyProvides()的实例源码

def test_directlyProvides(self):
        class IA1(Interface): pass
        class IA2(Interface): pass
        class IB(Interface): pass
        class IC(Interface): pass
        class A(Odd):
            implements(IA1, IA2)

        class B(Odd):
            implements(IB)

        class C(A, B):
            implements(IC)


        ob = C()
        directlyProvides(ob, I1, I2)
        self.assert_(I1 in providedBy(ob))
        self.assert_(I2 in providedBy(ob))
        self.assert_(IA1 in providedBy(ob))
        self.assert_(IA2 in providedBy(ob))
        self.assert_(IB in providedBy(ob))
        self.assert_(IC in providedBy(ob))

        directlyProvides(ob, directlyProvidedBy(ob)-I2)
        self.assert_(I1 in providedBy(ob))
        self.failIf(I2 in providedBy(ob))
        self.failIf(I2 in providedBy(ob))
        directlyProvides(ob, directlyProvidedBy(ob), I2)
        self.assert_(I2 in providedBy(ob))

def test_directlyProvides_fails_for_odd_class(self):
        self.assertRaises(TypeError, directlyProvides, C, I5)

    # see above

def TODO_test_classProvides_fails_for_odd_class(self):
        try:
            class A(Odd):
                classProvides(I1)
        except TypeError:
            pass # Sucess
        self.assert_(False,
                     "Shouldn't be able to use directlyProvides on odd class."
                     )

def test_directlyProvides(self):
        class IA1(Interface): pass
        class IA2(Interface): pass
        class IB(Interface): pass
        class IC(Interface): pass
        class A(Odd):
            implements(IA1, IA2)

        class B(Odd):
            implements(IB)

        class C(A, B):
            implements(IC)


        ob = C()
        directlyProvides(ob, I1, I2)
        self.assert_(I1 in providedBy(ob))
        self.assert_(I2 in providedBy(ob))
        self.assert_(IA1 in providedBy(ob))
        self.assert_(IA2 in providedBy(ob))
        self.assert_(IB in providedBy(ob))
        self.assert_(IC in providedBy(ob))

        directlyProvides(ob, directlyProvidedBy(ob)-I2)
        self.assert_(I1 in providedBy(ob))
        self.failIf(I2 in providedBy(ob))
        self.failIf(I2 in providedBy(ob))
        directlyProvides(ob, directlyProvidedBy(ob), I2)
        self.assert_(I2 in providedBy(ob))

def test_directlyProvides_fails_for_odd_class(self):
        self.assertRaises(TypeError, directlyProvides, C, I5)

    # see above

def TODO_test_classProvides_fails_for_odd_class(self):
        try:
            class A(Odd):
                classProvides(I1)
        except TypeError:
            pass # Sucess
        self.assert_(False,
                     "Shouldn't be able to use directlyProvides on odd class."
                     )

def test_directlyProvides(self):
        class IA1(Interface): pass
        class IA2(Interface): pass
        class IB(Interface): pass
        class IC(Interface): pass
        class A(Odd):
            pass
        classImplements(A, IA1, IA2)

        class B(Odd):
            pass
        classImplements(B, IB)

        class C(A, B):
            pass
        classImplements(C, IC)


        ob = C()
        directlyProvides(ob, I1, I2)
        self.assertTrue(I1 in providedBy(ob))
        self.assertTrue(I2 in providedBy(ob))
        self.assertTrue(IA1 in providedBy(ob))
        self.assertTrue(IA2 in providedBy(ob))
        self.assertTrue(IB in providedBy(ob))
        self.assertTrue(IC in providedBy(ob))

        directlyProvides(ob, directlyProvidedBy(ob)-I2)
        self.assertTrue(I1 in providedBy(ob))
        self.assertFalse(I2 in providedBy(ob))
        self.assertFalse(I2 in providedBy(ob))
        directlyProvides(ob, directlyProvidedBy(ob), I2)
        self.assertTrue(I2 in providedBy(ob))

def test_directlyProvides_fails_for_odd_class(self):
        self.assertRaises(TypeError, directlyProvides, C, I5)

    # see above
    #def TODO_test_classProvides_fails_for_odd_class(self):
    #    try:
    #        class A(Odd):
    #            classProvides(I1)
    #    except TypeError:
    #        pass # Sucess
    #    self.assert_(False,
    #                 "Shouldn't be able to use directlyProvides on odd class."
    #                 )

def test_directlyProvides(self):
        class IA1(Interface): pass
        class IA2(Interface): pass
        class IB(Interface): pass
        class IC(Interface): pass
        class A(Odd):
            pass
        classImplements(A, IA1, IA2)

        class B(Odd):
            pass
        classImplements(B, IB)

        class C(A, B):
            pass
        classImplements(C, IC)


        ob = C()
        directlyProvides(ob, I1, I2)
        self.assertTrue(I1 in providedBy(ob))
        self.assertTrue(I2 in providedBy(ob))
        self.assertTrue(IA1 in providedBy(ob))
        self.assertTrue(IA2 in providedBy(ob))
        self.assertTrue(IB in providedBy(ob))
        self.assertTrue(IC in providedBy(ob))

        directlyProvides(ob, directlyProvidedBy(ob)-I2)
        self.assertTrue(I1 in providedBy(ob))
        self.assertFalse(I2 in providedBy(ob))
        self.assertFalse(I2 in providedBy(ob))
        directlyProvides(ob, directlyProvidedBy(ob), I2)
        self.assertTrue(I2 in providedBy(ob))

def test_directlyProvides_fails_for_odd_class(self):
        self.assertRaises(TypeError, directlyProvides, C, I5)

    # see above
    #def TODO_test_classProvides_fails_for_odd_class(self):
    #    try:
    #        class A(Odd):
    #            classProvides(I1)
    #    except TypeError:
    #        pass # Sucess
    #    self.assert_(False,
    #                 "Shouldn't be able to use directlyProvides on odd class."
    #                 )

def test_ObjectSpecification(self):
        c = C()
        directlyProvides(c, I4)
        self.assertEqual([i.getName() for i in providedBy(c)],
                         ['I4', 'I31', 'I1', 'I2']
                         )
        self.assertEqual([i.getName() for i in providedBy(c).flattened()],
                         ['I4', 'I31', 'I3', 'I1', 'I2', 'Interface']
                         )
        self.assert_(I1 in providedBy(c))
        self.failIf(I3 in providedBy(c))
        self.assert_(providedBy(c).extends(I3))
        self.assert_(providedBy(c).extends(I31))
        self.failIf(providedBy(c).extends(I5))

        class COnly(A, B):
            implementsOnly(I31)

        class D(COnly):
            implements(I5)

        classImplements(D, I5)

        c = D()
        directlyProvides(c, I4)
        self.assertEqual([i.getName() for i in providedBy(c)],
                         ['I4', 'I5', 'I31'])
        self.assertEqual([i.getName() for i in providedBy(c).flattened()],
                         ['I4', 'I5', 'I31', 'I3', 'Interface'])
        self.failIf(I1 in providedBy(c))
        self.failIf(I3 in providedBy(c))
        self.assert_(providedBy(c).extends(I3))
        self.failIf(providedBy(c).extends(I1))
        self.assert_(providedBy(c).extends(I31))
        self.assert_(providedBy(c).extends(I5))

        class COnly(A, B): __implemented__ = I31
        class D(COnly):
            implements(I5)

        classImplements(D, I5)
        c = D()
        directlyProvides(c, I4)
        self.assertEqual([i.getName() for i in providedBy(c)],
                         ['I4', 'I5', 'I31'])
        self.assertEqual([i.getName() for i in providedBy(c).flattened()],
                         ['I4', 'I5', 'I31', 'I3', 'Interface'])
        self.failIf(I1 in providedBy(c))
        self.failIf(I3 in providedBy(c))
        self.assert_(providedBy(c).extends(I3))
        self.failIf(providedBy(c).extends(I1))
        self.assert_(providedBy(c).extends(I31))
        self.assert_(providedBy(c).extends(I5))

def test_ObjectSpecification(self):
        c = C()
        directlyProvides(c, I4)
        self.assertEqual([i.getName() for i in providedBy(c)],
                         ['I4', 'I31', 'I1', 'I2']
                         )
        self.assertEqual([i.getName() for i in providedBy(c).flattened()],
                         ['I4', 'I31', 'I3', 'I1', 'I2', 'Interface']
                         )
        self.assert_(I1 in providedBy(c))
        self.failIf(I3 in providedBy(c))
        self.assert_(providedBy(c).extends(I3))
        self.assert_(providedBy(c).extends(I31))
        self.failIf(providedBy(c).extends(I5))

        class COnly(A, B):
            implementsOnly(I31)

        class D(COnly):
            implements(I5)

        classImplements(D, I5)

        c = D()
        directlyProvides(c, I4)
        self.assertEqual([i.getName() for i in providedBy(c)],
                         ['I4', 'I5', 'I31'])
        self.assertEqual([i.getName() for i in providedBy(c).flattened()],
                         ['I4', 'I5', 'I31', 'I3', 'Interface'])
        self.failIf(I1 in providedBy(c))
        self.failIf(I3 in providedBy(c))
        self.assert_(providedBy(c).extends(I3))
        self.failIf(providedBy(c).extends(I1))
        self.assert_(providedBy(c).extends(I31))
        self.assert_(providedBy(c).extends(I5))

        class COnly(A, B): __implemented__ = I31
        class D(COnly):
            implements(I5)

        classImplements(D, I5)
        c = D()
        directlyProvides(c, I4)
        self.assertEqual([i.getName() for i in providedBy(c)],
                         ['I4', 'I5', 'I31'])
        self.assertEqual([i.getName() for i in providedBy(c).flattened()],
                         ['I4', 'I5', 'I31', 'I3', 'Interface'])
        self.failIf(I1 in providedBy(c))
        self.failIf(I3 in providedBy(c))
        self.assert_(providedBy(c).extends(I3))
        self.failIf(providedBy(c).extends(I1))
        self.assert_(providedBy(c).extends(I31))
        self.assert_(providedBy(c).extends(I5))

def test_ObjectSpecification(self):
        c = C()
        directlyProvides(c, I4)
        self.assertEqual([i.getName() for i in providedBy(c)],
                         ['I4', 'I31', 'I1', 'I2']
                         )
        self.assertEqual([i.getName() for i in providedBy(c).flattened()],
                         ['I4', 'I31', 'I3', 'I1', 'I2', 'Interface']
                         )
        self.assertTrue(I1 in providedBy(c))
        self.assertFalse(I3 in providedBy(c))
        self.assertTrue(providedBy(c).extends(I3))
        self.assertTrue(providedBy(c).extends(I31))
        self.assertFalse(providedBy(c).extends(I5))

        class COnly(A, B):
            pass
        classImplementsOnly(COnly, I31)

        class D(COnly):
            pass
        classImplements(D, I5)

        classImplements(D, I5)

        c = D()
        directlyProvides(c, I4)
        self.assertEqual([i.getName() for i in providedBy(c)],
                         ['I4', 'I5', 'I31'])
        self.assertEqual([i.getName() for i in providedBy(c).flattened()],
                         ['I4', 'I5', 'I31', 'I3', 'Interface'])
        self.assertFalse(I1 in providedBy(c))
        self.assertFalse(I3 in providedBy(c))
        self.assertTrue(providedBy(c).extends(I3))
        self.assertFalse(providedBy(c).extends(I1))
        self.assertTrue(providedBy(c).extends(I31))
        self.assertTrue(providedBy(c).extends(I5))

        class COnly(A, B): __implemented__ = I31
        class D(COnly):
            pass
        classImplements(D, I5)

        classImplements(D, I5)
        c = D()
        directlyProvides(c, I4)
        self.assertEqual([i.getName() for i in providedBy(c)],
                         ['I4', 'I5', 'I31'])
        self.assertEqual([i.getName() for i in providedBy(c).flattened()],
                         ['I4', 'I5', 'I31', 'I3', 'Interface'])
        self.assertFalse(I1 in providedBy(c))
        self.assertFalse(I3 in providedBy(c))
        self.assertTrue(providedBy(c).extends(I3))
        self.assertFalse(providedBy(c).extends(I1))
        self.assertTrue(providedBy(c).extends(I31))
        self.assertTrue(providedBy(c).extends(I5))

def test_ObjectSpecification(self):
        c = C()
        directlyProvides(c, I4)
        self.assertEqual([i.getName() for i in providedBy(c)],
                         ['I4', 'I31', 'I1', 'I2']
                         )
        self.assertEqual([i.getName() for i in providedBy(c).flattened()],
                         ['I4', 'I31', 'I3', 'I1', 'I2', 'Interface']
                         )
        self.assertTrue(I1 in providedBy(c))
        self.assertFalse(I3 in providedBy(c))
        self.assertTrue(providedBy(c).extends(I3))
        self.assertTrue(providedBy(c).extends(I31))
        self.assertFalse(providedBy(c).extends(I5))

        class COnly(A, B):
            pass
        classImplementsOnly(COnly, I31)

        class D(COnly):
            pass
        classImplements(D, I5)

        classImplements(D, I5)

        c = D()
        directlyProvides(c, I4)
        self.assertEqual([i.getName() for i in providedBy(c)],
                         ['I4', 'I5', 'I31'])
        self.assertEqual([i.getName() for i in providedBy(c).flattened()],
                         ['I4', 'I5', 'I31', 'I3', 'Interface'])
        self.assertFalse(I1 in providedBy(c))
        self.assertFalse(I3 in providedBy(c))
        self.assertTrue(providedBy(c).extends(I3))
        self.assertFalse(providedBy(c).extends(I1))
        self.assertTrue(providedBy(c).extends(I31))
        self.assertTrue(providedBy(c).extends(I5))

        class COnly(A, B): __implemented__ = I31
        class D(COnly):
            pass
        classImplements(D, I5)

        classImplements(D, I5)
        c = D()
        directlyProvides(c, I4)
        self.assertEqual([i.getName() for i in providedBy(c)],
                         ['I4', 'I5', 'I31'])
        self.assertEqual([i.getName() for i in providedBy(c).flattened()],
                         ['I4', 'I5', 'I31', 'I3', 'Interface'])
        self.assertFalse(I1 in providedBy(c))
        self.assertFalse(I3 in providedBy(c))
        self.assertTrue(providedBy(c).extends(I3))
        self.assertFalse(providedBy(c).extends(I1))
        self.assertTrue(providedBy(c).extends(I31))
        self.assertTrue(providedBy(c).extends(I5))

def synchronous(func):
    """Decorator to ensure that `func` never runs in the reactor thread.

    If the wrapped function is called from the reactor thread, this will
    raise a :class:`AssertionError`, implying that this is a programming
    error. Calls from outside the reactor will proceed unaffected.

    There is an asymmetry with the `asynchronous` decorator. The reason
    is that it is essential to be aware when `deferToThread()` is being
    used, so that in-reactor code knows to synchronise with it, to add a
    callback to the :class:`Deferred` that it returns, for example. The
    expectation with `asynchronous` is that the return value is always
    important, and will be appropriate to the environment in which it is
    utilised.

    It is possible to programmatically determine if a function has been thusly
    decorated by checking if `ISynchronous` is provided::

      if ISynchronous.providedBy(a_function):
          ...  # a_function has been decorated with @synchronous

    This also serves a secondary documentation purpose; functions decorated
    with this are readily identifiable as synchronous, or blocking.

    :raises AssertionError: When called inside the reactor thread.
    """
    try:
        # A function or method; see PEP 3155.
        func_name = func.__qualname__
    except AttributeError:
        # An instance with a __call__ method.
        func_name = type(func).__qualname__

    @wraps(func)
    def wrapper(*args, **kwargs):
        # isInIOThread() can return True if the reactor has previously been
        # started but has now stopped, so don't test isInIOThread() until
        # we've also checked if the reactor is running.
        if reactor.running and isInIOThread():
            raise AssertionError(
                "Function %s(...) must not be called in the "
                "reactor thread." % func.__name__)
        else:
            result = func(*args, **kwargs)
            if isinstance(result, Deferred):
                args_reprs = chain(
                    map(repr, args), starmap(
                        "{}={!r}".format, kwargs.items()))
                raise TypeError(
                    "Synchronous call returned a Deferred: %s(%s)"
                    % (func_name, ", ".join(args_reprs)))
            else:
                return result

    # This makes it possible to reliably determine programmatically if a
    # function has been decorated with @synchronous.
    interface.directlyProvides(wrapper, ISynchronous)

    return wrapper