[Zope-Checkins] CVS: Zope3/lib/python/Zope/App/OFS/Services/LocalEventService/tests - EventSetup.py:1.2 __init__.py:1.2 testEventService.py:1.2

Jim Fulton jim@zope.com
Mon, 10 Jun 2002 19:28:42 -0400


Update of /cvs-repository/Zope3/lib/python/Zope/App/OFS/Services/LocalEventService/tests
In directory cvs.zope.org:/tmp/cvs-serv17445/lib/python/Zope/App/OFS/Services/LocalEventService/tests

Added Files:
	EventSetup.py __init__.py testEventService.py 
Log Message:
Merged Zope-3x-branch into newly forked Zope3 CVS Tree.


=== Zope3/lib/python/Zope/App/OFS/Services/LocalEventService/tests/EventSetup.py 1.1 => 1.2 ===
+#
+# 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.
+# 
+##############################################################################
+"""
+
+Revision information:
+$Id$
+"""
+from Zope.App.OFS.Services.ServiceManager.tests.PlacefulSetup \
+  import PlacefulSetup
+from Zope.ComponentArchitecture import getService, getServiceManager
+from Zope.App.Traversing.Traverser import Traverser
+from Zope.App.Traversing.ITraverser import ITraverser
+from Zope.App.Traversing.DefaultTraversable import DefaultTraversable
+from Zope.App.Traversing.ITraversable import ITraversable
+from Zope.App.OFS.Container.ContainerTraversable import ContainerTraversable
+from Zope.App.OFS.Container.IContainer import IReadContainer
+from Zope.App.OFS.Services.LocalEventService.LocalEventService import LocalEventService
+
+class EventSetup(PlacefulSetup):
+    
+    def setUp(self):
+        PlacefulSetup.setUp(self)
+        self.buildFolders()
+        adapterService=getService(None, "Adapters")
+        adapterService.provideAdapter(
+            None, ITraverser, Traverser)
+        adapterService.provideAdapter(
+            None, ITraversable, DefaultTraversable)
+        adapterService.provideAdapter(
+            IReadContainer, ITraversable, ContainerTraversable)
+        from Zope.Event.IEventService import IEventService
+        from Zope.Event.GlobalEventService import eventService
+        globsm=getServiceManager(None)
+        globsm.defineService("Events", IEventService)
+        globsm.provideService("Events", eventService)
+        self.createEventService()
+    
+    def createEventService(self, folder=None):
+        if folder is None: folder=self.rootFolder
+        if not folder.hasServiceManager():
+            self.createServiceManager(folder)
+        sm=getServiceManager(folder) # wrapped now
+        sm.setObject("myEventService",LocalEventService())
+        sm.bindService("Events","myEventService")
+        
+    


=== Zope3/lib/python/Zope/App/OFS/Services/LocalEventService/tests/__init__.py 1.1 => 1.2 ===
+#
+# 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.
+# 
+##############################################################################
+"""Local Event Service"""
+
+


=== Zope3/lib/python/Zope/App/OFS/Services/LocalEventService/tests/testEventService.py 1.1 => 1.2 ===
+#
+# 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.
+# 
+##############################################################################
+"""
+
+Revision information:
+$Id$
+"""
+
+from unittest import TestCase, TestLoader, TextTestRunner
+from EventSetup import EventSetup
+from Zope.ComponentArchitecture import getService, getServiceManager
+from Zope.App.OFS.Services.LocalEventService.LocalEventService import \
+  LocalEventService
+from Zope.Event.tests.subscriber import DummySubscriber, DummyFilter
+from Zope.Event.ObjectEvent import ObjectAddedEvent
+from Zope.Event import subscribe, unsubscribe, listSubscriptions, publishEvent
+
+from Zope.Event.IObjectEvent import IObjectEvent
+from Zope.Event.IObjectEvent import IObjectAddedEvent
+from Zope.Event.IObjectEvent import IObjectRemovedEvent
+from Zope.Event.IObjectEvent import IObjectModifiedEvent
+from Zope.Event.ObjectEvent import ObjectAddedEvent, ObjectModifiedEvent
+from Zope.Event.GlobalEventService import GlobalEventService
+from Zope.Exceptions import NotFoundError
+from Zope.Event.IEvent import IEvent
+from Zope.Proxy.ContextWrapper import ContextWrapper
+
+from Zope.Event.tests.subscriber import DummySubscriber, DummyFilter
+
+class UnpromotingLocalEventService(LocalEventService):
+    
+    def isPromotableEvent(self, event):
+        "see ILocalEventService"
+        return 0
+
+class DummyEvent:
+
+    __implements__ = IObjectAddedEvent, IObjectRemovedEvent
+
+class ObjectEvent:
+
+    __implements__ = IObjectEvent
+
+class EventServiceTests(EventSetup, TestCase):
+    
+    def _createNestedServices(self):
+        self.createEventService(self.folder1)
+        self.createEventService(self.folder1_1)
+        self.createEventService(self.folder1_1_1)
+    
+    def _createSubscribers(self):
+        self.rootFolder.setObject("rootFolderSubscriber", DummySubscriber())
+        self.rootFolderSubscriber=ContextWrapper(
+            self.rootFolder["rootFolderSubscriber"],
+            self.rootFolder,
+            name="rootFolderSubscriber")
+        self.folder1.setObject("folder1Subscriber", DummySubscriber())
+        self.folder1Subscriber=ContextWrapper(
+            self.folder1["folder1Subscriber"],
+            self.folder1,
+            name="folder1Subscriber")
+        self.folder1_1.setObject("folder1_1Subscriber", DummySubscriber())
+        self.folder1_1Subscriber=ContextWrapper(
+            self.folder1_1["folder1_1Subscriber"],
+            self.folder1_1,
+            name="folder1_1Subscriber")
+        
+    def testCreateNestedServices(self):
+        self._createNestedServices()
+    
+    def testListSubscriptions1(self):
+        "a non-subscribed subscriber gets an empty array"
+        self._createSubscribers()
+        self.assertEqual([], self.rootFolder.getServiceManager().getService(
+            "Events").listSubscriptions(self.rootFolderSubscriber))
+    
+    def testListSubscriptions2(self):
+        "one subscription"
+        self._createSubscribers()
+        subscribe(
+            self.rootFolderSubscriber,
+            event_type=IObjectAddedEvent
+            )
+        self.assertEqual(
+            [(IObjectAddedEvent,None)],
+            self.rootFolder.getServiceManager().getService(
+            "Events").listSubscriptions(self.rootFolderSubscriber))
+    
+    def testListSubscriptions3(self):
+        "listing limited subscription"
+        self._createSubscribers()
+        subscribe(
+            self.rootFolderSubscriber,
+            event_type=IObjectAddedEvent
+            )
+        self.assertEqual(
+            [],
+            self.rootFolder.getServiceManager().getService(
+            "Events").listSubscriptions(self.rootFolderSubscriber, IObjectRemovedEvent))
+    
+    def testSubscribe1(self):
+        "Test subscribe method with one parameter"
+        self._createSubscribers()
+        subscribe(self.rootFolderSubscriber)
+        publishEvent(self.rootFolder, ObjectAddedEvent('/foo'))
+        self.assertEqual(self.rootFolderSubscriber.notified, 1)
+        
+    def testSubscribe2(self):
+        "Test subscribe method with two parameters"
+        self._createSubscribers()
+        subscribe(
+            self.rootFolderSubscriber,
+            event_type=IObjectAddedEvent
+            )
+        publishEvent(self.folder1_1_1, ObjectAddedEvent('/foo'))
+        self.assertEqual(self.rootFolderSubscriber.notified, 1)
+
+    def testSubscribe3(self):
+        "Test subscribe method with three parameters"
+        self._createSubscribers()
+        subscribe(
+            self.rootFolderSubscriber,
+            event_type=IObjectAddedEvent,
+            filter=DummyFilter()
+            )
+        publishEvent(self.folder1_1_1, ObjectAddedEvent('/foo'))
+        self.assertEqual(self.rootFolderSubscriber.notified, 1)
+
+    def testSubscribe4(self):
+        """Test subscribe method with three parameters
+        and an always failing filter.
+        """
+        self._createSubscribers()
+        subscribe(
+            self.rootFolderSubscriber,
+            event_type=IObjectAddedEvent,
+            filter=DummyFilter(0)
+            )
+        publishEvent(self.folder1_1_1, ObjectAddedEvent('/foo'))
+        self.assertEqual(self.rootFolderSubscriber.notified, 0)
+
+    def testSubscribe5(self):
+        """Test subscribe method with three parameters
+        and an irrelevent event type.
+        """
+        self._createSubscribers()
+        subscribe(
+            self.rootFolderSubscriber,
+            event_type=IObjectModifiedEvent,
+            filter=DummyFilter()
+            )
+        publishEvent(self.folder1_1_1, ObjectAddedEvent('/foo'))
+        self.assertEqual(self.rootFolderSubscriber.notified, 0)
+
+    def testSubscribe6(self):
+        """Test subscribe method where the event type
+        registered is a generalised interface of the
+        event passed to the 'publishEvent' method.
+        """
+        self._createSubscribers()
+        subscribe(
+            self.rootFolderSubscriber,
+            event_type=IObjectEvent
+            )
+        publishEvent(self.folder1_1_1, ObjectAddedEvent('/foo'))
+        self.assertEqual(self.rootFolderSubscriber.notified, 1)
+
+    def testSubscribe7(self):
+        """Test subscribe method where one of the
+        event types registered is not interested in
+        the publishEvented event.
+        """
+        self._createSubscribers()
+        subscribe(
+            self.rootFolderSubscriber,
+            event_type=IObjectModifiedEvent
+            )
+        subscribe(
+            self.rootFolderSubscriber,
+            event_type=IObjectAddedEvent
+            )
+        publishEvent(self.folder1_1_1, ObjectAddedEvent('/foo'))
+        self.assertEqual(self.rootFolderSubscriber.notified, 1)
+
+    def testSubscribe8(self):
+        """Test subscribe method where the same subscriber
+        subscribes multiple times. 
+        """
+        self._createSubscribers()
+        subscribe(
+            self.rootFolderSubscriber,
+            event_type=IObjectAddedEvent
+            )
+        subscribe(
+            self.rootFolderSubscriber,
+            event_type=IObjectAddedEvent
+            )
+        subscribe(
+            self.rootFolderSubscriber,
+            event_type=IObjectAddedEvent,
+            filter=DummyFilter(0)
+            )
+        publishEvent(self.folder1_1_1, ObjectAddedEvent('/foo'))
+        self.assertEqual(self.rootFolderSubscriber.notified, 2)
+
+    def testUnsubscribe1(self):
+        "Test unsubscribe method"
+        self._createSubscribers()
+        subscribe(
+            self.rootFolderSubscriber
+            )
+        publishEvent(self.folder1_1_1, ObjectAddedEvent('/foo'))
+        self.assertEqual(self.rootFolderSubscriber.notified, 1)
+        unsubscribe(
+            self.rootFolderSubscriber
+            )
+        publishEvent(self.folder1_1_1, ObjectAddedEvent('/foo'))
+        self.assertEqual(self.rootFolderSubscriber.notified, 1)
+
+    def testUnsubscribe2(self):
+        "Test unsubscribe of something that hasn't been subscribed"
+        self._createSubscribers()
+        self.assertRaises(NotFoundError,
+                          unsubscribe,
+                          self.rootFolderSubscriber,
+                          IObjectEvent)
+        self.assertEqual(None,
+                         unsubscribe(self.rootFolderSubscriber))
+    
+    def testUnsubscribe3(self):
+        "Test selective unsubscribe"
+        self._createSubscribers()
+        subscriber=self.rootFolderSubscriber
+        subscriber2=self.folder1Subscriber
+        filter=DummyFilter()
+        event=ObjectAddedEvent('/foo')
+        event2=ObjectModifiedEvent('/foo')
+        subscribe(
+            subscriber)
+        subscribe(
+            subscriber,
+            event_type=IObjectAddedEvent,
+            filter=filter
+            )
+        subscribe(
+            subscriber,
+            event_type=IObjectAddedEvent
+            )
+        subscribe(
+            subscriber2,
+            event_type=IObjectAddedEvent
+            )
+        publishEvent(self.folder1_1_1, event)
+        self.assertEqual(subscriber.notified, 3)
+        self.assertEqual(subscriber2.notified, 1)
+        publishEvent(self.folder1_1_1, event2)
+        self.assertEqual(subscriber.notified, 4)
+        self.assertEqual(subscriber2.notified, 1)
+        unsubscribe(subscriber, IObjectAddedEvent)
+        publishEvent(self.folder1_1_1, event)
+        self.assertEqual(subscriber.notified, 6)
+        self.assertEqual(subscriber2.notified, 2)
+        unsubscribe(subscriber, IEvent)
+        publishEvent(self.folder1_1_1, event2)
+        self.assertEqual(subscriber.notified, 6)
+        self.assertEqual(subscriber2.notified, 2)
+        publishEvent(self.folder1_1_1, event)
+        self.assertEqual(subscriber.notified, 7)
+        self.assertEqual(subscriber2.notified, 3)
+        self.assertRaises(NotFoundError, unsubscribe, subscriber, IObjectAddedEvent)
+        unsubscribe(subscriber, IObjectAddedEvent, filter)
+        publishEvent(self.folder1_1_1, event)
+        self.assertEqual(subscriber.notified, 7)
+        self.assertEqual(subscriber2.notified, 4)
+        unsubscribe(subscriber2, IObjectAddedEvent)
+        publishEvent(self.folder1_1_1, event)
+        self.assertEqual(subscriber.notified, 7)
+        self.assertEqual(subscriber2.notified, 4)
+    
+    def testUnsubscribe4(self):
+        "Test selective unsubscribe with nested services"
+        self._createNestedServices()
+        self.testUnsubscribe3()
+
+    def testpublishEvent1(self):
+        "Test publishEvent method"
+        self._createSubscribers()
+        subscriber = self.rootFolderSubscriber
+        subscribe(subscriber)
+        self.assertEqual(subscriber.notified, 0)
+        publishEvent(self.folder1_1_1, ObjectAddedEvent('/foo'))
+        self.assertEqual(subscriber.notified, 1)
+
+    def testpublishEvent2(self):
+        """Test publishEvent method where subscriber has been
+        subscribed twice, with a more generalised
+        version of the initially subscribed interface
+        in the second subscription.
+        """
+        self._createSubscribers()
+        subscriber = self.rootFolderSubscriber
+        subscribe(
+            subscriber,
+            event_type=IObjectEvent,
+            )
+        subscribe(
+            subscriber,
+            event_type=IObjectAddedEvent,
+            )
+        publishEvent(self.folder1_1_1, ObjectAddedEvent('/foo'))
+        self.assertEqual(subscriber.notified, 2)
+
+    def testpublishEvent3(self):
+        """Test publishEvent method where subscriber has been
+        to two interfaces and a single event implements both
+        of those interfaces.
+        """
+        self._createSubscribers()
+        subscriber = self.rootFolderSubscriber
+        subscribe(
+            subscriber,
+            event_type=IObjectRemovedEvent
+            )
+        subscribe(
+            subscriber,
+            event_type=IObjectAddedEvent
+            )
+        publishEvent(self.folder1_1_1, DummyEvent())
+        self.assertEqual(subscriber.notified, 2)
+
+    def testpublishEvent4(self):
+        """Test publishEvent method to make sure that we don't
+        'leak registrations up' sez Jim.
+        """
+        self._createSubscribers()
+        subscriber = self.rootFolderSubscriber
+        subscribe(
+            subscriber,
+            event_type=IObjectEvent
+            )
+        subscribe(
+            subscriber,
+            event_type=IObjectAddedEvent
+            )
+        publishEvent(self.folder1_1_1, ObjectEvent())
+        self.assertEqual(subscriber.notified, 1)
+    
+    def _createAlternateService(self, service):
+        self._createSubscribers()
+        self.folder2.setObject("folder2Subscriber", DummySubscriber())
+        self.folder2Subscriber=ContextWrapper(
+            self.folder2["folder2Subscriber"],
+            self.folder2,
+            name="folder2Subscriber")
+        if not self.folder2.hasServiceManager():
+            self.createServiceManager(self.folder2)
+        sm=getServiceManager(self.folder2) # wrapped now
+        sm.setObject("myEventService",service) # this is the change
+        sm.bindService("Events","myEventService")
+        subscribe(
+            self.rootFolderSubscriber,
+            event_type=IObjectAddedEvent
+            )
+        subscribe(
+            self.folder2Subscriber,
+            event_type=IObjectAddedEvent
+            )
+    
+    def testNonPromotingEventService1(self):
+        """test to see if events are not passed on to a parent event
+        service with the appropriate isPromotableEvent setting"""
+        self._createAlternateService(UnpromotingLocalEventService())
+        publishEvent(self.folder2, ObjectAddedEvent('/foo'))
+        self.assertEqual(self.folder2Subscriber.notified, 1)
+        self.assertEqual(self.rootFolderSubscriber.notified, 0)
+    
+    def testPromotingEventService1(self):
+        """test to see if events are passed on to a parent event
+        service with the appropriate isPromotableEvent setting"""
+        self._createAlternateService(LocalEventService())
+        publishEvent(self.folder2, ObjectAddedEvent('/foo'))
+        self.assertEqual(self.folder2Subscriber.notified, 1)
+        self.assertEqual(self.rootFolderSubscriber.notified, 1)
+    
+    def testUnbindingResubscribing1(self):
+        """an event service is unbound; a lower event service should
+        then rebind to upper event service"""
+        self._createNestedServices()
+        self._createSubscribers()
+        subscribe(
+            self.rootFolderSubscriber,
+            event_type=IObjectAddedEvent
+            )
+        subscribe(
+            self.folder1Subscriber,
+            event_type=IObjectAddedEvent
+            )
+        subscribe(
+            self.folder1_1Subscriber,
+            event_type=IObjectAddedEvent
+            )
+        sm=getServiceManager(self.folder1)
+        sm.unbindService("Events")
+        publishEvent(self.rootFolder, ObjectAddedEvent('/foo'))
+        self.assertEqual(self.folder1Subscriber.notified, 0)
+        self.assertEqual(self.folder1_1Subscriber.notified, 1)
+    
+    def testNoSubscribeOnBind(self):
+        """if subscribeOnBind is 0, service should not subscribe to
+        parent"""
+        sv=LocalEventService()
+        sv.subscribeOnBind=0
+        self._createAlternateService(sv)
+        publishEvent(self.rootFolder, ObjectAddedEvent('/foo'))
+        self.assertEqual(self.folder2Subscriber.notified, 0)
+        self.assertEqual(self.rootFolderSubscriber.notified, 1)
+        sm=getServiceManager(self.folder2)
+        sm.unbindService("Events") # make sure it doesn't raise any errors
+        
+
+        
+def test_suite():
+    loader=TestLoader()
+    return loader.loadTestsFromTestCase(EventServiceTests)
+
+if __name__=='__main__':
+    TextTestRunner().run(test_suite())