[Zope-Checkins] CVS: Zope3/lib/python/Zope/App/OFS/Services/LocalEventService/tests - testEventService.py:1.1.2.3
Gary Poster
garyposter@earthlink.net
Sat, 18 May 2002 22:24:22 -0400
Update of /cvs-repository/Zope3/lib/python/Zope/App/OFS/Services/LocalEventService/tests
In directory cvs.zope.org:/tmp/cvs-serv18331/lib/python/Zope/App/OFS/Services/LocalEventService/tests
Modified Files:
Tag: Zope-3x-branch
testEventService.py
Log Message:
More tests of the event service code--the majority of it is a translation of the global event service tests for the local event service, but there are a few new ones as well. I plan for a few more to come. A fix or two for errors that the tests found also included.
=== Zope3/lib/python/Zope/App/OFS/Services/LocalEventService/tests/testEventService.py 1.1.2.2 => 1.1.2.3 ===
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):
@@ -31,9 +58,356 @@
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.getObject("rootFolderSubscriber"),
+ self.rootFolder,
+ name="rootFolderSubscriber")
+ self.folder1.setObject("folder1Subscriber", DummySubscriber())
+ self.folder1Subscriber=ContextWrapper(
+ self.folder1.getObject("folder1Subscriber"),
+ self.folder1,
+ name="folder1Subscriber")
+ self.folder1_1.setObject("folder1_1Subscriber", DummySubscriber())
+ self.folder1_1Subscriber=ContextWrapper(
+ self.folder1_1.getObject("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 testNonPromotingEventService1(self):
+ """test to see if events are not passed on to a parent event
+ service with the appropriate isPromotableEvent setting"""
+ self._createSubscribers()
+ self.folder2.setObject("folder2Subscriber", DummySubscriber())
+ self.folder2Subscriber=ContextWrapper(
+ self.folder2.getObject("folder2Subscriber"),
+ self.folder2,
+ name="folder2Subscriber")
+ if not self.folder2.hasServiceManager():
+ self.createServiceManager(self.folder2)
+ sm=getServiceManager(self.folder2) # wrapped now
+ sm.setObject("myEventService",UnpromotingLocalEventService()) # this is the change
+ sm.bindService("Events","myEventService")
+ subscribe(
+ self.rootFolderSubscriber,
+ event_type=IObjectAddedEvent
+ )
+ subscribe(
+ self.folder2Subscriber,
+ event_type=IObjectAddedEvent
+ )
+ 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 not passed on to a parent event
+ service with the appropriate isPromotableEvent setting"""
+ self._createSubscribers()
+ self.folder2.setObject("folder2Subscriber", DummySubscriber())
+ self.folder2Subscriber=ContextWrapper(
+ self.folder2.getObject("folder2Subscriber"),
+ self.folder2,
+ name="folder2Subscriber")
+ if not self.folder2.hasServiceManager():
+ self.createServiceManager(self.folder2)
+ sm=getServiceManager(self.folder2) # wrapped now
+ sm.setObject("myEventService",LocalEventService()) # this is the change
+ sm.bindService("Events","myEventService")
+ subscribe(
+ self.rootFolderSubscriber,
+ event_type=IObjectAddedEvent
+ )
+ subscribe(
+ self.folder2Subscriber,
+ event_type=IObjectAddedEvent
+ )
+ publishEvent(self.folder2, ObjectAddedEvent('/foo'))
+ self.assertEqual(self.folder2Subscriber.notified, 1)
+ self.assertEqual(self.rootFolderSubscriber.notified, 1)
+
def test_suite():