[Zope3-checkins] CVS: Zope3/src/zope/app/adapter - __init__.py:1.1
adapter.py:1.1 configure.zcml:1.1 tests.py:1.1
Jim Fulton
jim at zope.com
Mon Mar 8 14:40:26 EST 2004
Update of /cvs-repository/Zope3/src/zope/app/adapter
In directory cvs.zope.org:/tmp/cvs-serv28862/src/zope/app/adapter
Added Files:
__init__.py adapter.py configure.zcml tests.py
Log Message:
Moved the adapter and presentation services from zope.app.services to
their own packages under zope.app. Also folded
zope.app.services.surrogate into the new adapter module,
zope.app.adapter.adapter.
=== Added File Zope3/src/zope/app/adapter/__init__.py ===
##############################################################################
#
# Copyright (c) 2004 Zope Corporation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.0 (ZPL). A copy of the ZPL should accompany this distribution.
# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
# FOR A PARTICULAR PURPOSE.
#
##############################################################################
"""Local-adapter support
$Id: __init__.py,v 1.1 2004/03/08 19:40:25 jim Exp $
"""
from zope.app.adapter.adapter import IAdapterRegistration
from zope.app.adapter.adapter import LocalAdapterRegistry
from zope.app.adapter.adapter import LocalAdapterBasedService
=== Added File Zope3/src/zope/app/adapter/adapter.py ===
##############################################################################
#
# Copyright (c) 2002 Zope Corporation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.0 (ZPL). A copy of the ZPL should accompany this distribution.
# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
# FOR A PARTICULAR PURPOSE.
#
##############################################################################
"""Adapter Service
$Id: adapter.py,v 1.1 2004/03/08 19:40:25 jim Exp $
"""
__metaclass__ = type
from persistent.dict import PersistentDict
from persistent import Persistent
from zope.app import zapi
from zope.app.services.registration import NotifyingRegistrationStack
from zope.interface.adapter import adapterImplied, Default
from zope.interface.adapter import Surrogate, AdapterRegistry
import sys
import zope.app.component.interfacefield
import zope.app.component.nextservice
import zope.app.container.contained
import zope.app.interfaces.services.registration
import zope.app.interfaces.services.service
import zope.app.security.permission
import zope.app.services.registration
import zope.component.interfaces
import zope.interface
import zope.schema
class LocalSurrogate(Surrogate):
"""Local surrogates
Local surrogates are transient, rather than persistent.
Their adapter data are stored in their registry objects.
"""
def __init__(self, spec, registry):
Surrogate.__init__(self, spec, registry)
self.registry = registry
registry.baseFor(spec).subscribe(self)
def clean(self):
spec = self.spec()
base = self.registry.baseFor(spec)
ladapters = self.registry.adapters.get(spec)
if ladapters:
adapters = base.adapters.copy()
adapters.update(ladapters)
else:
adapters = base.adapters
self.adapters = adapters
Surrogate.clean(self)
class LocalAdapterRegistry(AdapterRegistry, Persistent):
"""Local/persistent surrogate registry
"""
zope.interface.implements(
zope.app.interfaces.services.registration.IRegistry,
)
_surrogateClass = LocalSurrogate
next = None
subs = ()
def __init__(self, base, next=None):
self.base = base
self.adapters = {}
self.stacks = PersistentDict()
AdapterRegistry.__init__(self)
self.setNext(next)
def setNext(self, next, base=None):
if base is not None:
self.base = base
if self.next is not None:
self.next.removeSub(self)
if next is not None:
next.addSub(self)
self.next = next
self.adaptersChanged()
def addSub(self, sub):
self.subs += (sub, )
def removeSub(self, sub):
self.subs = tuple([s for s in self.subs if s is not sub])
def __getstate__(self):
state = Persistent.__getstate__(self).copy()
del state['_surrogates']
del state['_default']
del state['_remove']
return state
def __setstate__(self, state):
Persistent.__setstate__(self, state)
AdapterRegistry.__init__(self)
def baseFor(self, spec):
return self.base.get(spec)
def queryRegistrationsFor(self, registration, default=None):
stacks = self.stacks.get(registration.required)
if stacks:
stack = stacks.get((False, registration.with, registration.name,
registration.provided))
if stack is not None:
return stack
return default
_stackType = NotifyingRegistrationStack
def createRegistrationsFor(self, registration):
stacks = self.stacks.get(registration.required)
if stacks is None:
stacks = PersistentDict()
self.stacks[registration.required] = stacks
key = False, registration.with, registration.name, registration.provided
stack = stacks.get(key)
if stack is None:
stack = self._stackType(self)
stacks[key] = stack
return stack
def adaptersChanged(self, *args):
adapters = {}
if self.next is not None:
for required, radapters in self.next.adapters.iteritems():
adapters[required] = radapters.copy()
for required, stacks in self.stacks.iteritems():
if required is None:
required = Default
radapters = adapters.get(required)
if not radapters:
radapters = {}
adapters[required] = radapters
for key, stack in stacks.iteritems():
registration = stack.active()
if registration is not None:
radapters[key] = registration.factories
if adapters != self.adapters:
self.adapters = adapters
for surrogate in self._surrogates.values():
surrogate.dirty()
for sub in self.subs:
sub.adaptersChanged()
notifyActivated = notifyDeactivated = adaptersChanged
class LocalAdapterBasedService(
zope.app.container.contained.Contained,
Persistent,
):
"""A service that uses local surrogate registries
A local surrogate-based service needs to maintain connections
between it's surrogate registries and those of containing ans
sub-services.
The service must implement a setNext method that will be called
with the next local service, which may be None, and the global
service. This method will be called when a service is bound.
"""
zope.interface.implements(
zope.app.interfaces.services.service.IBindingAware,
)
def __updateNext(self, servicename):
next = zope.app.component.nextservice.getNextService(self, servicename)
global_ = zapi.getService(None, servicename)
if next == global_:
next = None
self.setNext(next, global_)
def bound(self, servicename):
self.__updateNext(servicename)
# Now, we need to notify any sub-site services. This is
# a bit complicated because our immediate subsites might not have
# the same service. Sigh
sm = zapi.getServiceManager(self)
self.__notifySubs(sm.subSites, servicename)
def unbound(self, servicename):
sm = zapi.getServiceManager(self)
self.__notifySubs(sm.subSites, servicename)
def __notifySubs(self, subs, servicename):
for sub in subs:
s = sub.queryLocalService(servicename)
if s is not None:
s.__updateNext(servicename)
else:
self.__notifySubs(sub.subSites, servicename)
class LocalAdapterService(LocalAdapterRegistry, LocalAdapterBasedService):
zope.interface.implements(
zope.component.interfaces.IAdapterService,
zope.app.interfaces.services.service.ISimpleService,
)
def __init__(self):
LocalAdapterRegistry.__init__(
self, zapi.getService(None, zapi.servicenames.Adapters)
)
class IAdapterRegistration(
zope.app.interfaces.services.registration.IRegistration):
required = zope.app.component.interfacefield.InterfaceField(
title = u"For interface",
description = u"The interface of the objects being adapted",
readonly = True,
basetype = None,
)
provided = zope.app.component.interfacefield.InterfaceField(
title = u"Provided interface",
description = u"The interface provided",
readonly = True,
required = True,
)
name = zope.schema.TextLine(
title=u"Name",
readonly=True,
required=False,
)
factoryName = zope.schema.BytesLine(
title=u"The dotted name of a factory for creating the adapter",
readonly = True,
required = True,
)
permission = zope.app.security.permission.PermissionField(
title=u"The permission required for use",
readonly=False,
required=False,
)
factories = zope.interface.Attribute(
"A sequence of factories to be called to construct the component"
)
class AdapterRegistration(zope.app.services.registration.SimpleRegistration):
zope.interface.implements(IAdapterRegistration)
serviceType = zapi.servicenames.Adapters
with = () # XXX Don't support multi-adapters yet
# XXX These should be positional arguments, except that required
# isn't passed in if it is omitted. To fix this, we need a
# required=False,explicitly_unrequired=True in the schema field
# so None will get passed in.
def __init__(self, provided, factoryName,
name='', required=None, permission=None):
self.required = required
self.provided = provided
self.name = name
self.factoryName = factoryName
self.permission = permission
def factories(self):
folder = self.__parent__.__parent__
factory = folder.resolve(self.factoryName)
return factory,
factories = property(factories)
# XXX Pickle backward compatability
AdapterConfiguration = AdapterRegistration
#BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB
import persistent
from zope.interface.adapter import ReadProperty
AdapterRegistration.required = ReadProperty(lambda self: self.forInterface)
AdapterRegistration.provided = ReadProperty(
lambda self: self.providedInterface)
AdapterRegistration.name = ReadProperty(lambda self: self.adapterName)
class AdapterService(persistent.Persistent):
pass
=== Added File Zope3/src/zope/app/adapter/configure.zcml ===
<configure
xmlns='http://namespaces.zope.org/zope'
xmlns:event='http://namespaces.zope.org/event'
xmlns:fssync='http://namespaces.zope.org/fssync'
i18n_domain='zope'
>
<!-- Adapter Service -->
<content class=".adapter.LocalAdapterService">
<factory permission="zope.ManageServices" />
<require
permission="zope.ManageServices"
interface="zope.app.interfaces.services.registration.IRegistry"
attributes="getRegisteredMatching"
/>
</content>
<content class=".adapter.AdapterRegistration">
<require
permission="zope.ManageServices"
interface=".adapter.IAdapterRegistration"
set_schema=".adapter.IAdapterRegistration"
/>
<require
permission="zope.ManageServices"
interface="zope.app.container.interfaces.IRemoveNotifiable"
/>
</content>
<modulealias module="zope.app.adapter.adapter"
alias="zope.app.services.adapter"
/>
<include package=".browser" />
</configure>
=== Added File Zope3/src/zope/app/adapter/tests.py ===
##############################################################################
#
# Copyright (c) 2003 Zope Corporation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.0 (ZPL). A copy of the ZPL should accompany this distribution.
# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
# FOR A PARTICULAR PURPOSE.
#
##############################################################################
"""Local Adapter Tests
Local surrogates and surrogate registries share declarations with
those "above" them.
Suppose we have a global AdapterRegistry:
>>> G = AdapterRegistry()
we also have a local surrogate registry, with G as it's base:
>>> L1 = LocalAdapterRegistry(G)
and so on:
>>> L2 = LocalAdapterRegistry(G, L1)
Now, if we declare an adapter globally:
>>> G.provideAdapter(IF1, IB1, [A11G])
we can query it locally:
>>> f2 = F2()
>>> a = L1.queryAdapter(f2, IB1)
>>> a.__class__.__name__
'A11G'
>>> a.args == (f2, )
True
>>> a = L2.queryAdapter(f2, IB1)
>>> a.__class__.__name__
'A11G'
>>> a.args == (f2, )
True
We can add local definitions:
>>> ra011 = Registration(required = IF0, provided=IB1, factory=A011)
>>> L1.createRegistrationsFor(ra011).activate(ra011)
and use it:
>>> f0 = F0()
>>> a = L1.queryAdapter(f0, IB1)
>>> a.__class__.__name__
'A011'
>>> a.args == (f0, )
True
>>> a = L2.queryAdapter(f0, IB1)
>>> a.__class__.__name__
'A011'
>>> a.args == (f0, )
True
but not outside L1:
>>> G.queryAdapter(f0, IB1)
Note that it doesn't override the non-local adapter:
>>> a = L1.queryAdapter(f2, IB1)
>>> a.__class__.__name__
'A11G'
>>> a.args == (f2, )
True
>>> a = L2.queryAdapter(f2, IB1)
>>> a.__class__.__name__
'A11G'
>>> a.args == (f2, )
True
because it was more specific.
Let's override the adapter in L2:
>>> ra112 = Registration(required = IF1, provided=IB1, factory=A112)
>>> L2.createRegistrationsFor(ra112).activate(ra112)
Now, in L2, we get the new adapter, because it's as specific and more
local than the one from G:
>>> a = L2.queryAdapter(f2, IB1)
>>> a.__class__.__name__
'A112'
>>> a.args == (f2, )
True
But we still get thye old one in L1
>>> a = L1.queryAdapter(f2, IB1)
>>> a.__class__.__name__
'A11G'
>>> a.args == (f2, )
True
Note that we can ask for less specific interfaces and still get the adapter:
>>> a = L2.queryAdapter(f2, IB0)
>>> a.__class__.__name__
'A112'
>>> a.args == (f2, )
True
>>> a = L1.queryAdapter(f2, IB0)
>>> a.__class__.__name__
'A11G'
>>> a.args == (f2, )
True
We get the more specific adapter even if there is a less-specific
adapter to B0:
>>> G.provideAdapter(IF1, IB1, [A10G])
>>> a = L2.queryAdapter(f2, IB0)
>>> a.__class__.__name__
'A112'
>>> a.args == (f2, )
True
But if we have an equally specific and equally local adapter to B0, it
will win:
>>> ra102 = Registration(required = IF1, provided=IB0, factory=A102)
>>> L2.createRegistrationsFor(ra102).activate(ra102)
>>> a = L2.queryAdapter(f2, IB0)
>>> a.__class__.__name__
'A102'
>>> a.args == (f2, )
True
We can deactivate registrations, which has the effect of deleting adapters:
>>> L2.queryRegistrationsFor(ra112).deactivate(ra112)
>>> a = L2.queryAdapter(f2, IB0)
>>> a.__class__.__name__
'A102'
>>> a.args == (f2, )
True
>>> a = L2.queryAdapter(f2, IB1)
>>> a.__class__.__name__
'A10G'
>>> a.args == (f2, )
True
>>> L2.queryRegistrationsFor(ra102).deactivate(ra102)
>>> a = L2.queryAdapter(f2, IB0)
>>> a.__class__.__name__
'A10G'
>>> a.args == (f2, )
True
$Id: tests.py,v 1.1 2004/03/08 19:40:25 jim Exp $
"""
def test_named_adapters():
"""
Suppose we have a global AdapterRegistry:
>>> G = AdapterRegistry()
we also have a local surrogate registry, with G as it's base:
>>> L1 = LocalAdapterRegistry(G)
and so on:
>>> L2 = LocalAdapterRegistry(G, L1)
Now, if we declare an adapter globally:
>>> G.provideAdapter(IF1, IB1, [A11G], name='bob')
we can query it locally:
>>> f2 = F2()
>>> L1.queryAdapter(f2, IB1)
>>> a = L1.queryNamedAdapter(f2, IB1, 'bob')
>>> a.__class__.__name__
'A11G'
>>> a.args == (f2, )
True
>>> L2.queryAdapter(f2, IB1)
>>> a = L2.queryNamedAdapter(f2, IB1, 'bob')
>>> a.__class__.__name__
'A11G'
>>> a.args == (f2, )
True
We can add local definitions:
>>> ra011 = Registration(required = IF0, provided=IB1, factory=A011,
... name='bob')
>>> L1.createRegistrationsFor(ra011).activate(ra011)
and use it:
>>> f0 = F0()
>>> L1.queryAdapter(f0, IB1)
>>> a = L1.queryNamedAdapter(f0, IB1, 'bob')
>>> a.__class__.__name__
'A011'
>>> a.args == (f0, )
True
>>> L2.queryAdapter(f0, IB1)
>>> a = L2.queryNamedAdapter(f0, IB1, 'bob')
>>> a.__class__.__name__
'A011'
>>> a.args == (f0, )
True
but not outside L1:
>>> G.queryNamedAdapter(f0, IB1, 'bob')
Note that it doesn't override the non-local adapter:
>>> L1.queryAdapter(f2, IB1)
>>> a = L1.queryNamedAdapter(f2, IB1, 'bob')
>>> a.__class__.__name__
'A11G'
>>> a.args == (f2, )
True
>>> L2.queryAdapter(f2, IB1)
>>> a = L2.queryNamedAdapter(f2, IB1, 'bob')
>>> a.__class__.__name__
'A11G'
>>> a.args == (f2, )
True
because it was more specific.
Let's override the adapter in L2:
>>> ra112 = Registration(required = IF1, provided=IB1, factory=A112,
... name='bob')
>>> L2.createRegistrationsFor(ra112).activate(ra112)
Now, in L2, we get the new adapter, because it's as specific and more
local than the one from G:
>>> L2.queryAdapter(f2, IB1)
>>> a = L2.queryNamedAdapter(f2, IB1, 'bob')
>>> a.__class__.__name__
'A112'
>>> a.args == (f2, )
True
But we still get thye old one in L1
>>> L1.queryAdapter(f2, IB1)
>>> a = L1.queryNamedAdapter(f2, IB1, 'bob')
>>> a.__class__.__name__
'A11G'
>>> a.args == (f2, )
True
Note that we can ask for less specific interfaces and still get the adapter:
>>> L2.queryAdapter(f2, IB0)
>>> a = L2.queryNamedAdapter(f2, IB0, 'bob')
>>> a.__class__.__name__
'A112'
>>> a.args == (f2, )
True
>>> L1.queryAdapter(f2, IB0)
>>> a = L1.queryNamedAdapter(f2, IB0, 'bob')
>>> a.__class__.__name__
'A11G'
>>> a.args == (f2, )
True
We get the more specific adapter even if there is a less-specific
adapter to B0:
>>> G.provideAdapter(IF1, IB1, [A10G], name='bob')
>>> L2.queryAdapter(f2, IB0)
>>> a = L2.queryNamedAdapter(f2, IB0, 'bob')
>>> a.__class__.__name__
'A112'
>>> a.args == (f2, )
True
But if we have an equally specific and equally local adapter to B0, it
will win:
>>> ra102 = Registration(required = IF1, provided=IB0, factory=A102,
... name='bob')
>>> L2.createRegistrationsFor(ra102).activate(ra102)
>>> L2.queryAdapter(f2, IB0)
>>> a = L2.queryNamedAdapter(f2, IB0, 'bob')
>>> a.__class__.__name__
'A102'
>>> a.args == (f2, )
True
We can deactivate registrations, which has the effect of deleting adapters:
>>> L2.queryRegistrationsFor(ra112).deactivate(ra112)
>>> L2.queryAdapter(f2, IB0)
>>> a = L2.queryNamedAdapter(f2, IB0, 'bob')
>>> a.__class__.__name__
'A102'
>>> a.args == (f2, )
True
>>> L2.queryAdapter(f2, IB1)
>>> a = L2.queryNamedAdapter(f2, IB1, 'bob')
>>> a.__class__.__name__
'A10G'
>>> a.args == (f2, )
True
>>> L2.queryRegistrationsFor(ra102).deactivate(ra102)
>>> L2.queryAdapter(f2, IB0)
>>> a = L2.queryNamedAdapter(f2, IB0, 'bob')
>>> a.__class__.__name__
'A10G'
>>> a.args == (f2, )
True
"""
def test_multi_adapters():
"""
Suppose we have a global AdapterRegistry:
>>> G = AdapterRegistry()
we also have a local surrogate registry, with G as it's base:
>>> L1 = LocalAdapterRegistry(G)
and so on:
>>> L2 = LocalAdapterRegistry(G, L1)
Now, if we declare an adapter globally:
>>> G.provideAdapter(IF1, IB1, [A11G], name='bob', with=(IR0,))
we can query it locally:
>>> f2 = F2()
>>> r = R1()
>>> a = L1.queryMultiAdapter((f2, r), IB1, 'bob')
>>> a.__class__.__name__
'A11G'
>>> a.args == (f2, r)
True
>>> a = L2.queryMultiAdapter((f2, r), IB1, 'bob')
>>> a.__class__.__name__
'A11G'
>>> a.args == (f2, r)
True
We can add local definitions:
>>> ra011 = Registration(required = IF0, provided=IB1, factory=A011,
... name='bob', with=(IR0,))
>>> L1.createRegistrationsFor(ra011).activate(ra011)
and use it:
>>> f0 = F0()
>>> a = L1.queryMultiAdapter((f0, r), IB1, 'bob')
>>> a.__class__.__name__
'A011'
>>> a.args == (f0, r)
True
>>> a = L2.queryMultiAdapter((f0, r), IB1, 'bob')
>>> a.__class__.__name__
'A011'
>>> a.args == (f0, r)
True
but not outside L1:
>>> G.queryMultiAdapter((f0, r), IB1, 'bob')
Note that it doesn't override the non-local adapter:
>>> a = L1.queryMultiAdapter((f2, r), IB1, 'bob')
>>> a.__class__.__name__
'A11G'
>>> a.args == (f2, r)
True
>>> a = L2.queryMultiAdapter((f2, r), IB1, 'bob')
>>> a.__class__.__name__
'A11G'
>>> a.args == (f2, r)
True
because it was more specific.
Let's override the adapter in L2:
>>> ra112 = Registration(required = IF1, provided=IB1, factory=A112,
... name='bob', with=(IR0,))
>>> L2.createRegistrationsFor(ra112).activate(ra112)
Now, in L2, we get the new adapter, because it's as specific and more
local than the one from G:
>>> a = L2.queryMultiAdapter((f2, r), IB1, 'bob')
>>> a.__class__.__name__
'A112'
>>> a.args == (f2, r)
True
But we still get the old one in L1
>>> a = L1.queryMultiAdapter((f2, r), IB1, 'bob')
>>> a.__class__.__name__
'A11G'
>>> a.args == (f2, r)
True
Note that we can ask for less specific interfaces and still get
the adapter:
>>> a = L2.queryMultiAdapter((f2, r), IB0, 'bob')
>>> a.__class__.__name__
'A112'
>>> a.args == (f2, r)
True
>>> a = L1.queryMultiAdapter((f2, r), IB0, 'bob')
>>> a.__class__.__name__
'A11G'
>>> a.args == (f2, r)
True
We get the more specific adapter even if there is a less-specific
adapter to B0:
>>> G.provideAdapter(IF1, IB1, [A10G], name='bob', with=(IR0,))
>>> a = L2.queryMultiAdapter((f2, r), IB0, 'bob')
>>> a.__class__.__name__
'A112'
>>> a.args == (f2, r)
True
But if we have an equally specific and equally local adapter to B0, it
will win:
>>> ra102 = Registration(required = IF1, provided=IB0, factory=A102,
... name='bob', with=(IR0,))
>>> L2.createRegistrationsFor(ra102).activate(ra102)
>>> a = L2.queryMultiAdapter((f2, r), IB0, 'bob')
>>> a.__class__.__name__
'A102'
>>> a.args == (f2, r)
True
We can deactivate registrations, which has the effect of deleting adapters:
>>> L2.queryRegistrationsFor(ra112).deactivate(ra112)
>>> a = L2.queryMultiAdapter((f2, r), IB0, 'bob')
>>> a.__class__.__name__
'A102'
>>> a.args == (f2, r)
True
>>> a = L2.queryMultiAdapter((f2, r), IB1, 'bob')
>>> a.__class__.__name__
'A10G'
>>> a.args == (f2, r)
True
>>> L2.queryRegistrationsFor(ra102).deactivate(ra102)
>>> L2.queryAdapter(f2, IB0)
>>> a = L2.queryMultiAdapter((f2, r), IB0, 'bob')
>>> a.__class__.__name__
'A10G'
>>> a.args == (f2, r)
True
"""
def test_persistence():
"""
>>> db = DB()
>>> conn1 = db.open()
>>> G = globalAdapterRegistry
>>> L1 = LocalAdapterRegistry(G)
>>> L2 = LocalAdapterRegistry(G, L1)
>>> conn1.root()['L1'] = L1
>>> conn1.root()['L2'] = L2
>>> G.provideAdapter(IF1, IB1, [A11G], name='bob')
>>> f2 = F2()
>>> L1.queryAdapter(f2, IB1)
>>> a = L1.queryNamedAdapter(f2, IB1, 'bob')
>>> a.__class__.__name__
'A11G'
>>> a.args == (f2, )
True
>>> L2.queryAdapter(f2, IB1)
>>> a = L2.queryNamedAdapter(f2, IB1, 'bob')
>>> a.__class__.__name__
'A11G'
>>> a.args == (f2, )
True
We can add local definitions:
>>> ra011 = Registration(required = IF0, provided=IB1, factory=A011,
... name='bob')
>>> L1.createRegistrationsFor(ra011).activate(ra011)
and use it:
>>> f0 = F0()
>>> L1.queryAdapter(f0, IB1)
>>> a = L1.queryNamedAdapter(f0, IB1, 'bob')
>>> a.__class__.__name__
'A011'
>>> a.args == (f0, )
True
>>> L2.queryAdapter(f0, IB1)
>>> a = L2.queryNamedAdapter(f0, IB1, 'bob')
>>> a.__class__.__name__
'A011'
>>> a.args == (f0, )
True
but not outside L1:
>>> G.queryAdapter(f0, IB1)
Note that it doesn't override the non-local adapter:
>>> L1.queryAdapter(f2, IB1)
>>> a = L1.queryNamedAdapter(f2, IB1, 'bob')
>>> a.__class__.__name__
'A11G'
>>> a.args == (f2, )
True
>>> L2.queryAdapter(f2, IB1)
>>> a = L2.queryNamedAdapter(f2, IB1, 'bob')
>>> a.__class__.__name__
'A11G'
>>> a.args == (f2, )
True
because it was more specific.
Let's override the adapter in L2:
>>> ra112 = Registration(required = IF1, provided=IB1, factory=A112,
... name='bob')
>>> L2.createRegistrationsFor(ra112).activate(ra112)
Now, in L2, we get the new adapter, because it's as specific and more
local than the one from G:
>>> L2.queryAdapter(f2, IB1)
>>> a = L2.queryNamedAdapter(f2, IB1, 'bob')
>>> a.__class__.__name__
'A112'
>>> a.args == (f2, )
True
But we still get the old one in L1
>>> L1.queryAdapter(f2, IB1)
>>> a = L1.queryNamedAdapter(f2, IB1, 'bob')
>>> a.__class__.__name__
'A11G'
>>> a.args == (f2, )
True
Note that we can ask for less specific interfaces and still get the adapter:
>>> L2.queryAdapter(f2, IB0)
>>> a = L2.queryNamedAdapter(f2, IB0, 'bob')
>>> a.__class__.__name__
'A112'
>>> a.args == (f2, )
True
>>> L1.queryAdapter(f2, IB0)
>>> a = L1.queryNamedAdapter(f2, IB0, 'bob')
>>> a.__class__.__name__
'A11G'
>>> a.args == (f2, )
True
We get the more specific adapter even if there is a less-specific
adapter to B0:
>>> G.provideAdapter(IF0, IB0, [A00G], name='bob')
>>> L2.queryAdapter(f2, IB0)
>>> a = L2.queryNamedAdapter(f2, IB0, 'bob')
>>> a.__class__.__name__
'A112'
>>> a.args == (f2, )
True
But if we have an equally specific and equally local adapter to B0, it
will win:
>>> ra102 = Registration(required = IF1, provided=IB0, factory=A102,
... name='bob')
>>> L2.createRegistrationsFor(ra102).activate(ra102)
>>> L2.queryAdapter(f2, IB0)
>>> a = L2.queryNamedAdapter(f2, IB0, 'bob')
>>> a.__class__.__name__
'A102'
>>> a.args == (f2, )
True
>>> L1.queryNamedAdapter(f2, IB0, 'bob').__class__.__name__
'A11G'
>>> L1.queryNamedAdapter(f2, IB1, 'bob').__class__.__name__
'A11G'
>>> L2.queryNamedAdapter(f2, IB0, 'bob').__class__.__name__
'A102'
>>> L2.queryNamedAdapter(f2, IB1, 'bob').__class__.__name__
'A112'
>>> get_transaction().commit()
Now, let's open another transaction:
>>> conn2 = db.open()
>>> L1 = conn2.root()['L1']
>>> L2 = conn2.root()['L2']
We should get the same outputs:
>>> L1.queryNamedAdapter(f2, IB0, 'bob').__class__.__name__
'A11G'
>>> L1.queryNamedAdapter(f2, IB1, 'bob').__class__.__name__
'A11G'
>>> L2.queryNamedAdapter(f2, IB0, 'bob').__class__.__name__
'A102'
>>> L2.queryNamedAdapter(f2, IB1, 'bob').__class__.__name__
'A112'
We can deactivate registrations, which has the effect of deleting adapters:
>>> L2.queryRegistrationsFor(ra112).deactivate(ra112)
>>> L2.queryRegistrationsFor(ra102).deactivate(ra102)
>>> L1.queryNamedAdapter(f2, IB0, 'bob').__class__.__name__
'A11G'
>>> L1.queryNamedAdapter(f2, IB1, 'bob').__class__.__name__
'A11G'
>>> L2.queryNamedAdapter(f2, IB0, 'bob').__class__.__name__
'A11G'
>>> L2.queryNamedAdapter(f2, IB1, 'bob').__class__.__name__
'A11G'
>>> get_transaction().commit()
If we look back at the first connection, we should get the same data:
>>> conn1.sync()
>>> L1 = conn1.root()['L1']
>>> L2 = conn1.root()['L2']
We should see the result of the deactivations:
>>> L1.queryNamedAdapter(f2, IB0, 'bob').__class__.__name__
'A11G'
>>> L1.queryNamedAdapter(f2, IB1, 'bob').__class__.__name__
'A11G'
>>> L2.queryNamedAdapter(f2, IB0, 'bob').__class__.__name__
'A11G'
>>> L2.queryNamedAdapter(f2, IB1, 'bob').__class__.__name__
'A11G'
Cleanup:
>>> G.__init__()
>>> db.close()
"""
def test_local_default():
"""
>>> G = AdapterRegistry()
>>> L1 = LocalAdapterRegistry(G)
>>> r = Registration(required = None, provided=IB1, factory=Adapter)
>>> L1.createRegistrationsFor(r).activate(r)
>>> f2 = F2()
>>> L1.queryAdapter(f2, IB1).__class__.__name__
'Adapter'
"""
def test_changing_next():
"""
>>> G = AdapterRegistry()
>>> L1 = LocalAdapterRegistry(G)
>>> L2 = LocalAdapterRegistry(G, L1)
>>> f2 = F2()
>>> L2.queryAdapter(f2, IB1).__class__.__name__
'NoneType'
>>> G.provideAdapter(IF1, IB1, [A11G])
>>> L2.queryAdapter(f2, IB1).__class__.__name__
'A11G'
>>> class A111(Adapter):
... pass
>>> ra111 = Registration(required = IF1, provided=IB1, factory=A111)
>>> L1.createRegistrationsFor(ra111).activate(ra111)
>>> L2.queryAdapter(f2, IB1).__class__.__name__
'A111'
>>> L1.next
>>> L2.next == L1
True
>>> L1.subs == (L2,)
True
>>> L3 = LocalAdapterRegistry(G, L1)
>>> L2.setNext(L3)
>>> L2.next == L3
True
>>> L3.next == L1
True
>>> L1.subs == (L3,)
True
>>> L3.subs == (L2,)
True
>>> class A113(Adapter):
... pass
>>> ra113 = Registration(required = IF1, provided=IB1, factory=A113)
>>> L3.createRegistrationsFor(ra113).activate(ra113)
>>> L2.queryAdapter(f2, IB1).__class__.__name__
'A113'
>>> L2.setNext(L1)
>>> L2.next == L1
True
>>> L3.next == L1
True
>>> L1.subs == (L3, L2)
True
>>> L3.subs == ()
True
>>> L2.queryAdapter(f2, IB1).__class__.__name__
'A111'
"""
def test_LocalAdapterBasedService():
"""
Setup folders and service managers:
>>> from zope.app.tests import setup
>>> setup.placefulSetUp()
>>> root = setup.buildSampleFolderTree()
>>> sm = setup.createServiceManager(root)
>>> sm1 = setup.createServiceManager(root['folder1'])
>>> sm1_1 = setup.createServiceManager(root['folder1']['folder1_1'])
>>> sm1_1_1 = setup.createServiceManager(
... root['folder1']['folder1_1']['folder1_1_1'])
Define the service
>>> gsm = zapi.getServiceManager(None)
>>> gsm.defineService('F', IF1)
Create the global service
>>> g = F2()
>>> gsm.provideService('F', g)
Create a local service class, which must define setNext:
>>> import zope.app.interfaces.services.service
>>> class LocalF(LocalAdapterBasedService):
... zope.interface.implements(
... IF2,
... zope.app.interfaces.services.service.ISimpleService,
... )
... def setNext(self, next, global_):
... self.next, self.global_ = next, global_
If we add a local service, It gets it's next and global_ attrs set:
>>> f1 = LocalF()
>>> hasattr(f1, 'next') or hasattr(f1, 'global_')
False
>>> setup.addService(sm1, 'F', f1) is f1
True
>>> (f1.next, f1.global_) == (None, g)
True
If we add another service below, it's next will point to the one
above:
>>> f1_1_1 = LocalF()
>>> setup.addService(sm1_1_1, 'F', f1_1_1) is f1_1_1
True
>>> (f1_1_1.next, f1_1_1.global_) == (f1, g)
True
We can insert a service in an intermediate site:
>>> f1_1 = LocalF()
>>> setup.addService(sm1_1, 'F', f1_1) is f1_1
True
>>> (f1_1.next, f1_1.global_) == (f1, g)
True
>>> (f1_1_1.next, f1_1_1.global_) == (f1_1, g)
True
Deactivating services adjust the relevant next pointers
>>> default = zapi.traverse(sm1_1, 'default')
>>> rm = default.getRegistrationManager()
>>> rm.values()[0].status = RegisteredStatus
>>> (f1_1_1.next, f1_1_1.global_) == (f1, g)
True
>>> default = zapi.traverse(sm1, 'default')
>>> rm = default.getRegistrationManager()
>>> rm.values()[0].status = RegisteredStatus
>>> (f1_1_1.next, f1_1_1.global_) == (None, g)
True
>>> setup.placefulTearDown()
"""
import unittest
from zope.testing.doctestunit import DocTestSuite
from zope.interface.adapter import AdapterRegistry
from zope.app.adapter.adapter import LocalAdapterRegistry
from zope.app.adapter.adapter import LocalAdapterBasedService
import zope.interface
from ZODB.tests.util import DB
from transaction import get_transaction
from zope.app import zapi
from zope.app.interfaces.services.registration import RegisteredStatus
class IF0(zope.interface.Interface):
pass
class IF1(IF0):
pass
class IF2(IF1):
pass
class IB0(zope.interface.Interface):
pass
class IB1(IB0):
pass
class IR0(zope.interface.Interface):
pass
class IR1(IR0):
pass
class R1:
zope.interface.implements(IR1)
class F0:
zope.interface.implements(IF0)
class F2:
zope.interface.implements(IF2)
class Adapter:
def __init__(self, *args):
self.args = args
class A00G(Adapter):
pass
class A11G(Adapter):
pass
class A112(Adapter):
pass
class A10G(Adapter):
pass
class A102(Adapter):
pass
class A011(Adapter):
pass
class Registration:
name=u''
with=()
provided=zope.interface.Interface
required=None
def __init__(self, **kw):
self.__dict__.update(kw)
def __repr__(self):
return "<Registration %s>" % self.__dict__
def factories(self):
return self.factory,
factories = property(factories)
# Create a picklable global registry. The pickleability of other
# global surrogate registries is beyond the scope of these tests:
class GlobalAdapterRegistry(AdapterRegistry):
def __reduce__(self):
return 'globalAdapterRegistry'
globalAdapterRegistry = GlobalAdapterRegistry()
class TestStack:
registration = None
def __init__(self, parent):
self.__parent__ = parent
def activate(self, registration):
self.registration = registration
self.__parent__.notifyActivated(self, registration)
def deactivate(self, registration):
self.registration = None
self.__parent__.notifyDeactivated(self, registration)
def active(self):
return self.registration
class LocalAdapterRegistry(LocalAdapterRegistry):
"""For testing, use custom stack type
"""
_stackType = TestStack
def test_suite():
return unittest.TestSuite((
DocTestSuite(),
))
if __name__ == '__main__': unittest.main()
More information about the Zope3-Checkins
mailing list