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)
python类implementer()的实例源码
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)
test_odd_declarations.py 文件源码
项目：AskTanmay-NLQA-System-
作者: tanmayb123
项目源码
文件源码
阅读 15
收藏 0
点赞 0
评论 0
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)
