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

def test_get_interfaces_explicit(self):
        from zope.interface import Interface
        from zope.interface import implementer
        class IFoo(Interface):
            pass
        class IBar(Interface):
            pass
        class IBaz(Interface):
            pass
        @implementer(IBaz)
        def _callable():
            pass
        factory = self._makeOne(_callable, interfaces=(IFoo, IBar))
        spec = factory.get_interfaces()
        self.assertEqual(spec.__name__, '_callable')
        self.assertEqual(list(spec), [IFoo, IBar])

def test_anonymous_no_provides_no_adapts(self):
        from zope.interface import Interface
        from zope.interface import implementer
        from guillotina.component.globalregistry import get_global_components
        from guillotina.component._declaration import adapter
        class IFoo(Interface):
            pass
        class IBar(Interface):
            pass
        @implementer(IFoo)
        class Foo(object):
            pass
        @adapter(IFoo)
        @implementer(IBar)
        class Bar(object):
            def __init__(self, context):
                self.context = context
        self._callFUT(Bar)
        gsm = get_global_components()
        foo = Foo()
        adapted = gsm.getAdapter(foo, IBar)
        self.assertTrue(isinstance(adapted, Bar))
        self.assertTrue(adapted.context is foo)

def test_named_w_provides_w_adapts(self):
        from zope.interface import Interface
        from zope.interface import implementer
        from guillotina.component.globalregistry import get_global_components
        class IFoo(Interface):
            pass
        class IBar(Interface):
            pass
        @implementer(IFoo)
        class Foo(object):
            pass
        class Bar(object):
            def __init__(self, context):
                self.context = context
        self._callFUT(Bar, (IFoo,), IBar, 'test')
        gsm = get_global_components()
        foo = Foo()
        adapted = gsm.getAdapter(foo, IBar, name='test')
        self.assertTrue(isinstance(adapted, Bar))
        self.assertTrue(adapted.context is foo)

def test_w_provides_w_adapts(self):
        from zope.interface import Interface
        from zope.interface import implementer
        from guillotina.component.globalregistry import get_global_components
        class IFoo(Interface):
            pass
        class IBar(Interface):
            pass
        @implementer(IFoo)
        class Foo(object):
            pass
        class Bar(object):
            def __init__(self, context):
                self.context = context
        self._callFUT(Bar, (IFoo,), IBar)
        gsm = get_global_components()
        foo = Foo()
        adapted = gsm.subscribers((foo,), IBar)
        self.assertEqual(len(adapted), 1)
        self.assertTrue(isinstance(adapted[0], Bar))
        self.assertTrue(adapted[0].context is foo)

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)

def test_hit(self):
        from zope.interface import Interface
        from zope.interface import implementer
        from guillotina.component import get_global_components
        class IFoo(Interface):
            pass
        class IBar(Interface):
            pass
        @implementer(IBar)
        class Bar(object):
            pass
        @implementer(IFoo)
        class Baz(object):
            def __init__(self, context):
                self.context = context
        get_global_components().registerAdapter(Baz, (IBar,), IFoo, '')
        bar = Bar()
        adapted = IFoo(bar)
        self.assertTrue(adapted.__class__ is Baz)
        self.assertTrue(adapted.context is bar)

def test_anonymous_hit(self):
        from zope.interface import Interface
        from zope.interface import implementer
        from guillotina.component import get_global_components
        class IFoo(Interface):
            pass
        class IBar(Interface):
            pass
        @implementer(IBar)
        class Bar(object):
            pass
        @implementer(IFoo)
        class Baz(object):
            def __init__(self, context):
                self.context = context
        get_global_components().registerAdapter(Baz, (IBar,), IFoo, '')
        bar = Bar()
        adapted = self._callFUT(bar, IFoo, '')
        self.assertTrue(adapted.__class__ is Baz)
        self.assertTrue(adapted.context is bar)

def test_named_hit(self):
        from zope.interface import Interface
        from zope.interface import implementer
        from guillotina.component import get_global_components
        class IFoo(Interface):
            pass
        class IBar(Interface):
            pass
        @implementer(IBar)
        class Bar(object):
            pass
        @implementer(IFoo)
        class Baz(object):
            def __init__(self, context):
                self.context = context
        get_global_components().registerAdapter(Baz, (IBar,), IFoo, 'named')
        bar = Bar()
        adapted = self._callFUT(bar, IFoo, 'named')
        self.assertTrue(adapted.__class__ is Baz)
        self.assertTrue(adapted.context is bar)

def test_anonymous_hit(self):
        from zope.interface import Interface
        from zope.interface import implementer
        from guillotina.component import get_global_components
        class IFoo(Interface):
            pass
        class IBar(Interface):
            pass
        @implementer(IBar)
        class Bar(object):
            pass
        @implementer(IFoo)
        class Baz(object):
            def __init__(self, context):
                self.context = context
        get_global_components().registerAdapter(Baz, (IBar,), IFoo, '')
        bar = Bar()
        adapted = self._callFUT(bar, IFoo, '')
        self.assertTrue(adapted.__class__ is Baz)
        self.assertTrue(adapted.context is bar)

def test_named_hit(self):
        from zope.interface import Interface
        from zope.interface import implementer
        from guillotina.component import get_global_components
        class IFoo(Interface):
            pass
        class IBar(Interface):
            pass
        @implementer(IBar)
        class Bar(object):
            pass
        @implementer(IFoo)
        class Baz(object):
            def __init__(self, context):
                self.context = context
        get_global_components().registerAdapter(Baz, (IBar,), IFoo, 'named')
        bar = Bar()
        adapted = self._callFUT(bar, IFoo, 'named')
        self.assertTrue(adapted.__class__ is Baz)
        self.assertTrue(adapted.context is bar)

def test_wo_sitemanager(self):
        from zope.interface import Interface
        from zope.interface import implementer
        from guillotina.component.interfaces import ComponentLookupError
        class IFoo(Interface):
            pass
        class IBar(Interface):
            pass
        class IBaz(Interface):
            pass
        @implementer(IBar)
        class Bar(object):
            pass
        @implementer(IBaz)
        class Baz(object):
            pass
        class Context(object):
            def __conform__(self, iface):
                raise ComponentLookupError
        bar = Bar()
        baz = Baz()
        adapted = self._callFUT((bar, baz), IFoo, '', context=Context())
        self.assertTrue(adapted is None)

def test_hit(self):
        from guillotina.component import get_global_components
        from zope.interface import Interface
        from zope.interface import implementer
        class IFoo(Interface):
            pass
        @implementer(IFoo)
        class Foo(object):
            pass
        _called = []
        def _bar(context):
                _called.append('_bar')
        def _baz(context):
                _called.append('_baz')
        gsm = get_global_components()
        gsm.registerHandler(_bar, (IFoo,))
        gsm.registerHandler(_baz, (IFoo,))
        self._callFUT(Foo())
        self.assertEqual(len(_called), 2, _called)
        self.assertTrue('_bar' in _called)
        self.assertTrue('_baz' in _called)

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)])

def _makeMyUtility(name, sm):
    global IMyUtility
    from zope.interface import Interface
    from zope.interface import implementer
    from guillotina.component.tests.examples import ConformsToIComponentLookup

    if IMyUtility is None:
        class IMyUtility(Interface):
            pass

    @implementer(IMyUtility)
    class MyUtility(ConformsToIComponentLookup):
        def __init__(self, id, sm):
            self.id = id
            self.sitemanager = sm

    return MyUtility(name, sm)

def test_no_name(self):
        from zope.interface import Interface

        class IFoo(Interface):
            pass

        class IBar(Interface):
            pass

        from guillotina.component import adapter
        from zope.interface import implementer, named

        @adapter(IFoo)
        @implementer(IBar)
        @named('bar')
        class _Factory(object):
            def __init__(self, context):
                self.context = context
        _cfg_ctx = _makeConfigContext()
        self._callFUT(_cfg_ctx, [_Factory])
        # Register the adapter
        action = _cfg_ctx._actions[0][1]
        self.assertEqual(action['args'][4], 'bar')

def test_w_component_wo_provides_wo_name(self):
        from zope.interface import Interface, implementer, named

        class IFoo(Interface):
            pass

        @implementer(IFoo)
        @named('foo')
        class Foo(object):
            pass
        foo = Foo()
        _cfg_ctx = _makeConfigContext()
        self._callFUT(_cfg_ctx, component=foo)
        action = _cfg_ctx._actions[0][1]
        self.assertEqual(action['args'][1], foo)
        self.assertEqual(action['args'][2], IFoo)
        self.assertEqual(action['args'][3], 'foo')

def test_bind_w_voc_is_ICSB(self):
        from zope.interface import implementer
        from guillotina.schema.interfaces import IContextSourceBinder
        from guillotina.schema.interfaces import ISource

        @implementer(IContextSourceBinder)
        @implementer(ISource)
        class Vocab(object):
            def __init__(self, context):
                self.context = context

            def __call__(self, context):
                return self.__class__(context)

        # Chicken-egg
        source = self._makeOne(vocabulary='temp')
        source.vocabulary = Vocab(source)
        source.vocabularyName = None
        instance = DummyInstance()
        target = source.bind(instance)
        self.assertEqual(target.vocabulary.context, instance)

def test__validate_source_is_ICSB_bound(self):
        from zope.interface import implementer
        from guillotina.schema.interfaces import IContextSourceBinder
        from guillotina.schema.exceptions import ConstraintNotSatisfied
        from guillotina.schema.tests.test_vocabulary import _makeSampleVocabulary

        @implementer(IContextSourceBinder)
        class SampleContextSourceBinder(object):
            def __call__(self, context):
                return _makeSampleVocabulary()

        s = SampleContextSourceBinder()
        choice = self._makeOne(source=s)
        # raises not iterable with unbound field
        self.assertRaises(TypeError, choice.validate, 1)
        o = object()
        clone = choice.bind(o)
        clone._validate(1)
        clone._validate(3)
        self.assertRaises(ConstraintNotSatisfied, clone._validate, 42)

def test__validate_w_value_providing_schema_but_invalid_fields(self):
        from zope.interface import implementer
        from guillotina.schema.exceptions import WrongContainedType
        from guillotina.schema.exceptions import RequiredMissing
        from guillotina.schema.exceptions import WrongType
        from guillotina.schema._bootstrapfields import Text
        schema = self._makeSchema(foo=Text(), bar=Text())

        @implementer(schema)
        class Broken(object):
            foo = None
            bar = 1

        objf = self._makeOne(schema)
        self.assertRaises(WrongContainedType, objf.validate, Broken())
        errors = self._getErrors(objf.validate, Broken())
        self.assertEqual(len(errors), 2)
        errors = sorted(errors, key=lambda x: type(x).__name__)
        err = errors[0]
        self.assertTrue(isinstance(err, RequiredMissing))
        self.assertEqual(err.args, ('foo',))
        err = errors[1]
        self.assertTrue(isinstance(err, WrongType))
        self.assertEqual(err.args, (1, str, 'bar'))

def test_class_has_required_method_derived(self):
        from zope.interface import Interface
        from zope.interface import implementer

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

        class IDerived(IBase):
            pass

        @implementer(IDerived)
        class Current(object):

            def method(self):
                pass

        self._callFUT(IDerived, Current)

def test_method_takes_wrong_arg_names_but_OK(self):
        # We no longer require names to match.
        from zope.interface import Interface
        from zope.interface import implementer

        class ICurrent(Interface):

            def method(a):
                pass

        @implementer(ICurrent)
        class Current(object):

            def method(self, b):
                pass

        self._callFUT(ICurrent, Current)

def test_method_takes_not_enough_args(self):
        from zope.interface import Interface
        from zope.interface import implementer
        from zope.interface.exceptions import BrokenMethodImplementation

        class ICurrent(Interface):

            def method(a):
                pass

        @implementer(ICurrent)
        class Current(object):

            def method(self):
                pass

        self.assertRaises(BrokenMethodImplementation,
                          self._callFUT, ICurrent, Current)

def test_method_doesnt_take_required_starargs(self):
        from zope.interface import Interface
        from zope.interface import implementer
        from zope.interface.exceptions import BrokenMethodImplementation

        class ICurrent(Interface):

            def method(*args):
                pass

        @implementer(ICurrent)
        class Current(object):

            def method(self):
                pass

        self.assertRaises(BrokenMethodImplementation,
                          self._callFUT, ICurrent, Current)

def test_method_doesnt_take_required_only_kwargs(self):
        from zope.interface import Interface
        from zope.interface import implementer
        from zope.interface.exceptions import BrokenMethodImplementation

        class ICurrent(Interface):

            def method(**kw):
                pass

        @implementer(ICurrent)
        class Current(object):

            def method(self):
                pass

        self.assertRaises(BrokenMethodImplementation,
                          self._callFUT, ICurrent, Current)

def test_method_takes_extra_arg_with_default(self):
        from zope.interface import Interface
        from zope.interface import implementer

        class ICurrent(Interface):

            def method(a):
                pass

        @implementer(ICurrent)
        class Current(object):

            def method(self, a, b=None):
                pass

        self._callFUT(ICurrent, Current)

def test_method_takes_only_positional_args(self):
        from zope.interface import Interface
        from zope.interface import implementer

        class ICurrent(Interface):

            def method(a):
                pass

        @implementer(ICurrent)
        class Current(object):

            def method(self, *args):
                pass

        self._callFUT(ICurrent, Current)

def test_method_takes_only_kwargs(self):
        from zope.interface import Interface
        from zope.interface import implementer
        from zope.interface.exceptions import BrokenMethodImplementation

        class ICurrent(Interface):

            def method(a):
                pass

        @implementer(ICurrent)
        class Current(object):

            def method(self, **kw):
                pass

        self.assertRaises(BrokenMethodImplementation,
                          self._callFUT, ICurrent, Current)

def test_method_takes_extra_starargs(self):
        from zope.interface import Interface
        from zope.interface import implementer

        class ICurrent(Interface):

            def method(a):
                pass

        @implementer(ICurrent)
        class Current(object):

            def method(self, a, *args):
                pass

        self._callFUT(ICurrent, Current)

def test_method_takes_extra_starargs_and_kwargs(self):
        from zope.interface import Interface
        from zope.interface import implementer

        class ICurrent(Interface):

            def method(a):
                pass

        @implementer(ICurrent)
        class Current(object):

            def method(self, a, *args, **kw):
                pass

        self._callFUT(ICurrent, Current)

def test_method_takes_required_positional_and_starargs(self):
        from zope.interface import Interface
        from zope.interface import implementer

        class ICurrent(Interface):

            def method(a, *args):
                pass

        @implementer(ICurrent)
        class Current(object):

            def method(self, a, *args):
                pass

        self._callFUT(ICurrent, Current)

def test_method_takes_only_starargs(self):
        from zope.interface import Interface
        from zope.interface import implementer

        class ICurrent(Interface):

            def method(a, *args):
                pass

        @implementer(ICurrent)
        class Current(object):

            def method(self, *args):
                pass

        self._callFUT(ICurrent, Current)

def test_method_takes_required_kwargs(self):
        from zope.interface import Interface
        from zope.interface import implementer

        class ICurrent(Interface):

            def method(**kwargs):
                pass

        @implementer(ICurrent)
        class Current(object):

            def method(self, **kw):
                pass

        self._callFUT(ICurrent, Current)

def test_method_takes_positional_plus_required_starargs(self):
        from zope.interface import Interface
        from zope.interface import implementer
        from zope.interface.exceptions import BrokenMethodImplementation

        class ICurrent(Interface):

            def method(*args):
                pass

        @implementer(ICurrent)
        class Current(object):

            def method(self, a, *args):
                pass

        self.assertRaises(BrokenMethodImplementation,
                          self._callFUT, ICurrent, Current)

def test_method_doesnt_take_required_kwargs(self):
        from zope.interface import Interface
        from zope.interface import implementer
        from zope.interface.exceptions import BrokenMethodImplementation

        class ICurrent(Interface):

            def method(**kwargs):
                pass

        @implementer(ICurrent)
        class Current(object):

            def method(self, a):
                pass

        self.assertRaises(BrokenMethodImplementation,
                          self._callFUT, ICurrent, Current)

def test_w_callable_non_func_method(self):
        from zope.interface.interface import Method
        from zope.interface import Interface
        from zope.interface import implementer

        class QuasiMethod(Method):
            def __call__(self, *args, **kw):
                pass

        class QuasiCallable(object):
            def __call__(self, *args, **kw):
                pass

        class ICurrent(Interface):
            attr = QuasiMethod('This is callable')

        @implementer(ICurrent)
        class Current:
            attr = QuasiCallable()

        self._callFUT(ICurrent, Current)

def test_w_decorated_method(self):
        from zope.interface import Interface
        from zope.interface import implementer

        def decorator(func):
            # this is, in fact, zope.proxy.non_overridable
            return property(lambda self: func.__get__(self))

        class ICurrent(Interface):

            def method(a):
                pass

        @implementer(ICurrent)
        class Current(object):

            @decorator
            def method(self, a):
                pass

        self._callFUT(ICurrent, Current)

def test_classImplements(self):

        @implementer(I3)
        class A(Odd):
            pass

        @implementer(I4)
        class B(Odd):
            pass

        class C(A, B):
            pass
        classImplements(C, I1, I2)
        self.assertEqual([i.getName() for i in implementedBy(C)],
                         ['I1', 'I2', 'I3', 'I4'])
        classImplements(C, I5)
        self.assertEqual([i.getName() for i in implementedBy(C)],
                         ['I1', 'I2', 'I5', 'I3', 'I4'])

def make_dummy_request(dbsession: Session, registry: Registry) -> IRequest:
    """Creates a non-functional HTTP request with registry and dbsession configured.

    Useful for crafting requests with custom settings

    See also :func:`make_routable_request`.
    """

    @implementer(IRequest)
    class DummyRequest:
        pass

    _request = DummyRequest()
    _request.dbsession = dbsession
    _request.user = None
    _request.registry = registry

    return _request

def test_interfacesForTransport(self):
        """
        If the protocol objects returned by the factory given to
        L{ClientService} provide special "marker" interfaces for their
        transport - L{IHalfCloseableProtocol} or L{IFileDescriptorReceiver} -
        those interfaces will be provided by the protocol objects passed on to
        the reactor.
        """
        @implementer(IHalfCloseableProtocol, IFileDescriptorReceiver)
        class FancyProtocol(Protocol, object):
            """
            Provider of various interfaces.
            """
        cq, service = self.makeReconnector(protocolType=FancyProtocol)
        reactorFacing = cq.constructedProtocols[0]
        self.assertTrue(IFileDescriptorReceiver.providedBy(reactorFacing))
        self.assertTrue(IHalfCloseableProtocol.providedBy(reactorFacing))

def test_serializeIRenderable(self):
        """
        Test that flattening respects all of the IRenderable interface.
        """
        @implementer(IRenderable)
        class FakeElement(object):
            def render(ign,ored):
                return tags.p(
                    'hello, ',
                    tags.transparent(render='test'), ' - ',
                    tags.transparent(render='test'))
            def lookupRenderMethod(ign, name):
                self.assertEqual(name, 'test')
                return lambda ign, node: node('world')

        return gatherResults([
            self.assertFlattensTo(FakeElement(), b'<p>hello, world - world</p>'),
        ])

def test_renderable(self):
        """
        If a L{FlattenerError} is created with an L{IRenderable} provider root,
        the repr of that object is included in the string representation of the
        exception.
        """
        @implementer(IRenderable)
        class Renderable(object):
            def __repr__(self):
                return "renderable repr"

        self.assertEqual(
            str(FlattenerError(
                    RuntimeError("reason"), [Renderable()], [])),
            "Exception while flattening:\n"
            "  renderable repr\n"
            "RuntimeError: reason\n")

def integrationTest(self, hostName, expectedAddress, addressType):
        """
        Wrap L{AgentTestsMixin.integrationTest} with TLS.
        """
        authority, server = certificatesForAuthorityAndServer(hostName
                                                              .decode('ascii'))
        def tlsify(serverFactory):
            return TLSMemoryBIOFactory(server.options(), False, serverFactory)
        def tlsagent(reactor):
            from twisted.web.iweb import IPolicyForHTTPS
            from zope.interface import implementer
            @implementer(IPolicyForHTTPS)
            class Policy(object):
                def creatorForNetloc(self, hostname, port):
                    return optionsForClientTLS(hostname.decode("ascii"),
                                               trustRoot=authority)
            return client.Agent(reactor, contextFactory=Policy())
        (super(AgentHTTPSTests, self)
         .integrationTest(hostName, expectedAddress, addressType,
                          serverWrapper=tlsify,
                          createAgent=tlsagent,
                          scheme=b'https'))

def test_startedListeningLogMessage(self):
        """
        When a port starts, a message including a description of the associated
        factory is logged.
        """
        loggedMessages = self.observe()
        reactor = self.buildReactor()

        @implementer(ILoggingContext)
        class SomeFactory(ServerFactory):
            def logPrefix(self):
                return "Crazy Factory"

        factory = SomeFactory()
        p = self.getListeningPort(reactor, factory)
        expectedMessage = self.getExpectedStartListeningLogMessage(
            p, "Crazy Factory")
        self.assertEqual((expectedMessage,), loggedMessages[0]['message'])

def test_doubleHalfClose(self):
        """
        If one side half-closes its connection, and then the other side of the
        connection calls C{loseWriteConnection}, and then C{loseConnection} in
        {writeConnectionLost}, the connection is closed correctly.

        This rather obscure case used to fail (see ticket #3037).
        """
        @implementer(IHalfCloseableProtocol)
        class ListenerProtocol(ConnectableProtocol):

            def readConnectionLost(self):
                self.transport.loseWriteConnection()

            def writeConnectionLost(self):
                self.transport.loseConnection()

        class Client(ConnectableProtocol):
            def connectionMade(self):
                self.transport.loseConnection()

        # If test fails, reactor won't stop and we'll hit timeout:
        runProtocolsWithReactor(
            self, ListenerProtocol(), Client(), TCPCreator())

def test_consumerToProtocol(self):
        """
        L{IConsumer} providers can be adapted to L{IProtocol} providers using
        L{ProtocolToConsumerAdapter}.
        """
        result = []
        @implementer(IConsumer)
        class Consumer(object):
            def write(self, d):
                result.append(d)

        c = Consumer()
        protocol = IProtocol(c)
        protocol.dataReceived(b"hello")
        self.assertEqual(result, [b"hello"])
        self.assertIsInstance(protocol, ConsumerToProtocolAdapter)

def test_zope35(self):
        """
        Version 3.5 of L{zope.interface} has a C{implementer} method which
        cannot be used as a class decorator.
        """
        with SetAsideModule("zope"):
            self.install((3, 5))
            from zope.interface import Interface, implementer
            class IDummy(Interface):
                pass
            try:
                @implementer(IDummy)
                class Dummy(object):
                    pass
            except TypeError as exc:
                self.assertEqual(
                    "Can't use implementer with classes.  "
                    "Use one of the class-declaration functions instead.",
                    str(exc))

def test_honorsAcceptableCiphersArgument(self):
        """
        If acceptable ciphers are passed, they are used.
        """
        @implementer(interfaces.IAcceptableCiphers)
        class FakeAcceptableCiphers(object):
            def selectCiphers(self, _):
                return [sslverify.OpenSSLCipher(u'sentinel')]

        opts = sslverify.OpenSSLCertificateOptions(
            privateKey=self.sKey,
            certificate=self.sCert,
            acceptableCiphers=FakeAcceptableCiphers(),
        )
        opts._contextFactory = FakeContext
        ctx = opts.getContext()
        self.assertEqual(b'sentinel', ctx._cipherList)

def test_pullProducer(self):
        """
        Test a pull producer registered against a loopback transport.
        """
        @implementer(IPullProducer)
        class PullProducer(object):
            def __init__(self, toProduce):
                self.toProduce = toProduce

            def start(self, consumer):
                self.consumer = consumer
                self.consumer.registerProducer(self, False)

            def resumeProducing(self):
                self.consumer.write(self.toProduce.pop(0))
                if not self.toProduce:
                    self.consumer.unregisterProducer()
        return self._producertest(PullProducer)

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

            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_class_doesnt_have_required_method_derived(self):
        from zope.interface import Interface
        from zope.interface import implementer
        from zope.interface.exceptions import BrokenImplementation

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

        class IDerived(IBase):
            pass

        @implementer(IDerived)
        class Current(object):
            pass

        self.assertRaises(BrokenImplementation,
                          self._callFUT, IDerived, Current)