Python zope.interface 模块，provider() 实例源码

def __init__(self, logPath=None, timeout=_REQUEST_TIMEOUT,
                 logFormatter=None, reactor=None):
        """
        @param logFormatter: An object to format requests into log lines for
            the access log.
        @type logFormatter: L{IAccessLogFormatter} provider

        @param reactor: A L{IReactorTime} provider used to compute logging
            timestamps.
        """
        if not reactor:
            from twisted.internet import reactor
        self._reactor = reactor

        if logPath is not None:
            logPath = os.path.abspath(logPath)
        self.logPath = logPath
        self.timeOut = timeout
        if logFormatter is None:
            logFormatter = combinedLogFormatter
        self._logFormatter = logFormatter

        # For storing the cached log datetime and the callback to update it
        self._logDateTime = None
        self._logDateTimeCall = None

def __init__(self, connectedDeferred, wrappedProtocol):
        """
        @param connectedDeferred: The L{Deferred} that will callback
            with the C{wrappedProtocol} when it is connected.

        @param wrappedProtocol: An L{IProtocol} provider that will be
            connected.
        """
        self._connectedDeferred = connectedDeferred
        self._wrappedProtocol = wrappedProtocol

        for iface in [interfaces.IHalfCloseableProtocol,
                      interfaces.IFileDescriptorReceiver,
                      interfaces.IHandshakeListener]:
            if iface.providedBy(self._wrappedProtocol):
                directlyProvides(self, iface)

def connect(self, protocolFactory):
        """
        Implement L{IStreamClientEndpoint.connect} to launch a child process
        and connect it to a protocol created by C{protocolFactory}.

        @param protocolFactory: A factory for an L{IProtocol} provider which
            will be notified of all events related to the created process.
        """
        proto = protocolFactory.buildProtocol(_ProcessAddress())
        try:
            self._spawnProcess(
                _WrapIProtocol(proto, self._executable, self._errFlag),
                self._executable, self._args, self._env, self._path, self._uid,
                self._gid, self._usePTY, self._childFDs)
        except:
            return defer.fail()
        else:
            return defer.succeed(proto)

def __init__(self, reactor, port, backlog, interface):
        """
        @param reactor: An L{IReactorTCP} provider.

        @param port: The port number used for listening
        @type port: int

        @param backlog: Size of the listen queue
        @type backlog: int

        @param interface: The hostname to bind to
        @type interface: str
        """
        self._reactor = reactor
        self._port = port
        self._backlog = backlog
        self._interface = interface

def __init__(self, reactor, host, port, timeout=30, bindAddress=None):
        """
        @param reactor: An L{IReactorTCP} provider

        @param host: A hostname, used when connecting
        @type host: str

        @param port: The port number, used when connecting
        @type port: int

        @param timeout: The number of seconds to wait before assuming the
            connection has failed.
        @type timeout: int

        @param bindAddress: A (host, port) tuple of local address to bind to,
            or None.
        @type bindAddress: tuple
        """
        self._reactor = reactor
        self._host = host
        self._port = port
        self._timeout = timeout
        self._bindAddress = bindAddress

def __init__(self, reactor, port, sslContextFactory,
                 backlog=50, interface=''):
        """
        @param reactor: An L{IReactorSSL} provider.

        @param port: The port number used for listening
        @type port: int

        @param sslContextFactory: An instance of
            L{interfaces.IOpenSSLContextFactory}.

        @param backlog: Size of the listen queue
        @type backlog: int

        @param interface: The hostname to bind to, defaults to '' (all)
        @type interface: str
        """
        self._reactor = reactor
        self._port = port
        self._sslContextFactory = sslContextFactory
        self._backlog = backlog
        self._interface = interface

def __init__(self, reactor, path, timeout=30, checkPID=0):
        """
        @param reactor: An L{IReactorUNIX} provider.

        @param path: The path to the Unix socket file, used when connecting
        @type path: str

        @param timeout: Number of seconds to wait before assuming the
            connection has failed.
        @type timeout: int

        @param checkPID: If True, check for a pid file to verify that a server
            is listening.
        @type checkPID: bool
        """
        self._reactor = reactor
        self._path = path
        self._timeout = timeout
        self._checkPID = checkPID

def _parseServer(self, reactor, port, backlog=50, interface='::'):
        """
        Internal parser function for L{_parseServer} to convert the string
        arguments into structured arguments for the L{TCP6ServerEndpoint}

        @param reactor: An L{IReactorTCP} provider.

        @param port: The port number used for listening
        @type port: int

        @param backlog: Size of the listen queue
        @type backlog: int

        @param interface: The hostname to bind to
        @type interface: str
        """
        port = int(port)
        backlog = int(backlog)
        return TCP6ServerEndpoint(reactor, port, backlog, interface)

def spawnProcess(self, processProtocol, executable, args=(), env={},
                     path=None, uid=None, gid=None, usePTY=0, childFDs=None):
        """
        @ivar processProtocol: Stores the protocol passed to the reactor.
        @return: An L{IProcessTransport} provider.
        """
        self.processProtocol = processProtocol
        self.executable = executable
        self.args = args
        self.env = env
        self.path = path
        self.uid = uid
        self.gid = gid
        self.usePTY = usePTY
        self.childFDs = childFDs

        self.processTransport = MemoryProcessTransport()
        self.processProtocol.makeConnection(self.processTransport)
        return self.processTransport

def retries(timeout=30, intervals=1, clock=reactor):
    """Helper for retrying something, sleeping between attempts.

    Returns a generator that yields ``(elapsed, remaining, wait)`` tuples,
    giving times in seconds. The last item, `wait`, is the suggested amount of
    time to sleep before trying again.

    :param timeout: From now, how long to keep iterating, in seconds. This can
        be specified as a number, or as an iterable. In the latter case, the
        iterator is advanced each time an interval is needed. This allows for
        back-off strategies.
    :param intervals: The sleep between each iteration, in seconds, an an
        iterable from which to obtain intervals.
    :param clock: An optional `IReactorTime` provider. Defaults to the
        installed reactor.
    """
    start = clock.seconds()
    end = start + timeout

    if isinstance(intervals, Iterable):
        intervals = iter(intervals)
    else:
        intervals = repeat(intervals)

    return gen_retries(start, end, intervals, clock)

def getPeer(self):
        """
        Get the remote address of this connection.

        @return: An L{IAddress} provider.
        """
        return self.transport.getPeer()

def getHost(self):
        """
        Get the local address of this connection.

        @return: An L{IAddress} provider.
        """
        return self.transport.getHost()

def registerProducer(self, producer, streaming):
        """
        Register to receive data from a producer.

        This sets self to be a consumer for a producer.  When this object runs
        out of data (as when a send(2) call on a socket succeeds in moving the
        last data from a userspace buffer into a kernelspace buffer), it will
        ask the producer to resumeProducing().

        For L{IPullProducer} providers, C{resumeProducing} will be called once
        each time data is required.

        For L{IPushProducer} providers, C{pauseProducing} will be called
        whenever the write buffer fills up and C{resumeProducing} will only be
        called when it empties.

        @type producer: L{IProducer} provider
        @param producer: The L{IProducer} that will be producing data.

        @type streaming: L{bool}
        @param streaming: C{True} if C{producer} provides L{IPushProducer},
        C{False} if C{producer} provides L{IPullProducer}.

        @raise RuntimeError: If a producer is already registered.

        @return: L{None}
        """
        if self._requestProducer is not None:
            raise RuntimeError(
                "Cannot register producer %s, because producer %s was never "
                "unregistered." % (producer, self._requestProducer))

        if not streaming:
            producer = _PullToPush(producer, self)

        self._requestProducer = producer
        self._requestProducerStreaming = streaming

        if not streaming:
            producer.startStreaming()

def __init__(self, proto, executable, errFlag):
        """
        @param proto: An L{IProtocol} provider.
        @param errFlag: A constant belonging to L{StandardErrorBehavior}
            that determines if stderr is logged or dropped.
        @param executable: The file name (full path) to spawn.
        """
        self.protocol = proto
        self.errFlag = errFlag
        self.executable = executable

def makeConnection(self, process):
        """
        Call L{IProtocol} provider's makeConnection method with an
        L{ITransport} provider.

        @param process: An L{IProcessTransport} provider.
        """
        self.transport = _ProcessEndpointTransport(process)
        return self.protocol.makeConnection(self.transport)

def processEnded(self, reason):
        """
        If the process ends with L{error.ProcessDone}, this method calls the
        L{IProtocol} provider's L{connectionLost} with a
        L{error.ConnectionDone}

        @see: L{ProcessProtocol.processEnded}
        """
        if (reason.check(error.ProcessDone) == error.ProcessDone) and (
                reason.value.status == 0):
            return self.protocol.connectionLost(
                Failure(error.ConnectionDone()))
        else:
            return self.protocol.connectionLost(reason)

def __init__(self, reactor, port, backlog=50, interface='::'):
        """
        @param reactor: An L{IReactorTCP} provider.

        @param port: The port number used for listening
        @type port: int

        @param backlog: Size of the listen queue
        @type backlog: int

        @param interface: The hostname to bind to, defaults to C{::} (all)
        @type interface: str
        """
        _TCPServerEndpoint.__init__(self, reactor, port, backlog, interface)

def __init__(self, reactor, host, port, timeout=30, bindAddress=None,
                 attemptDelay=None):
        """
        Create a L{HostnameEndpoint}.

        @param reactor: The reactor to use for connections and delayed calls.
        @type reactor: provider of L{IReactorTCP}, L{IReactorTime} and either
            L{IReactorPluggableNameResolver} or L{IReactorPluggableResolver}.

        @param host: A hostname to connect to.
        @type host: L{bytes} or L{unicode}

        @param port: The port number to connect to.
        @type port: L{int}

        @param timeout: For each individual connection attempt, the number of
            seconds to wait before assuming the connection has failed.
        @type timeout: L{int}

        @param bindAddress: the local address of the network interface to make
            the connections from.
        @type bindAddress: L{bytes}

        @param attemptDelay: The number of seconds to delay between connection
            attempts.
        @type attemptDelay: L{float}

        @see: L{twisted.internet.interfaces.IReactorTCP.connectTCP}
        """
        self._reactor = reactor
        [self._badHostname, self._hostBytes, self._hostText] = (
            self._hostAsBytesAndText(host)
        )
        self._hostStr = self._hostBytes if bytes is str else self._hostText
        self._port = port
        self._timeout = timeout
        self._bindAddress = bindAddress
        if attemptDelay is None:
            attemptDelay = self._DEFAULT_ATTEMPT_DELAY
        self._attemptDelay = attemptDelay

def __init__(self, reactor, host, port, sslContextFactory,
                 timeout=30, bindAddress=None):
        """
        @param reactor: An L{IReactorSSL} provider.

        @param host: A hostname, used when connecting
        @type host: str

        @param port: The port number, used when connecting
        @type port: int

        @param sslContextFactory: SSL Configuration information as an instance
            of L{interfaces.IOpenSSLContextFactory}.

        @param timeout: Number of seconds to wait before assuming the
            connection has failed.
        @type timeout: int

        @param bindAddress: A (host, port) tuple of local address to bind to,
            or None.
        @type bindAddress: tuple
        """
        self._reactor = reactor
        self._host = host
        self._port = port
        self._sslContextFactory = sslContextFactory
        self._timeout = timeout
        self._bindAddress = bindAddress

def __init__(self, reactor, fileno, addressFamily):
        """
        @param reactor: An L{IReactorSocket} provider.

        @param fileno: An integer file descriptor corresponding to a listening
            I{SOCK_STREAM} socket.

        @param addressFamily: The address family of the socket given by
            C{fileno}.
        """
        self.reactor = reactor
        self.fileno = fileno
        self.addressFamily = addressFamily
        self._used = False

def _parseServer(self, reactor, domain, index):
        """
        Internal parser function for L{_parseServer} to convert the string
        arguments for a systemd server endpoint into structured arguments for
        L{AdoptedStreamServerEndpoint}.

        @param reactor: An L{IReactorSocket} provider.

        @param domain: The domain (or address family) of the socket inherited
            from systemd.  This is a string like C{"INET"} or C{"UNIX"}, ie the
            name of an address family from the L{socket} module, without the
            C{"AF_"} prefix.
        @type domain: C{str}

        @param index: An offset into the list of file descriptors inherited from
            systemd.
        @type index: C{str}

        @return: A two-tuple of parsed positional arguments and parsed keyword
            arguments (a tuple and a dictionary).  These can be used to
            construct an L{AdoptedStreamServerEndpoint}.
        """
        index = int(index)
        fileno = self._sddaemon.inheritedDescriptors()[index]
        addressFamily = getattr(socket, 'AF_' + domain)
        return AdoptedStreamServerEndpoint(reactor, fileno, addressFamily)

def _loadCAsFromDir(directoryPath):
    """
    Load certificate-authority certificate objects in a given directory.

    @param directoryPath: a L{unicode} or L{bytes} pointing at a directory to
        load .pem files from, or L{None}.

    @return: an L{IOpenSSLTrustRoot} provider.
    """
    caCerts = {}
    for child in directoryPath.children():
        if not child.asTextMode().basename().split(u'.')[-1].lower() == u'pem':
            continue
        try:
            data = child.getContent()
        except IOError:
            # Permission denied, corrupt disk, we don't care.
            continue
        try:
            theCert = Certificate.loadPEM(data)
        except SSLError:
            # Duplicate certificate, invalid certificate, etc.  We don't care.
            pass
        else:
            caCerts[theCert.digest()] = theCert
    return trustRootFromCertificates(caCerts.values())

def test_constructor(self):
        """
        Stores an L{IProtocol} provider and the flag to log/drop stderr
        """
        d = self.ep.connect(self.factory)
        self.successResultOf(d)
        wpp = self.reactor.processProtocol
        self.assertIsInstance(wpp.protocol, StubApplicationProtocol)
        self.assertEqual(wpp.errFlag, self.ep._errFlag)

def gen_retries(start, end, intervals, clock=reactor):
    """Helper for retrying something, sleeping between attempts.

    Yields ``(elapsed, remaining, wait)`` tuples, giving times in seconds. The
    last item, `wait`, is the suggested amount of time to sleep before trying
    again.

    This function works in concert with `retries`. It's split out so that
    `retries` can capture the correct start time rather than the time at which
    it is first iterated.

    :param start: The start time, in seconds, of this generator. This must be
        congruent with the `IReactorTime` argument passed to this generator.
    :param end: The desired end time, in seconds, of this generator. This must
        be congruent with the `IReactorTime` argument passed to this
        generator.
    :param intervals: A iterable of intervals, each in seconds, which should
        be used as hints for the `wait` value that's generated.
    :param clock: An optional `IReactorTime` provider. Defaults to the
        installed reactor.

    """
    for interval in intervals:
        now = clock.seconds()
        if now < end:
            wait = min(interval, end - now)
            yield now - start, end - now, wait
        else:
            yield now - start, end - now, 0
            break

def test_register_behavior(self):
        cur_count = len(
            configure.get_configurations('plone.server.tests', 'behavior'))

        from plone.server.interfaces import IFormFieldProvider
        from zope.interface import provider
        from zope import schema

        @provider(IFormFieldProvider)
        class IMyBehavior(Interface):
            foobar = schema.Text()

        configure.behavior(
            title="MyBehavior",
            provides=IMyBehavior,
            factory="plone.behavior.AnnotationStorage",
            for_="plone.server.interfaces.IResource"
        )()

        self.assertEqual(
            len(configure.get_configurations('plone.server.tests', 'behavior')),
            cur_count + 1)

        class IMyType(Interface):
            pass

        class MyType(Item):
            pass

        configure.register_configuration(MyType, dict(
            context=ISite,
            schema=IMyType,
            portal_type="MyType2",
            behaviors=[IMyBehavior]
        ), 'contenttype')

        # now test it...
        configure.load_configuration(
            self.layer.app.app.config, 'plone.server.tests', 'contenttype')
        self.layer.app.app.config.execute_actions()

        resp = self.layer.requester('GET', '/plone/plone/@types')
        response = json.loads(resp.text)
        type_ = [s for s in response if s['title'] == 'MyType2'][0]
        self.assertTrue('foobar' in type_['definitions']['IMyBehavior']['properties'])

def deterministicResolvingReactor(reactor, expectedAddresses=(),
                                  hostMap=None):
    """
    Create a reactor that will deterministically resolve all hostnames it is
    passed to the list of addresses given.

    @param reactor: An object that we wish to add an
        L{IReactorPluggableNameResolver} to.
    @type reactor: Any object with some formally-declared interfaces (i.e. one
        where C{list(providedBy(reactor))} is not empty); usually C{IReactor*}
        interfaces.

    @param expectedAddresses: (optional); the addresses expected to be returned
        for every address.  If these are strings, they should be IPv4 or IPv6
        literals, and they will be wrapped in L{IPv4Address} and L{IPv6Address}
        objects in the resolution result.
    @type expectedAddresses: iterable of C{object} or C{str}

    @param hostMap: (optional); the names (unicode) mapped to lists of
        addresses (str or L{IAddress}); in the same format as expectedAddress,
        which map the results for I{specific} hostnames to addresses.

    @return: A new reactor which provides all the interfaces previously
        provided by C{reactor} as well as L{IReactorPluggableNameResolver}.
        All name resolutions performed with its C{nameResolver} attribute will
        resolve reentrantly and synchronously with the given
        C{expectedAddresses}.  However, it is not a complete implementation as
        it does not have an C{installNameResolver} method.
    """
    if hostMap is None:
        hostMap = {}
    hostMap = hostMap.copy()
    @provider(IHostnameResolver)
    class SimpleNameResolver(object):
        @staticmethod
        def resolveHostName(resolutionReceiver, hostName, portNumber=0,
                            addressTypes=None, transportSemantics='TCP'):
            resolutionReceiver.resolutionBegan(None)
            for expectedAddress in hostMap.get(hostName, expectedAddresses):
                if isinstance(expectedAddress, str):
                    expectedAddress = ([IPv4Address, IPv6Address]
                                       [isIPv6Address(expectedAddress)]
                                       ('TCP', expectedAddress, portNumber))
                resolutionReceiver.addressResolved(expectedAddress)
            resolutionReceiver.resolutionComplete()


    @provider(IReactorPluggableNameResolver)
    class WithResolver(proxyForInterface(
            InterfaceClass('*', tuple(providedBy(reactor)))
    )):
        nameResolver = SimpleNameResolver()
    return WithResolver(reactor)