[Zope3-checkins] CVS: Zope3/src/zope/app/event/tests - subscriber.py:1.1 test_eventpublisher.py:1.1 test_globaleventchannel.py:1.1 placelesssetup.py:1.3 test_directives.py:1.3 test_logger.py:1.3 test_objectevent.py:1.3 test_eventservice.py:NONE
Steve Alexander
steve@cat-box.net
Mon, 30 Dec 2002 09:03:37 -0500
Update of /cvs-repository/Zope3/src/zope/app/event/tests
In directory cvs.zope.org:/tmp/cvs-serv23914/src/zope/app/event/tests
Modified Files:
placelesssetup.py test_directives.py test_logger.py
test_objectevent.py
Added Files:
subscriber.py test_eventpublisher.py
test_globaleventchannel.py
Removed Files:
test_eventservice.py
Log Message:
Large refactoring of the event service.
=== Added File Zope3/src/zope/app/event/tests/subscriber.py ===
##############################################################################
#
# 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.
#
##############################################################################
"""
This contains some dummy stuff to do with subscribing to event channels
that's useful in several test modules.
Revision information:
$Id: subscriber.py,v 1.1 2002/12/30 14:03:05 stevea Exp $
"""
from zope.app.interfaces.event import IFilter, ISubscriber
class DummySubscriber:
__implements__ = ISubscriber
def __init__(self):
self.notified = 0
def notify(self, event):
self.notified += 1
subscriber = DummySubscriber()
class DummyFilter:
__implements__ = IFilter
def __init__(self,value=1):
self.value = value
def __call__(self, event):
return self.value
filter = DummyFilter
=== Added File Zope3/src/zope/app/event/tests/test_eventpublisher.py ===
##############################################################################
#
# 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: test_eventpublisher.py,v 1.1 2002/12/30 14:03:05 stevea Exp $
"""
import unittest, sys
from zope.testing.cleanup import CleanUp # Base class w registry cleanup
from zope.app.interfaces.event import IObjectEvent, IObjectAddedEvent
from zope.app.interfaces.event import IObjectRemovedEvent
from zope.app.interfaces.event import IObjectModifiedEvent
from zope.app.event.objectevent import ObjectAddedEvent, ObjectModifiedEvent
from zope.app.event.globalservice import GlobalEventPublisher
from zope.exceptions import NotFoundError
from zope.app.interfaces.event import IEvent
from zope.app.event.tests.subscriber import DummySubscriber, DummyFilter
class DummyEvent:
__implements__ = IObjectAddedEvent, IObjectRemovedEvent
class ObjectEvent:
__implements__ = IObjectEvent
class TestEventService(CleanUp, unittest.TestCase):
def setUp(self):
CleanUp.setUp(self)
self.service = GlobalEventPublisher()
self.event = ObjectAddedEvent(None, '/foo')
self.subscriber = DummySubscriber()
def testSubscribe1(self):
# Test subscribe method with one parameter
self.service.globalSubscribe(self.subscriber)
self.service.publish(self.event)
self.assertEqual(self.subscriber.notified, 1)
def testSubscribe2(self):
# Test subscribe method with two parameters
self.service.globalSubscribe(
self.subscriber,
event_type=IObjectAddedEvent
)
self.service.publish(self.event)
self.assertEqual(self.subscriber.notified, 1)
def testSubscribe3(self):
# Test subscribe method with three parameters
self.service.globalSubscribe(
self.subscriber,
event_type=IObjectAddedEvent,
filter=DummyFilter()
)
self.service.publish(self.event)
self.assertEqual(self.subscriber.notified, 1)
def testSubscribe4(self):
# Test subscribe method with three parameters and an always failing
# filter.
self.service.globalSubscribe(
self.subscriber,
event_type=IObjectAddedEvent,
filter=DummyFilter(0)
)
self.service.publish(self.event)
self.assertEqual(self.subscriber.notified, 0)
def testSubscribe5(self):
# Test subscribe method with three parameters and an irrelevent event
# type.
self.service.globalSubscribe(
self.subscriber,
event_type=IObjectModifiedEvent,
filter=DummyFilter()
)
self.service.publish(self.event)
self.assertEqual(self.subscriber.notified, 0)
def testSubscribe6(self):
# Test subscribe method where the event type registered is a
# generalised interface of the event passed to the 'publish' method.
self.service.globalSubscribe(
self.subscriber,
event_type=IObjectEvent
)
self.service.publish(self.event)
self.assertEqual(self.subscriber.notified, 1)
def testSubscribe7(self):
# Test subscribe method where one of the event types registered is not
# interested in the published event.
self.service.globalSubscribe(
self.subscriber,
event_type=IObjectModifiedEvent
)
self.service.globalSubscribe(
self.subscriber,
event_type=IObjectAddedEvent
)
self.service.publish(self.event)
self.assertEqual(self.subscriber.notified, 1)
def testSubscribe8(self):
# Test subscribe method where the same subscriber subscribes multiple
# times.
self.service.globalSubscribe(
self.subscriber,
event_type=IObjectAddedEvent,
filter=DummyFilter()
)
self.service.globalSubscribe(
self.subscriber,
event_type=IObjectAddedEvent,
filter=DummyFilter()
)
self.service.globalSubscribe(
self.subscriber,
event_type=IObjectAddedEvent,
filter=DummyFilter(0)
)
self.service.publish(self.event)
self.assertEqual(self.subscriber.notified, 2)
def testUnsubscribe1(self):
# Test unsubscribe method
subscriber = self.subscriber
self.service.globalSubscribe(subscriber)
self.service.publish(self.event)
self.assertEqual(self.subscriber.notified, 1)
self.service.unsubscribe(subscriber)
self.service.publish(self.event)
self.assertEqual(self.subscriber.notified, 1)
def testUnsubscribe2(self):
# Test unsubscribe of something that hasn't been subscribed
subscriber = self.subscriber
self.assertRaises(NotFoundError,
self.service.unsubscribe,
subscriber, IObjectEvent)
self.assertEqual(None,
self.service.unsubscribe(subscriber))
def testUnsubscribe3(self):
# Test selective unsubscribe
subscriber2=DummySubscriber()
filter=DummyFilter()
event2=ObjectModifiedEvent(None, '/foo')
self.service.globalSubscribe(
self.subscriber)
self.service.globalSubscribe(
self.subscriber,
event_type=IObjectAddedEvent,
filter=filter
)
self.service.globalSubscribe(
self.subscriber,
event_type=IObjectAddedEvent
)
self.service.globalSubscribe(
subscriber2,
event_type=IObjectAddedEvent
)
self.service.publish(self.event)
self.assertEqual(self.subscriber.notified, 3)
self.assertEqual(subscriber2.notified, 1)
self.service.publish(event2)
self.assertEqual(self.subscriber.notified, 4)
self.assertEqual(subscriber2.notified, 1)
self.service.unsubscribe(self.subscriber, IObjectAddedEvent)
self.service.publish(self.event)
self.assertEqual(self.subscriber.notified, 6)
self.assertEqual(subscriber2.notified, 2)
self.service.unsubscribe(self.subscriber, IEvent)
self.service.publish(event2)
self.assertEqual(self.subscriber.notified, 6)
self.assertEqual(subscriber2.notified, 2)
self.assertRaises(NotFoundError, self.service.unsubscribe,
self.subscriber, IObjectAddedEvent)
self.service.unsubscribe(self.subscriber, IObjectAddedEvent, filter)
self.service.publish(self.event)
self.assertEqual(self.subscriber.notified, 6)
self.assertEqual(subscriber2.notified, 3)
self.service.unsubscribe(subscriber2, IObjectAddedEvent)
self.service.publish(self.event)
self.assertEqual(self.subscriber.notified, 6)
self.assertEqual(subscriber2.notified, 3)
def testpublish1(self):
# Test publish method
subscriber = self.subscriber
self.service.globalSubscribe(subscriber)
self.assertEqual(self.subscriber.notified, 0)
self.service.publish(self.event)
self.assertEqual(self.subscriber.notified, 1)
def testpublish2(self):
# Test publish method where subscriber has been subscribed twice, with
# a more generalised version of the initially subscribed interface in
# the second subscription.
subscriber = self.subscriber
self.service.globalSubscribe(
self.subscriber,
event_type=IObjectEvent,
)
self.service.globalSubscribe(
self.subscriber,
event_type=IObjectAddedEvent,
)
self.service.publish(self.event)
self.assertEqual(self.subscriber.notified, 2)
def testpublish3(self):
# Test publish method where subscriber has been to two interfaces and
# a single event implements both of those interfaces.
subscriber = self.subscriber
self.service.globalSubscribe(
self.subscriber,
event_type=IObjectRemovedEvent
)
self.service.globalSubscribe(
self.subscriber,
event_type=IObjectAddedEvent
)
self.service.publish(DummyEvent())
self.assertEqual(self.subscriber.notified, 2)
def testpublish4(self):
# Test publish method to make sure that we don't 'leak registrations
# up' sez Jim.
subscriber = self.subscriber
self.service.globalSubscribe(
self.subscriber,
event_type=IObjectEvent
)
self.service.globalSubscribe(
self.subscriber,
event_type=IObjectAddedEvent
)
self.service.publish(ObjectEvent())
self.assertEqual(self.subscriber.notified, 1)
def testListSubscriptions1(self):
# a non-subscribed subscriber gets an empty array
self.assertEqual([], self.service.listSubscriptions(self.subscriber))
def testListSubscriptions2(self):
# one subscription
self.service.globalSubscribe(
self.subscriber, event_type=IObjectAddedEvent)
self.assertEqual([(IObjectAddedEvent, None)],
self.service.listSubscriptions(self.subscriber))
def testListSubscriptions3(self):
# listing limited subscription
self.service.globalSubscribe(
self.subscriber, event_type=IObjectAddedEvent)
L = self.service.listSubscriptions(self.subscriber,
IObjectRemovedEvent)
self.assertEqual([], L)
def test_suite():
return unittest.makeSuite(TestEventService)
if __name__ == '__main__':
unittest.main(defaultTest='test_suite')
=== Added File Zope3/src/zope/app/event/tests/test_globaleventchannel.py ===
##############################################################################
#
# 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.
#
##############################################################################
"""A functional GlobalEventChannel test.
$Id: test_globaleventchannel.py,v 1.1 2002/12/30 14:03:05 stevea Exp $
"""
from unittest import TestCase, TestSuite, main, makeSuite
from zope.app.interfaces.event import IEvent
class ISomeEvent(IEvent):
pass
class ISomeSubEvent(ISomeEvent):
pass
class ISomeOtherEvent(IEvent):
pass
class SomeEvent:
__implements__ = ISomeEvent
class SomeSubEvent:
__implements__ = ISomeSubEvent
class SomeOtherEvent:
__implements__ = ISomeOtherEvent
class SubscriberStub:
received = None
def notify(self, event):
self.received = event
class Test(TestCase):
def test_notify(self):
from zope.app.event.globalservice import GlobalEventChannel
subscriber = SubscriberStub()
ec = GlobalEventChannel()
ec.globalSubscribe(subscriber, ISomeEvent)
ev = SomeEvent()
ec.notify(ev)
self.assertEquals(subscriber.received, ev,
"Did not get event registered for")
ev = SomeSubEvent()
ec.notify(ev)
self.assertEquals(
subscriber.received, ev, "Did not get subclassed event")
ev = SomeOtherEvent()
ec.notify(ev)
self.assertNotEquals(subscriber.received, ev, "Got unrelated event")
def test_notify_filter(self):
from zope.app.event.globalservice import GlobalEventChannel
true = lambda x: True
false = lambda x: False
subscriber = SubscriberStub()
ec = GlobalEventChannel()
ec.globalSubscribe(subscriber, ISomeEvent, true)
ev = SomeEvent()
ec.notify(ev)
self.assertEquals(subscriber.received, ev,
"Did not get event registered for")
subscriber = SubscriberStub()
ec = GlobalEventChannel()
ec.globalSubscribe(subscriber, ISomeEvent, false)
ev = SomeEvent()
ec.notify(ev)
self.assertEquals(subscriber.received, None,
"Event was not filtered")
def test_suite():
return TestSuite((
makeSuite(Test),
))
if __name__=='__main__':
main(defaultTest='test_suite')
=== Zope3/src/zope/app/event/tests/placelesssetup.py 1.2 => 1.3 ===
--- Zope3/src/zope/app/event/tests/placelesssetup.py:1.2 Wed Dec 25 09:12:51 2002
+++ Zope3/src/zope/app/event/tests/placelesssetup.py Mon Dec 30 09:03:05 2002
@@ -18,8 +18,8 @@
"""
from zope.component import getServiceManager
-from zope.interfaces.event import IEventService
-from zope.app.event.globaleventservice import eventService
+from zope.app.interfaces.event import IPublisher
+from zope.app.event.globalservice import eventPublisher
from zope.interface import Interface
events = []
@@ -29,7 +29,7 @@
EventRecorder = EventRecorderClass()
-def getEvents(event_type = None, filter = None):
+def getEvents(event_type=None, filter=None):
r = []
for event in events:
if event_type is not None and not event_type.isImplementedBy(event):
@@ -40,18 +40,15 @@
return r
-
-
class PlacelessSetup:
def setUp(self):
+ sm = getServiceManager(None)
+ defineService = sm.defineService
+ provideService = sm.provideService
- sm=getServiceManager(None)
- defineService=sm.defineService
- provideService=sm.provideService
-
- defineService("Events", IEventService)
- provideService("Events", eventService)
+ defineService("Events", IPublisher)
+ provideService("Events", eventPublisher)
del events[:]
- eventService.globalSubscribe(EventRecorder)
+ eventPublisher.globalSubscribe(EventRecorder)
=== Zope3/src/zope/app/event/tests/test_directives.py 1.2 => 1.3 ===
--- Zope3/src/zope/app/event/tests/test_directives.py:1.2 Wed Dec 25 09:12:51 2002
+++ Zope3/src/zope/app/event/tests/test_directives.py Mon Dec 30 09:03:05 2002
@@ -17,63 +17,65 @@
$Id$
"""
-from unittest import TestCase, TestSuite, main, makeSuite
-from StringIO import StringIO
+from unittest import TestCase, main, makeSuite
-from zope.configuration.xmlconfig import xmlconfig
+from zope.configuration.xmlconfig import xmlconfig, XMLConfig
+import zope.app.event
+from StringIO import StringIO
from zope.exceptions import NotFoundError
-from zope.event import subscribe, unsubscribe, publish
+from zope.app.event import globalUnsubscribe, publish
from zope.app.event.objectevent import ObjectAddedEvent
from zope.app.event.objectevent import ObjectRemovedEvent
from zope.app.event.objectevent import ObjectModifiedEvent
-from zope.app.event.tests.test_eventservice \
- import DummySubscriber, DummyFilter, DummyEvent
+from zope.app.event.tests.test_eventpublisher \
+ import DummyEvent
from zope.component.tests.placelesssetup import PlacelessSetup
-from zope.component import getServiceManager, getService
+from zope.component import getServiceManager
from zope.configuration.tests.basetestdirectivesxml import makeconfig
-
+from zope.app.interfaces.event import IEvent
class Test(PlacelessSetup, TestCase):
def setUp(self):
PlacelessSetup.setUp(self)
- from zope.interfaces.event import IEventService
- getServiceManager(None).defineService("Events", IEventService)
- from zope.app.event.globaleventservice import eventService
- getServiceManager(None).provideService("Events", eventService)
+ from zope.app.interfaces.event import IPublisher
+ getServiceManager(None).defineService("Events", IPublisher)
+ from zope.app.event.globalservice import eventPublisher
+ getServiceManager(None).provideService("Events", eventPublisher)
def testSubscribe(self):
- from zope.event.tests.subscriber import subscriber
+ from zope.app.event.tests.subscriber import subscriber
+ # This should do nothing silently, as the event_type is default
+ globalUnsubscribe(subscriber)
# This should fail, since we're not subscribed
- self.assertRaises(NotFoundError,unsubscribe,None,subscriber)
-
- xmlconfig(makeconfig(
- '''<directive
- name="subscribe"
- attributes="subscriber event_types filter"
- handler="zope.app.event.metaconfigure.subscribe" />''',
- '''<test:subscribe
- subscriber="zope.event.tests.subscriber.subscriber"
+ self.assertRaises(NotFoundError, globalUnsubscribe, subscriber, IEvent)
+ XMLConfig('meta.zcml', zope.app.event)()
+ xmlconfig(StringIO(
+ '''<zopeConfigure xmlns='http://namespaces.zope.org/zope'
+ xmlns:test='http://namespaces.zope.org/event'>
+ <test:subscribe
+ subscriber="zope.app.event.tests.subscriber.subscriber"
event_types=
"zope.app.interfaces.event.IObjectAddedEvent
zope.app.interfaces.event.IObjectRemovedEvent"
- filter="zope.event.tests.subscriber.filter" />'''
+ filter="zope.app.event.tests.subscriber.filter" />
+ </zopeConfigure>'''
))
- publish(None,ObjectAddedEvent(None, 'foo'))
- self.assertEqual(subscriber.notified,1)
- publish(None,ObjectRemovedEvent(object(), 'foo'))
- self.assertEqual(subscriber.notified,2)
- publish(None,ObjectModifiedEvent(None, 'foo'))
- self.assertEqual(subscriber.notified,2) # NB: no increase ;-)
- publish(None,DummyEvent())
- self.assertEqual(subscriber.notified,4) # NB: increased by 2 ;-)
+ publish(None, ObjectAddedEvent(None, 'foo'))
+ self.assertEqual(subscriber.notified, 1)
+ publish(None, ObjectRemovedEvent(object(), 'foo'))
+ self.assertEqual(subscriber.notified, 2)
+ publish(None, ObjectModifiedEvent(None, 'foo'))
+ self.assertEqual(subscriber.notified, 2) # NB: no increase ;-)
+ publish(None, DummyEvent())
+ self.assertEqual(subscriber.notified, 4) # NB: increased by 2 ;-)
- unsubscribe(subscriber)
+ globalUnsubscribe(subscriber)
def test_suite():
return makeSuite(Test)
-if __name__=='__main__':
+if __name__ == '__main__':
main(defaultTest='test_suite')
=== Zope3/src/zope/app/event/tests/test_logger.py 1.2 => 1.3 ===
--- Zope3/src/zope/app/event/tests/test_logger.py:1.2 Wed Dec 25 09:12:51 2002
+++ Zope3/src/zope/app/event/tests/test_logger.py Mon Dec 30 09:03:05 2002
@@ -11,8 +11,7 @@
# FOR A PARTICULAR PURPOSE.
#
##############################################################################
-"""Test suite for Zope.Event.Logger.
-
+"""
$Id$
"""
@@ -20,14 +19,11 @@
import logging
from zope.component.tests.placelesssetup import PlacelessSetup
-from zope.component import getServiceManager, getService
+from zope.component import getServiceManager
-from zope.app.event import globalSubscribe
-from zope.event import unsubscribe, publish
+from zope.app.event import globalSubscribe, globalUnsubscribe, publish
from zope.app.event.objectevent import ObjectAddedEvent
-from zope.app.event.logger import Logger
-
-from zope.app.event.globaleventservice import GlobalEventService
+from zope.app.event.globalservice import Logger
class DopeyHandler(logging.Handler):
@@ -44,10 +40,10 @@
def setUp(self):
PlacelessSetup.setUp(self)
- from zope.interfaces.event import IEventService
- getServiceManager(None).defineService("Events", IEventService)
- from zope.app.event.globaleventservice import eventService
- getServiceManager(None).provideService("Events", eventService)
+ from zope.app.interfaces.event import IPublisher
+ getServiceManager(None).defineService("Events", IPublisher)
+ from zope.app.event.globalservice import eventPublisher
+ getServiceManager(None).provideService("Events", eventPublisher)
# futz a handler in for testing
self.logger = logging.getLogger("Event.Logger")
self.oldlevel = self.logger.level
@@ -62,7 +58,7 @@
publish(None, ObjectAddedEvent(None, 'foo'))
def tearDown(self):
- unsubscribe(self.eventlogger)
+ globalUnsubscribe(self.eventlogger)
self.logger.removeHandler(self.handler)
self.logger.setLevel(self.oldlevel)
self.logger.propagate = self.oldprop
@@ -102,5 +98,5 @@
unittest.makeSuite(TestLogger2),
])
-if __name__=='__main__':
+if __name__ == '__main__':
unittest.main(defaultTest='test_suite')
=== Zope3/src/zope/app/event/tests/test_objectevent.py 1.2 => 1.3 ===
--- Zope3/src/zope/app/event/tests/test_objectevent.py:1.2 Wed Dec 25 09:12:51 2002
+++ Zope3/src/zope/app/event/tests/test_objectevent.py Mon Dec 30 09:03:05 2002
@@ -17,7 +17,7 @@
$Id$
"""
-import unittest, sys
+import unittest
from zope.app.event.objectevent import ObjectAddedEvent
from zope.app.event.objectevent import ObjectModifiedEvent
@@ -80,12 +80,14 @@
self.assertEqual(self.event.fromLocation, self.from_location)
def test_suite():
- return unittest.TestSuite((unittest.makeSuite(TestObjectAddedEvent),
- unittest.makeSuite(TestObjectModifiedEvent),
- unittest.makeSuite(TestObjectAnnotationsModifiedEvent),
- unittest.makeSuite(TestObjectContentModifiedEvent),
- unittest.makeSuite(TestObjectRemovedEvent),
- unittest.makeSuite(TestObjectMovedEvent)))
+ return unittest.TestSuite((
+ unittest.makeSuite(TestObjectAddedEvent),
+ unittest.makeSuite(TestObjectModifiedEvent),
+ unittest.makeSuite(TestObjectAnnotationsModifiedEvent),
+ unittest.makeSuite(TestObjectContentModifiedEvent),
+ unittest.makeSuite(TestObjectRemovedEvent),
+ unittest.makeSuite(TestObjectMovedEvent)
+ ))
if __name__=='__main__':
unittest.main(defaultTest='test_suite')
=== Removed File Zope3/src/zope/app/event/tests/test_eventservice.py ===