Python zope.interface 模块,providedBy() 实例源码

我们从Python开源项目中,提取了以下50个代码示例,用于说明如何使用zope.interface.providedBy()

项目:guillotina    作者:plone    | 项目源码 | 文件源码
def test_no_adapts(self):
        from zope.interface import Interface
        from zope.interface import implementer
        from zope.interface import providedBy
        from guillotina.component.globalregistry import get_global_components
        from guillotina.component._declaration import adapter
        class IFoo(Interface):
            pass
        @implementer(IFoo)
        class Foo(object):
            pass
        @adapter(IFoo)
        def _handler(context):
            assert 0, "DON'T GO HERE"
        self._callFUT(_handler)
        gsm = get_global_components()
        regs = list(gsm.registeredHandlers())
        self.assertEqual(len(regs), 1)
        hr = regs[0]
        self.assertEqual(list(hr.required), list(providedBy(Foo())))
        self.assertEqual(hr.name, '')
        self.assertTrue(hr.factory is _handler)
项目:guillotina    作者:plone    | 项目源码 | 文件源码
def test_w_factory_returning_spec(self):
        from zope.interface import Interface
        from zope.interface import implementer
        from zope.interface import providedBy
        from guillotina.component.interfaces import IFactory
        class IFoo(Interface):
            pass
        class IBar(Interface):
            pass
        @implementer(IFoo, IBar)
        class _Factory(object):
            def get_interfaces(self):
                return providedBy(self)
        _factory = _Factory()
        class Context(object):
            def __conform__(self, iface):
                return self
            def getUtilitiesFor(self, iface):
                if iface is IFactory:
                    return [('test', _factory)]
        self.assertEqual(list(self._callFUT(IFoo, context=Context())),
                         [('test', _factory)])
        self.assertEqual(list(self._callFUT(IBar, context=Context())),
                         [('test', _factory)])
项目:guillotina    作者:plone    | 项目源码 | 文件源码
def query_multi_adapter(objects, interface=Interface, name=_BLANK, default=None,
                        context=None, args=[], kwargs={}):
    try:
        registry = get_component_registry(context)
    except ComponentLookupError:
        # Oh blast, no site manager. This should *never* happen!
        return default

    factory = registry.adapters.lookup(map(providedBy, objects), interface, name)
    if factory is None:
        return default

    result = factory(*objects, *args, **kwargs)
    if result is None:
        return default

    return result
项目:guillotina    作者:plone    | 项目源码 | 文件源码
def __get__(self, inst, owner):
        name, check = self._info
        default_factory = inst.__dict__.get('defaultFactory')
        # If there is no default factory, simply return the default.
        if default_factory is None:
            return inst.__dict__[name]
        # Get the default value by calling the factory. Some factories might
        # require a context to produce a value.
        if IContextAwareDefaultFactory.providedBy(default_factory):
            value = default_factory(inst.context)
        else:
            value = default_factory()
        # Check that the created value is valid.
        if check is not None:
            check(inst, value)
        elif value != inst.missing_value:
            inst.validate(value)
        return value
项目:guillotina    作者:plone    | 项目源码 | 文件源码
def __eq__(self, other):
        # should be the same type
        if type(self) != type(other):
            return False

        # should have the same properties
        names = {}  # used as set of property names, ignoring values
        for interface in providedBy(self):
            names.update(get_fields(interface))

        # order will be different always, don't compare it
        if 'order' in names:
            del names['order']
        for name in names:
            if getattr(self, name) != getattr(other, name):
                return False
        return True
项目:guillotina    作者:plone    | 项目源码 | 文件源码
def from_unicode(self, str):
        """
        >>> b = Bool()
        >>> IFromUnicode.providedBy(b)
        True
        >>> b.from_unicode('True')
        True
        >>> b.from_unicode('')
        False
        >>> b.from_unicode('true')
        True
        >>> b.from_unicode('false') or b.from_unicode('False')
        False
        """
        v = str == 'True' or str == 'true'
        self.validate(v)
        return v
项目:AskTanmay-NLQA-System-    作者:tanmayb123    | 项目源码 | 文件源码
def test_classImplements_base_not_derived(self):
        from zope.interface import Interface
        from zope.interface import implementedBy
        from zope.interface import providedBy
        class IBase(Interface):
            def method():
                pass
        class IDerived(IBase):
            pass
        class Current():
            __implemented__ = IBase
            def method(self):
                pass
        current = Current()

        self.assertTrue(IBase.implementedBy(Current))
        self.assertFalse(IDerived.implementedBy(Current))
        self.assertTrue(IBase in implementedBy(Current))
        self.assertFalse(IDerived in implementedBy(Current))
        self.assertTrue(IBase in providedBy(current))
        self.assertFalse(IDerived in providedBy(current))
项目:websauna    作者:websauna    | 项目源码 | 文件源码
def render_panel(context, request, name='', *args, **kw):
    """Render a panel.

    TODO: Clean this up

    Renders the named panel, returning a `unicode` object that is the
    rendered HTML for the panel.  The panel is looked up using the current
    context (or the context given as keyword argument, to override the
    context in which the panel is called) and an optional given name
    (which defaults to an empty string).
    The panel is called passing in the current
    context, request and any additional parameters passed into the
    `render_panel` call.  In case a panel isn't found, `None` is returned.
    """

    if not "controls" in kw:
        kw["controls"] = True

    adapters = request.registry.adapters
    panel = adapters.lookup((providedBy(context),), IPanel, name=name)
    assert panel, "Could not find panel {}".format(name)
    return Markup(panel(context, request, *args, **kw))
项目:zenchmarks    作者:squeaky-pl    | 项目源码 | 文件源码
def requestAvatarId(self, credentials):
        """
        Part of the L{ICredentialsChecker} interface.  Called by a portal with
        some credentials to check if they'll authenticate a user.  We check the
        interfaces that the credentials provide against our list of acceptable
        checkers.  If one of them matches, we ask that checker to verify the
        credentials.  If they're valid, we call our L{_cbGoodAuthentication}
        method to continue.

        @param credentials: the credentials the L{Portal} wants us to verify
        """
        ifac = providedBy(credentials)
        for i in ifac:
            c = self.checkers.get(i)
            if c is not None:
                d = defer.maybeDeferred(c.requestAvatarId, credentials)
                return d.addCallback(self._cbGoodAuthentication,
                        credentials)
        return defer.fail(UnhandledCredentials("No checker for %s" % \
            ', '.join(map(reflect.qual, ifac))))
项目:zenchmarks    作者:squeaky-pl    | 项目源码 | 文件源码
def makeConnection(self, transport):
        """
        Connect this wrapper to the given transport and initialize the
        necessary L{OpenSSL.SSL.Connection} with a memory BIO.
        """
        self._tlsConnection = self.factory._createConnection(self)
        self._appSendBuffer = []

        # Add interfaces provided by the transport we are wrapping:
        for interface in providedBy(transport):
            directlyProvides(self, interface)

        # Intentionally skip ProtocolWrapper.makeConnection - it might call
        # wrappedProtocol.makeConnection, which we want to make conditional.
        Protocol.makeConnection(self, transport)
        self.factory.registerProtocol(self)
        if self._connectWrapped:
            # Now that the TLS layer is initialized, notify the application of
            # the connection.
            ProtocolWrapper.makeConnection(self, transport)

        # Now that we ourselves have a transport (initialized by the
        # ProtocolWrapper.makeConnection call above), kick off the TLS
        # handshake.
        self._checkHandshakeStatus()
项目:zenchmarks    作者:squeaky-pl    | 项目源码 | 文件源码
def _checkHandshakeStatus(self):
        """
        Ask OpenSSL to proceed with a handshake in progress.

        Initially, this just sends the ClientHello; after some bytes have been
        stuffed in to the C{Connection} object by C{dataReceived}, it will then
        respond to any C{Certificate} or C{KeyExchange} messages.
        """
        # The connection might already be aborted (eg. by a callback during
        # connection setup), so don't even bother trying to handshake in that
        # case.
        if self._aborted:
            return
        try:
            self._tlsConnection.do_handshake()
        except WantReadError:
            self._flushSendBIO()
        except Error:
            self._tlsShutdownFinished(Failure())
        else:
            self._handshakeDone = True
            if IHandshakeListener.providedBy(self.wrappedProtocol):
                self.wrappedProtocol.handshakeCompleted()
项目:zenchmarks    作者:squeaky-pl    | 项目源码 | 文件源码
def _applyProtocolNegotiation(self, connection):
        """
        Applies ALPN/NPN protocol neogitation to the connection, if the factory
        supports it.

        @param connection: The OpenSSL connection object to have ALPN/NPN added
            to it.
        @type connection: L{OpenSSL.SSL.Connection}

        @return: Nothing
        @rtype: L{None}
        """
        if IProtocolNegotiationFactory.providedBy(self.wrappedFactory):
            protocols = self.wrappedFactory.acceptableProtocols()
            context = connection.get_context()
            _setAcceptableProtocols(context, protocols)

        return
项目:zenchmarks    作者:squeaky-pl    | 项目源码 | 文件源码
def test_classImplements_base_not_derived(self):
        from zope.interface import Interface
        from zope.interface import implementedBy
        from zope.interface import providedBy
        class IBase(Interface):
            def method():
                pass
        class IDerived(IBase):
            pass
        class Current():
            __implemented__ = IBase
            def method(self):
                pass
        current = Current()

        self.assertTrue(IBase.implementedBy(Current))
        self.assertFalse(IDerived.implementedBy(Current))
        self.assertTrue(IBase in implementedBy(Current))
        self.assertFalse(IDerived in implementedBy(Current))
        self.assertTrue(IBase in providedBy(current))
        self.assertFalse(IDerived in providedBy(current))
项目:hostapd-mana    作者:adde88    | 项目源码 | 文件源码
def requestAvatarId(self, credentials):
        ifac = interface.providedBy(credentials)
        for i in ifac:
            c = self.checkers.get(i)
            if c is not None:
                return c.requestAvatarId(credentials).addCallback(
                    self._cbGoodAuthentication, credentials)
        return defer.fail(UnhandledCredentials("No checker for %s" % \
            ', '.join(map(reflect.qal, ifac))))
项目:hostapd-mana    作者:adde88    | 项目源码 | 文件源码
def _generateCacheEntry(provider):
    dropin = CachedDropin(provider.__name__,
                          provider.__doc__)
    for k, v in provider.__dict__.iteritems():
        plugin = IPlugin(v, None)
        if plugin is not None:
            cachedPlugin = CachedPlugin(dropin, k, v.__doc__, list(providedBy(plugin)))
    return dropin
项目:hostapd-mana    作者:adde88    | 项目源码 | 文件源码
def login(self, credentials, mind, *interfaces):
        """
        @param credentials: an implementor of
        twisted.cred.credentials.ICredentials

        @param mind: an object which implements a client-side interface for
        your particular realm.  In many cases, this may be None, so if the word
        'mind' confuses you, just ignore it.

        @param interfaces: list of interfaces for the perspective that the mind
        wishes to attach to.  Usually, this will be only one interface, for
        example IMailAccount.  For highly dynamic protocols, however, this may
        be a list like (IMailAccount, IUserChooser, IServiceInfo).  To expand:
        if we are speaking to the system over IMAP, any information that will
        be relayed to the user MUST be returned as an IMailAccount implementor;
        IMAP clients would not be able to understand anything else.  Any
        information about unusual status would have to be relayed as a single
        mail message in an otherwise-empty mailbox.  However, in a web-based
        mail system, or a PB-based client, the ``mind'' object inside the web
        server (implemented with a dynamic page-viewing mechanism such as
        woven) or on the user's client program may be intelligent enough to
        respond to several ``server''-side interfaces.

        @return: A deferred which will fire a tuple of (interface,
        avatarAspect, logout).  The interface will be one of the interfaces
        passed in the 'interfaces' argument.  The 'avatarAspect' will implement
        that interface.  The 'logout' object is a callable which will detach
        the mind from the avatar.  It must be called when the user has
        conceptually disconnected from the service.  Although in some cases
        this will not be in connectionLost (such as in a web-based session), it
        will always be at the end of a user's interactive session.
        """
        ifac = providedBy(credentials)
        for i in ifac:
            c = self.checkers.get(i)
            if c is not None:
                return maybeDeferred(c.requestAvatarId, credentials
                    ).addCallback(self.realm.requestAvatar, mind, *interfaces
                    )
        return defer.fail(failure.Failure(error.UnhandledCredentials(
            "No checker for %s" % ', '.join(map(reflect.qual, ifac)))))
项目:hostapd-mana    作者:adde88    | 项目源码 | 文件源码
def makeConnection(self, transport):
        directlyProvides(self, *providedBy(self) + providedBy(transport))
        Protocol.makeConnection(self, transport)

    # Transport relaying
项目:hostapd-mana    作者:adde88    | 项目源码 | 文件源码
def testUtil(self):
        self.assert_(IC in implementedBy(C))
        self.assert_(I1 in implementedBy(A))
        self.assert_(not I1 in implementedBy(C))
        self.assert_(I2 in implementedBy(B))
        self.assert_(not I2 in implementedBy(C))

        self.assert_(IC in providedBy(C()))
        self.assert_(I1 in providedBy(A()))
        self.assert_(not I1 in providedBy(C()))
        self.assert_(I2 in providedBy(B()))
        self.assert_(not I2 in providedBy(C()))
项目:hostapd-mana    作者:adde88    | 项目源码 | 文件源码
def testVerifyImplementation(self):
        from zope.interface.verify import verifyClass
        self.assert_(verifyClass(FooInterface, Foo))
        self.assert_(Interface.providedBy(I1))
项目:hostapd-mana    作者:adde88    | 项目源码 | 文件源码
def testIssue228(self):
        # Test for http://collector.zope.org/Zope3-dev/228
        class I(Interface):
            "xxx"
        class Bad:
            __providedBy__ = None
        # Old style classes don't have a '__class__' attribute
        self.failUnlessRaises(AttributeError, I.providedBy, Bad)
项目:hostapd-mana    作者:adde88    | 项目源码 | 文件源码
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))
项目:hostapd-mana    作者:adde88    | 项目源码 | 文件源码
def adapter_hook(self, provided, object, name=u'', default=None):
        required = providedBy(object)
        cache = self._getcache(provided, name)
        factory = cache.get(required, _not_in_mapping)
        if factory is _not_in_mapping:
            factory = self.lookup((required, ), provided, name)

        if factory is not None:
            result = factory(object)
            if result is not None:
                return result

        return default
项目:hostapd-mana    作者:adde88    | 项目源码 | 文件源码
def subscribers(self, objects, provided):
        subscriptions = self.subscriptions(map(providedBy, objects), provided)
        if provided is None:
            result = ()
            for subscription in subscriptions:
                subscription(*objects)
        else:
            result = []
            for subscription in subscriptions:
                subscriber = subscription(*objects)
                if subscriber is not None:
                    result.append(subscriber)
        return result
项目:lagendacommun    作者:ecreall    | 项目源码 | 文件源码
def get_obj_value(obj, fields={}):
    interfaces = [i for i in providedBy(obj).interfaces()]
    if not interfaces:
        return {}

    result = {}
    objects_to_add = []
    if fields:
        for interface in interfaces:
            attributes = [a for a in interface if a in fields and
                          isinstance(interface[a], Attribute)]
            for attr in attributes:
                result[attr], to_add = get_attr_value(
                    interface[attr], obj, fields.get(attr, {}))
                objects_to_add.extend(to_add)

    else:
        for interface in interfaces:
            attributes = [a for a in interface
                          if isinstance(interface[a], Attribute)]
            for attr in attributes:
                result[attr], to_add = get_attr_value(interface[attr], obj)
                objects_to_add.extend(to_add)

    result = normalize_value(result)
    result['@id'] = '_:' + str(get_oid(obj, 'None'))
    result['@type'] = getattr(
        obj, 'type_title', obj.__class__.__name__)
    return result, objects_to_add
项目:plone.server    作者:plone    | 项目源码 | 文件源码
def asubscribers(self, objects, provided):
    subscriptions = self.subscriptions(map(providedBy, objects), provided)
    if provided is None:
        result = ()
        for subscription in subscriptions:
            if asyncio.iscoroutinefunction(subscription):
                await subscription(*objects)
    else:
        result = []
        for subscription in subscriptions:
            if asyncio.iscoroutinefunction(subscription):
                subscriber = await subscription(*objects)
            if subscriber is not None:
                result.append(subscriber)
    return result
项目:plone.server    作者:plone    | 项目源码 | 文件源码
def subscribers(self, objects, provided):
    subscriptions = self.subscriptions(map(providedBy, objects), provided)
    if provided is None:
        result = ()
        for subscription in subscriptions:
            if not asyncio.iscoroutinefunction(subscription):
                subscription(*objects)
    else:
        result = []
        for subscription in subscriptions:
            if not asyncio.iscoroutinefunction(subscription):
                subscriber = subscription(*objects)
            if subscriber is not None:
                result.append(subscriber)
    return result
项目:guillotina    作者:plone    | 项目源码 | 文件源码
def test_gsm_is_IComponentLookup(self):
        from guillotina.component.globalregistry import base
        from guillotina.component.interfaces import IComponentLookup
        gsm = self._callFUT()
        self.assertTrue(gsm is base)
        self.assertTrue(IComponentLookup.providedBy(gsm))
项目:guillotina    作者:plone    | 项目源码 | 文件源码
def test_sm_is_IComponentLookup(self):
        from guillotina.component.interfaces import IComponentLookup
        sm = self._callFUT()
        self.assertTrue(IComponentLookup.providedBy(sm))
项目:guillotina    作者:plone    | 项目源码 | 文件源码
def asubscribers(self, objects, provided):
    subscriptions = self.subscriptions(map(providedBy, objects), provided)
    results = []
    for subscription in subscriptions:
        if asyncio.iscoroutinefunction(subscription):
            results.append(await subscription(*objects))
    return results
项目:guillotina    作者:plone    | 项目源码 | 文件源码
def subscribers(self, objects, provided):
    subscriptions = self.subscriptions(map(providedBy, objects), provided)
    result = []
    for subscription in subscriptions:
        if not asyncio.iscoroutinefunction(subscription):
            result.append(subscription(*objects))
    return result
项目:guillotina    作者:plone    | 项目源码 | 文件源码
def utility(_context, provides=None, component=None, factory=None, name=''):
    if factory and component:
        raise TypeError("Can't specify factory and component.")

    if provides is None:
        if factory:
            provides = list(implementedBy(factory))
        else:
            provides = list(providedBy(component))
        if len(provides) == 1:
            provides = provides[0]
        else:
            raise TypeError("Missing 'provides' attribute")

    if name == '':
        if factory:
            name = getName(factory)
        else:
            name = getName(component)

    _context.action(
        discriminator=('utility', provides, name),
        callable=handler,
        args=('registerUtility', component, provides, name),
        kw=dict(factory=factory))
    _context.action(
        discriminator=None,
        callable=provide_interface,
        args=('', provides))
项目:guillotina    作者:plone    | 项目源码 | 文件源码
def __provides__(self):
        return providedBy(self.field) + implementedBy(FieldReadAccessor)
项目:guillotina    作者:plone    | 项目源码 | 文件源码
def test___provides___w_field_no_provides(self):
        from zope.interface import implementedBy
        from zope.interface import providedBy
        wrapped = self._makeOne(object())
        self.assertEqual(list(providedBy(wrapped)),
                         list(implementedBy(self._getTargetClass())))
项目:sslstrip-hsts-openwrt    作者:adde88    | 项目源码 | 文件源码
def requestAvatarId(self, credentials):
        ifac = interface.providedBy(credentials)
        for i in ifac:
            c = self.checkers.get(i)
            if c is not None:
                return c.requestAvatarId(credentials).addCallback(
                    self._cbGoodAuthentication, credentials)
        return defer.fail(UnhandledCredentials("No checker for %s" % \
            ', '.join(map(reflect.qal, ifac))))
项目:sslstrip-hsts-openwrt    作者:adde88    | 项目源码 | 文件源码
def _generateCacheEntry(provider):
    dropin = CachedDropin(provider.__name__,
                          provider.__doc__)
    for k, v in provider.__dict__.iteritems():
        plugin = IPlugin(v, None)
        if plugin is not None:
            cachedPlugin = CachedPlugin(dropin, k, v.__doc__, list(providedBy(plugin)))
    return dropin
项目:sslstrip-hsts-openwrt    作者:adde88    | 项目源码 | 文件源码
def login(self, credentials, mind, *interfaces):
        """
        @param credentials: an implementor of
        twisted.cred.credentials.ICredentials

        @param mind: an object which implements a client-side interface for
        your particular realm.  In many cases, this may be None, so if the word
        'mind' confuses you, just ignore it.

        @param interfaces: list of interfaces for the perspective that the mind
        wishes to attach to.  Usually, this will be only one interface, for
        example IMailAccount.  For highly dynamic protocols, however, this may
        be a list like (IMailAccount, IUserChooser, IServiceInfo).  To expand:
        if we are speaking to the system over IMAP, any information that will
        be relayed to the user MUST be returned as an IMailAccount implementor;
        IMAP clients would not be able to understand anything else.  Any
        information about unusual status would have to be relayed as a single
        mail message in an otherwise-empty mailbox.  However, in a web-based
        mail system, or a PB-based client, the ``mind'' object inside the web
        server (implemented with a dynamic page-viewing mechanism such as
        woven) or on the user's client program may be intelligent enough to
        respond to several ``server''-side interfaces.

        @return: A deferred which will fire a tuple of (interface,
        avatarAspect, logout).  The interface will be one of the interfaces
        passed in the 'interfaces' argument.  The 'avatarAspect' will implement
        that interface.  The 'logout' object is a callable which will detach
        the mind from the avatar.  It must be called when the user has
        conceptually disconnected from the service.  Although in some cases
        this will not be in connectionLost (such as in a web-based session), it
        will always be at the end of a user's interactive session.
        """
        ifac = providedBy(credentials)
        for i in ifac:
            c = self.checkers.get(i)
            if c is not None:
                return maybeDeferred(c.requestAvatarId, credentials
                    ).addCallback(self.realm.requestAvatar, mind, *interfaces
                    )
        return defer.fail(failure.Failure(error.UnhandledCredentials(
            "No checker for %s" % ', '.join(map(reflect.qual, ifac)))))
项目:sslstrip-hsts-openwrt    作者:adde88    | 项目源码 | 文件源码
def makeConnection(self, transport):
        directlyProvides(self, *providedBy(self) + providedBy(transport))
        Protocol.makeConnection(self, transport)

    # Transport relaying
项目:sslstrip-hsts-openwrt    作者:adde88    | 项目源码 | 文件源码
def testUtil(self):
        self.assert_(IC in implementedBy(C))
        self.assert_(I1 in implementedBy(A))
        self.assert_(not I1 in implementedBy(C))
        self.assert_(I2 in implementedBy(B))
        self.assert_(not I2 in implementedBy(C))

        self.assert_(IC in providedBy(C()))
        self.assert_(I1 in providedBy(A()))
        self.assert_(not I1 in providedBy(C()))
        self.assert_(I2 in providedBy(B()))
        self.assert_(not I2 in providedBy(C()))
项目:sslstrip-hsts-openwrt    作者:adde88    | 项目源码 | 文件源码
def testVerifyImplementation(self):
        from zope.interface.verify import verifyClass
        self.assert_(verifyClass(FooInterface, Foo))
        self.assert_(Interface.providedBy(I1))
项目:sslstrip-hsts-openwrt    作者:adde88    | 项目源码 | 文件源码
def testIssue228(self):
        # Test for http://collector.zope.org/Zope3-dev/228
        class I(Interface):
            "xxx"
        class Bad:
            __providedBy__ = None
        # Old style classes don't have a '__class__' attribute
        self.failUnlessRaises(AttributeError, I.providedBy, Bad)
项目:sslstrip-hsts-openwrt    作者:adde88    | 项目源码 | 文件源码
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))
项目:sslstrip-hsts-openwrt    作者:adde88    | 项目源码 | 文件源码
def adapter_hook(self, provided, object, name=u'', default=None):
        required = providedBy(object)
        cache = self._getcache(provided, name)
        factory = cache.get(required, _not_in_mapping)
        if factory is _not_in_mapping:
            factory = self.lookup((required, ), provided, name)

        if factory is not None:
            result = factory(object)
            if result is not None:
                return result

        return default
项目:sslstrip-hsts-openwrt    作者:adde88    | 项目源码 | 文件源码
def subscribers(self, objects, provided):
        subscriptions = self.subscriptions(map(providedBy, objects), provided)
        if provided is None:
            result = ()
            for subscription in subscriptions:
                subscription(*objects)
        else:
            result = []
            for subscription in subscriptions:
                subscriber = subscription(*objects)
                if subscriber is not None:
                    result.append(subscriber)
        return result
项目:txjuju    作者:juju    | 项目源码 | 文件源码
def makeConnection(self, transport):
        """
        Upon connection, provides the transport interface, and forwards ourself
        as the transport to C{self.wrappedProtocol}.

        @type transport: L{twisted.internet.interfaces.ITransport} provider.
        @param transport: The transport to use for the protocol.
        """
        directlyProvides(self, providedBy(transport))
        WebSocketsProtocol.makeConnection(self, transport)
        self.wrappedProtocol.makeConnection(self)
项目:AskTanmay-NLQA-System-    作者:tanmayb123    | 项目源码 | 文件源码
def test_providedBy_miss(self):
        from zope.interface import interface
        from zope.interface.declarations import _empty
        sb = self._makeOne()
        def _providedBy(obj):
            return _empty
        with _Monkey(interface, providedBy=_providedBy):
            self.assertFalse(sb.providedBy(object()))
项目:AskTanmay-NLQA-System-    作者:tanmayb123    | 项目源码 | 文件源码
def test_providedBy_hit(self):
        from zope.interface import interface
        sb = self._makeOne()
        class _Decl(object):
            _implied = {sb: {},}
        def _providedBy(obj):
            return _Decl()
        with _Monkey(interface, providedBy=_providedBy):
            self.assertTrue(sb.providedBy(object()))
项目:AskTanmay-NLQA-System-    作者:tanmayb123    | 项目源码 | 文件源码
def _makeOne(self, object_should_provide):
        class IB(self._getTargetClass()):
            def _call_conform(self, conform):
                return conform(self)
            def providedBy(self, obj):
                return object_should_provide
        return IB()
项目:AskTanmay-NLQA-System-    作者:tanmayb123    | 项目源码 | 文件源码
def test_classImplements_simple(self):
        from zope.interface import Interface
        from zope.interface import implementedBy
        from zope.interface import providedBy

        class ICurrent(Interface):
            def method1(a, b):
                pass
            def method2(a, b):
                pass

        class IOther(Interface):
            pass

        class Current(object):
            __implemented__ = ICurrent
            def method1(self, a, b):
                return 1
            def method2(self, a, b):
                return 2

        current = Current()

        self.assertTrue(ICurrent.implementedBy(Current))
        self.assertFalse(IOther.implementedBy(Current))
        self.assertTrue(ICurrent in implementedBy(Current))
        self.assertFalse(IOther in implementedBy(Current))
        self.assertTrue(ICurrent in providedBy(current))
        self.assertFalse(IOther in providedBy(current))
项目:AskTanmay-NLQA-System-    作者:tanmayb123    | 项目源码 | 文件源码
def test_classImplements_multiple(self):
        from zope.interface import Interface
        from zope.interface import implementedBy
        from zope.interface import providedBy

        class ILeft(Interface):
            def method():
                pass

        class IRight(ILeft):
            pass

        class Left(object):
            __implemented__ = ILeft

            def method(self):
                pass

        class Right(object):
            __implemented__ = IRight

        class Ambi(Left, Right):
            pass

        ambi = Ambi()

        self.assertTrue(ILeft.implementedBy(Ambi))
        self.assertTrue(IRight.implementedBy(Ambi))
        self.assertTrue(ILeft in implementedBy(Ambi))
        self.assertTrue(IRight in implementedBy(Ambi))
        self.assertTrue(ILeft in providedBy(ambi))
        self.assertTrue(IRight in providedBy(ambi))
项目:AskTanmay-NLQA-System-    作者:tanmayb123    | 项目源码 | 文件源码
def test_interface_object_provides_Interface(self):
        from zope.interface import Interface

        class AnInterface(Interface):
            pass

        self.assertTrue(Interface.providedBy(AnInterface))