[Zope3-checkins] SVN: Zope3/trunk/src/zope/app/component/ Renamed
nextservice.py to localservice.py, because that's what it is
Albertas Agejevas
alga at pov.lt
Fri May 21 21:59:25 EDT 2004
Log message for revision 24876:
Renamed nextservice.py to localservice.py, because that's what it is
about. Renamed ...ServiceManager() functions to ...Services() to be
in line with getServices().
-=-
Copied: Zope3/trunk/src/zope/app/component/localservice.py (from rev 24814, Zope3/trunk/src/zope/app/component/nextservice.py)
===================================================================
--- Zope3/trunk/src/zope/app/component/nextservice.py 2004-05-18 22:54:34 UTC (rev 24814)
+++ Zope3/trunk/src/zope/app/component/localservice.py 2004-05-22 01:59:25 UTC (rev 24876)
@@ -0,0 +1,165 @@
+##############################################################################
+#
+# 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.
+#
+##############################################################################
+"""Support for delegation among service managers
+
+$Id$
+"""
+
+from zope.interface import implements
+from zope.component.exceptions import ComponentLookupError
+from zope.component.service import serviceManager
+from zope.component.interfaces import IServiceService
+from zope.proxy import removeAllProxies
+from zope.app.site.interfaces import ISite, ISiteManager
+from zope.app.event.interfaces import ISubscriber
+from zope.thread import thread_globals
+from zope.testing.cleanup import addCleanUp
+
+# placeful service manager convenience tools
+
+# XXX <SteveA> What we really want here is
+#
+# getLocalServices(context)
+# # if context is contained in a service manager, returns the service manager
+# # if context is a service manager, returns context
+# # otherwise, raises ComponentLookupError('Services')
+#
+# getLocalService(context, name):
+# # Returns the local service with that name
+# # This is the same as getService, so why bother? parity.
+#
+# getNextServices(context, name):
+# # looks up the local service manager, then gets the next higher one
+# # and returns it
+#
+# getNextService(context, name):
+# # Returns the next service manager's service with that name.
+#
+# plus queryXXX versions.
+
+
+##def getLocalService(context, name):
+## service = queryLocalService(context, name)
+## if service is None:
+## raise ComponentLookupError('service', name)
+## return service
+##
+##def queryLocalService(context, name, default=None):
+## try:
+## sm = getLocalServices(context)
+## except ComponentLookupError:
+## return default
+## return sm.queryService(name, default)
+
+def queryNextService(context, name, default=None):
+ try:
+ return getNextService(context, name)
+ except ComponentLookupError:
+ return default
+
+def getNextService(context, name):
+ """Returns the service with the given name from the next service manager.
+ """
+ service = getNextServices(context).queryService(name)
+ if service is None:
+ raise ComponentLookupError(name)
+ return service
+
+def getNextServices(context):
+ """Returns the next service manager to the one that contains 'context'.
+ """
+ return getLocalServices(context).next
+
+def queryNextServices(context, default=None):
+ try:
+ return getNextServices(context)
+ except ComponentLookupError:
+ return default
+
+def queryLocalServices(context, default=None):
+ try:
+ return getLocalServices(context)
+ except ComponentLookupError:
+ return default
+
+def getLocalServices(context):
+ """Returns the service manager that contains 'context'.
+
+ If context is a local service, returns the service manager that
+ contains that service. If context is a service manager, returns context.
+
+ Otherwise, raises ComponentLookupError('Services')
+ """
+
+ # IMPORTANT
+ #
+ # This is not allowed to use any services to get its job done!
+
+ while not (context is None or
+ ISiteManager.providedBy(removeAllProxies(context))):
+ context = getattr(context, '__parent__', None)
+ if context is None:
+ raise ComponentLookupError('Services')
+ else:
+ return context
+
+def serviceServiceAdapter(ob):
+ """An adapter ILocation -> IServiceService.
+
+ The ILocation is interpreted flexibly, we just check for
+ __parent__.
+ """
+ current = ob
+ while True:
+ if ISite.providedBy(current):
+ return current.getSiteManager()
+ current = getattr(current, '__parent__', None)
+ if current is None:
+ raise ComponentLookupError("Could not adapt %r to"
+ " IServiceService" % (ob, ))
+
+
+class ThreadSiteSubscriber:
+ """A subscriber to BeforeTraverseEvent
+
+ Sets the 'site' thread global if the object traversed is a site.
+ """
+
+ implements(ISubscriber)
+ def notify(self, event):
+ if ISite.providedBy(event.object):
+ thread_globals().site = event.object
+
+
+threadSiteSubscriber = ThreadSiteSubscriber()
+
+
+class ClearThreadSiteSubscriber:
+ """A subscriber to EndRequestEvent
+
+ Cleans up the site thread global after the request is processed.
+ """
+
+ implements(ISubscriber)
+ def notify(self, event):
+ clearSite()
+
+clearThreadSiteSubscriber = ClearThreadSiteSubscriber()
+
+
+def clearSite():
+ """Clear the site thread global"""
+ thread_globals().site = None
+
+addCleanUp(clearSite)
Property changes on: Zope3/trunk/src/zope/app/component/localservice.py
___________________________________________________________________
Name: cvs2svn:cvs-rev
+ 1.10
Name: svn:keywords
+ Id
Name: svn:eol-style
+ native
Deleted: Zope3/trunk/src/zope/app/component/nextservice.py
===================================================================
--- Zope3/trunk/src/zope/app/component/nextservice.py 2004-05-22 01:57:30 UTC (rev 24875)
+++ Zope3/trunk/src/zope/app/component/nextservice.py 2004-05-22 01:59:25 UTC (rev 24876)
@@ -1,79 +0,0 @@
-##############################################################################
-#
-# 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.
-#
-##############################################################################
-"""Support for delegation among service managers
-
-$Id$
-"""
-
-from zope.component.exceptions import ComponentLookupError
-from zope.component.service import serviceManager
-from zope.proxy import removeAllProxies
-from zope.app.site.interfaces import ISite
-from zope.app.component.hooks import getServiceManager_hook
-
-# placeful service manager convenience tools
-
-def queryNextServiceManager(context, default=None):
- try:
- return getNextServiceManager(context)
- except ComponentLookupError:
- return default
-
-def getNextService(context, name):
- service = queryNextService(context, name)
- if service is None:
- raise ComponentLookupError('service', name)
- return service
-
-def queryNextService(context, name, default=None):
- try:
- sm = getNextServiceManager(context)
- except ComponentLookupError:
- return default
- return sm.queryService(name, default)
-
-def getNextServiceManager(context):
- """if the context is a service manager or a placeful service, tries
- to return the next highest service manager"""
-
- # IMPORTANT
- #
- # This is not allowed to use any services to get it's job done!
-
- # get this service manager
- sm = getServiceManager_hook(context)
- if sm is serviceManager:
- raise ComponentLookupError('Services')
-
- # get the service manager container, which ought to be the context
- # contaioner.
- container = sm.__parent__
-
- # But we're *really* paranoid, so we'll double check.
- while ((container is not None) and not
- ISite.providedBy(removeAllProxies(container))
- ):
- container = container.__parent__ # we should be
-
- # Now we need to step up so we can look for a service manager above.
- context = getattr(container, '__parent__', None)
-
- # But we have to make sure we haven't got the same object..
- while (context is not None) and (context == container):
- context = context.__parent__
-
- if context is None:
- return serviceManager
-
- return getServiceManager_hook(context, local=True)
Copied: Zope3/trunk/src/zope/app/component/tests/test_localservice.py (from rev 24814, Zope3/trunk/src/zope/app/component/tests/test_nextservice.py)
===================================================================
--- Zope3/trunk/src/zope/app/component/tests/test_nextservice.py 2004-05-18 22:54:34 UTC (rev 24814)
+++ Zope3/trunk/src/zope/app/component/tests/test_localservice.py 2004-05-22 01:59:25 UTC (rev 24876)
@@ -0,0 +1,268 @@
+##############################################################################
+#
+# Copyright (c) 2001, 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.
+#
+##############################################################################
+"""
+$Id$
+"""
+
+import unittest
+from zope.component import getGlobalServices
+from zope.app.component.hooks import getServices_hook
+from zope.app.component.localservice import serviceServiceAdapter
+from zope.app.site.interfaces import IPossibleSite, ISite, ISiteManager
+from zope.app.traversing import IContainmentRoot
+from zope.component.exceptions import ComponentLookupError
+from zope.component.interfaces import IServiceService
+from zope.component.service import serviceManager
+from zope.thread import thread_globals
+from zope.interface import implements, directlyProvides, directlyProvidedBy
+from zope.interface.verify import verifyObject
+from zope.app.event.interfaces import ISubscriber
+from zope.thread import thread_globals
+from zope.app.tests.setup import placelessSetUp, placelessTearDown
+from zope.app.tests import ztapi
+
+class ServiceManager:
+ implements(ISiteManager)
+
+ def __init__(self):
+ self.dummy_service = object()
+
+ def queryService(self, name, default=None):
+ if name == 'dummy':
+ return self.dummy_service
+ else:
+ return default
+
+class Folder:
+ implements(IPossibleSite)
+
+ sm = None
+
+ def getSiteManager(self, default=None):
+ return self.sm
+
+ def setSiteManager(self, sm):
+ self.sm = sm
+ sm.__parent__ = self
+ directlyProvides(self, ISite, directlyProvidedBy(self))
+
+class Package:
+ pass
+
+class Root(Folder):
+ implements(IContainmentRoot, ISite)
+ def getSiteManager(self):
+ return getGlobalServices()
+
+class ServiceServiceStub:
+ implements(IServiceService)
+
+
+def Wrapper(ob, container):
+ ob.__parent__ = container
+ return ob
+
+class Test(unittest.TestCase):
+
+ def setUp(self):
+ unittest.TestCase.setUp(self)
+ placelessSetUp()
+ root = Root()
+
+ f1 = Wrapper(Folder(), root)
+ sm1 = ServiceManager()
+ f1.setSiteManager(sm1)
+ p1 = Wrapper(Package(), sm1)
+
+ f2 = Wrapper(Folder(), f1)
+ sm2 = ServiceManager()
+ f2.setSiteManager(sm2)
+ p2 = Wrapper(Package(), sm2)
+
+ sm1.next = serviceManager
+ sm2.next = sm1
+
+ self.root = root
+ self.f1 = f1
+ self.f2 = f2
+ self.sm1 = sm1
+ self.sm2 = sm2
+ self.p1 = p1
+ self.p2 = p2
+ self.unparented_folder = Folder()
+ self.unrooted_subfolder = Wrapper(Folder(), self.unparented_folder)
+
+ ztapi.provideAdapter(None, IServiceService, serviceServiceAdapter)
+
+ def tearDown(self):
+ thread_globals().site = None
+ placelessTearDown()
+
+ def test_getServices(self):
+ self.assertEqual(getServices_hook(None), serviceManager)
+ self.assertEqual(getServices_hook(self.root), serviceManager)
+ self.assertEqual(getServices_hook(self.f1), self.sm1)
+ self.assertEqual(getServices_hook(self.f2), self.sm2)
+ thread_globals().site = self.f2
+ self.assertEqual(getServices_hook(None), self.sm2)
+
+ def test_queryNextService(self):
+ from zope.app.component.localservice import queryNextService
+ self.assert_(queryNextService(self.sm2, 'dummy') is
+ self.sm1.dummy_service)
+ self.assert_(queryNextService(self.p2, 'dummy') is
+ self.sm1.dummy_service)
+ marker = object()
+ self.assert_(queryNextService(self.p1, 'dummy', marker) is marker)
+
+ def test_getNextService(self):
+ from zope.app.component.localservice import getNextService
+ self.assert_(getNextService(self.sm2, 'dummy') is
+ self.sm1.dummy_service)
+ self.assert_(getNextService(self.p2, 'dummy') is
+ self.sm1.dummy_service)
+ self.assertRaises(ComponentLookupError,
+ getNextService, self.p1, 'dummy')
+
+ def test_queryNextServices(self):
+ from zope.app.component.localservice import queryNextServices
+ marker = object()
+ self.assert_(queryNextServices(self.root, marker) is marker)
+ self.assert_(queryNextServices(self.f1, marker) is marker)
+ self.assert_(queryNextServices(self.f2, marker) is marker)
+ self.assertEqual(queryNextServices(self.sm1), serviceManager)
+ self.assertEqual(queryNextServices(self.sm2), self.sm1)
+ self.assertEqual(queryNextServices(self.p1), serviceManager)
+ self.assertEqual(queryNextServices(self.p2), self.sm1)
+
+ self.assert_(queryNextServices(self.unparented_folder, marker)
+ is marker)
+ self.assert_(queryNextServices(self.unrooted_subfolder, marker)
+ is marker)
+
+ def test_getNextServices(self):
+ from zope.app.component.localservice import getNextServices
+ self.assertRaises(ComponentLookupError,
+ getNextServices, self.root)
+ self.assertRaises(ComponentLookupError,
+ getNextServices, self.f1)
+ self.assertRaises(ComponentLookupError,
+ getNextServices, self.f2)
+ self.assertEqual(getNextServices(self.sm1), serviceManager)
+ self.assertEqual(getNextServices(self.sm2), self.sm1)
+ self.assertEqual(getNextServices(self.p1), serviceManager)
+ self.assertEqual(getNextServices(self.p2), self.sm1)
+
+ self.assertRaises(ComponentLookupError,
+ getNextServices, self.unparented_folder)
+ self.assertRaises(ComponentLookupError,
+ getNextServices, self.unrooted_subfolder)
+
+ def test_queryLocalServices(self):
+ from zope.app.component.localservice import queryLocalServices
+ marker = object()
+ self.assert_(queryLocalServices(self.root, marker) is marker)
+ self.assert_(queryLocalServices(self.f1, marker) is marker)
+ self.assert_(queryLocalServices(self.f2, marker) is marker)
+ self.assertEqual(queryLocalServices(self.sm1), self.sm1)
+ self.assertEqual(queryLocalServices(self.sm2), self.sm2)
+ self.assertEqual(queryLocalServices(self.p1), self.sm1)
+ self.assertEqual(queryLocalServices(self.p2), self.sm2)
+
+ self.assert_(queryLocalServices(self.unparented_folder, marker)
+ is marker)
+ self.assert_(queryLocalServices(self.unrooted_subfolder, marker)
+ is marker)
+
+ def test_getLocalServices(self):
+ from zope.app.component.localservice import getLocalServices
+ self.assertRaises(ComponentLookupError,
+ getLocalServices, self.root)
+ self.assertRaises(ComponentLookupError,
+ getLocalServices, self.f1)
+ self.assertRaises(ComponentLookupError,
+ getLocalServices, self.f2)
+ self.assertEqual(getLocalServices(self.sm1), self.sm1)
+ self.assertEqual(getLocalServices(self.sm2), self.sm2)
+ self.assertEqual(getLocalServices(self.p1), self.sm1)
+ self.assertEqual(getLocalServices(self.p2), self.sm2)
+
+ unparented_folder = Folder()
+ self.assertRaises(ComponentLookupError,
+ getLocalServices, unparented_folder)
+ unrooted_subfolder = Wrapper(Folder(), unparented_folder)
+ self.assertRaises(ComponentLookupError,
+ getLocalServices, unrooted_subfolder)
+
+ def test_serviceServiceAdapter(self):
+ from zope.app.component.localservice import serviceServiceAdapter
+
+ # If it is a site, return the service service.
+ ss = ServiceServiceStub()
+ site = Folder()
+ site.setSiteManager(ss)
+ self.assertEqual(serviceServiceAdapter(site), ss)
+
+ # If it is locatable (has __parent__), "acquire" the site
+ # and return the service service
+ ob = Folder()
+ ob.__parent__ = site
+ self.assertEqual(serviceServiceAdapter(ob), ss)
+ ob2 = Folder()
+ ob2.__parent__ = ob
+ self.assertEqual(serviceServiceAdapter(ob2), ss)
+
+ # If it does we are unable to find a service service, raise
+ # ComponentLookupError
+ orphan = Folder()
+ self.assertRaises(ComponentLookupError, serviceServiceAdapter, orphan)
+
+ def test_setThreadSite_clearThreadSite(self):
+ from zope.app.component.localservice import threadSiteSubscriber
+ from zope.app.component.localservice import clearSite
+ from zope.app.publication.zopepublication import BeforeTraverseEvent
+
+ verifyObject(ISubscriber, threadSiteSubscriber)
+
+ globals = thread_globals()
+ self.assertEqual(globals.site, None)
+
+ # A non-site is traversed
+ ob = object()
+ request = object()
+ ev = BeforeTraverseEvent(ob, request)
+ threadSiteSubscriber.notify(ev)
+
+ self.assertEqual(globals.site, None)
+
+ # A site is traversed
+ ss = ServiceServiceStub()
+ site = Folder()
+ site.setSiteManager(ss)
+
+ ev = BeforeTraverseEvent(site, request)
+ threadSiteSubscriber.notify(ev)
+
+ self.assertEqual(globals.site, site)
+
+ clearSite()
+
+ self.assertEqual(globals.site, None)
+
+
+def test_suite():
+ return unittest.makeSuite(Test)
+
+if __name__ == '__main__':
+ unittest.main(defaultTest='test_suite')
Property changes on: Zope3/trunk/src/zope/app/component/tests/test_localservice.py
___________________________________________________________________
Name: cvs2svn:cvs-rev
+ 1.10
Name: svn:keywords
+ Id
Name: svn:eol-style
+ native
Deleted: Zope3/trunk/src/zope/app/component/tests/test_nextservice.py
===================================================================
--- Zope3/trunk/src/zope/app/component/tests/test_nextservice.py 2004-05-22 01:57:30 UTC (rev 24875)
+++ Zope3/trunk/src/zope/app/component/tests/test_nextservice.py 2004-05-22 01:59:25 UTC (rev 24876)
@@ -1,102 +0,0 @@
-##############################################################################
-#
-# Copyright (c) 2001, 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.
-#
-##############################################################################
-"""
-$Id$
-"""
-
-from unittest import TestCase, main, makeSuite
-from zope.app.component.hooks import getServiceManager_hook
-from zope.app.component.nextservice import getNextServiceManager
-from zope.app.site.interfaces import IPossibleSite, ISite
-from zope.app.traversing import IContainmentRoot
-from zope.component.exceptions import ComponentLookupError
-from zope.component.interfaces import IServiceService
-from zope.component.service import serviceManager
-from zope.interface import implements, directlyProvides, directlyProvidedBy
-
-
-class ServiceManager:
- implements(IServiceService)
-
-class Folder:
- implements(IPossibleSite)
-
- sm = None
-
- def getSiteManager(self, default=None):
- return self.sm
-
- def setSiteManager(self, sm):
- self.sm = sm
- sm.__parent__ = self
- directlyProvides(self, ISite, directlyProvidedBy(self))
-
-class Root(Folder):
- implements(IContainmentRoot)
-
-
-def Wrapper(ob, container):
- ob.__parent__ = container
- return ob
-
-class Test(TestCase):
-
- def setUp(self):
- TestCase.setUp(self)
- root = Root()
-
- f1 = Wrapper(Folder(), root)
- sm1 = ServiceManager()
- f1.setSiteManager(sm1)
-
- f2 = Wrapper(Folder(), f1)
- sm2 = ServiceManager()
- f2.setSiteManager(sm2)
-
- self.root = root
- self.f1 = f1
- self.f2 = f2
- self.sm1 = sm1
- self.sm2 = sm2
-
- def test_getServiceManager(self):
-
- self.assertEqual(getServiceManager_hook(self.root), serviceManager)
- self.assertEqual(getServiceManager_hook(self.f1), self.sm1)
- self.assertEqual(getServiceManager_hook(self.f2), self.sm2)
-
- def test_getNextServiceManager(self):
-
- self.assertRaises(ComponentLookupError,
- getNextServiceManager, self.root)
-
- self.assertEqual(getNextServiceManager(Wrapper(self.sm1, self.f1)),
- serviceManager)
- self.assertEqual(getNextServiceManager(Wrapper(self.sm2, self.f2)),
- self.sm1)
-
- def test_getNextServiceManager_fails_w_bad_root(self):
- root = Folder()
- f1 = Wrapper(Folder(), root)
- sm1 = ServiceManager()
- f1.setSiteManager(sm1)
- self.assertRaises(TypeError,
- getNextServiceManager, Wrapper(sm1, f1)
- )
-
-def test_suite():
- return makeSuite(Test)
-
-if __name__=='__main__':
- main(defaultTest='test_suite')
More information about the Zope3-Checkins
mailing list