[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