[Zope3-checkins] CVS: Zope3/lib/python/Zope/App/Event/tests - PlacelessSetup.py:1.1 testEventService.py:1.1 testLogger.py:1.1 testObjectEvent.py:1.1 testDirectives.py:1.2
Gary Poster
gary@zope.com
Sat, 21 Dec 2002 10:33:16 -0500
Update of /cvs-repository/Zope3/lib/python/Zope/App/Event/tests
In directory cvs.zope.org:/tmp/cvs-serv30423/lib/python/Zope/App/Event/tests
Modified Files:
testDirectives.py
Added Files:
PlacelessSetup.py testEventService.py testLogger.py
testObjectEvent.py
Log Message:
Moves ObjectEvent, GlobalEventService, and Logger from the Zope.Event package to the Zope.App.Event package.
Does *not* redesign/refactor LocalEventService to accept paths and hubids and reject non-wrapped objects. Maybe later.
This checkin may require you to hose your Data.fs.
For quadruple checking, I'll send a note to zope3-dev when I've confirmed that a fresh checkout works.
=== Added File Zope3/lib/python/Zope/App/Event/tests/PlacelessSetup.py ===
##############################################################################
#
# 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.
#
##############################################################################
"""Unit test logic for setting up and tearing down basic infrastructure
$Id: PlacelessSetup.py,v 1.1 2002/12/21 15:32:45 poster Exp $
"""
from Zope.ComponentArchitecture import getServiceManager
from Zope.Event.IEventService import IEventService
from Zope.App.Event.GlobalEventService import eventService
from Interface import Interface
events = []
class EventRecorderClass:
notify = events.append
EventRecorder = EventRecorderClass()
def getEvents(event_type = None, filter = None):
r = []
for event in events:
if event_type is not None and not event_type.isImplementedBy(event):
continue
if filter is not None and not filter(event):
continue
r.append(event)
return r
class PlacelessSetup:
def setUp(self):
sm=getServiceManager(None)
defineService=sm.defineService
provideService=sm.provideService
defineService("Events", IEventService)
provideService("Events", eventService)
del events[:]
eventService.globalSubscribe(EventRecorder)
=== Added File Zope3/lib/python/Zope/App/Event/tests/testEventService.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: testEventService.py,v 1.1 2002/12/21 15:32:45 poster Exp $
"""
import unittest, sys
from Zope.Testing.CleanUp import CleanUp # Base class w registry cleanup
from Zope.App.Event.IObjectEvent import IObjectEvent
from Zope.App.Event.IObjectEvent import IObjectAddedEvent
from Zope.App.Event.IObjectEvent import IObjectRemovedEvent
from Zope.App.Event.IObjectEvent import IObjectModifiedEvent
from Zope.App.Event.ObjectEvent import ObjectAddedEvent, ObjectModifiedEvent
from Zope.App.Event.GlobalEventService import GlobalEventService
from Zope.Exceptions import NotFoundError
from Zope.Event.IEvent import IEvent
from Zope.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 = GlobalEventService()
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/lib/python/Zope/App/Event/tests/testLogger.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.
#
##############################################################################
"""Test suite for Zope.Event.Logger.
$Id: testLogger.py,v 1.1 2002/12/21 15:32:45 poster Exp $
"""
import unittest
import logging
from Zope.ComponentArchitecture.tests.PlacelessSetup import PlacelessSetup
from Zope.ComponentArchitecture import getServiceManager, getService
from Zope.App.Event import globalSubscribe
from Zope.Event import unsubscribe, publish
from Zope.App.Event.ObjectEvent import ObjectAddedEvent
from Zope.App.Event.Logger import Logger
from Zope.App.Event.GlobalEventService import GlobalEventService
class DopeyHandler(logging.Handler):
def __init__(self):
logging.Handler.__init__(self)
self.results = []
def emit(self, record):
self.results.append(record)
class TestLogger1(PlacelessSetup,unittest.TestCase):
eventlogger = Logger()
def setUp(self):
PlacelessSetup.setUp(self)
from Zope.Event.IEventService import IEventService
getServiceManager(None).defineService("Events", IEventService)
from Zope.App.Event.GlobalEventService import eventService
getServiceManager(None).provideService("Events", eventService)
# futz a handler in for testing
self.logger = logging.getLogger("Event.Logger")
self.oldlevel = self.logger.level
self.oldprop = self.logger.propagate
self.logger.propagate = False
self.logger.setLevel(logging.DEBUG)
self.handler = DopeyHandler()
self.logger.addHandler(self.handler)
# register a logger
globalSubscribe(self.eventlogger)
# send an event
publish(None, ObjectAddedEvent(None, 'foo'))
def tearDown(self):
unsubscribe(self.eventlogger)
self.logger.removeHandler(self.handler)
self.logger.setLevel(self.oldlevel)
self.logger.propagate = self.oldprop
PlacelessSetup.tearDown(self)
def testLogger(self):
# Test the logger logs appropriately
results = self.handler.results
self.assertEqual(len(results), 1)
result = results[0]
self.assertEqual(result.name, "Event.Logger")
self.assertEqual(result.levelno, logging.INFO)
self.assertEqual(result.getMessage(),
"Zope.App.Event.ObjectEvent.ObjectAddedEvent: "
"XXX detail temporarily disabled\n")
self.assertEqual(result.exc_info, None)
class TestLogger2(TestLogger1):
eventlogger = Logger(logging.CRITICAL)
def testLogger(self):
# Test the logger logs appropriately
results = self.handler.results
self.assertEqual(len(results), 1)
result = results[0]
self.assertEqual(result.name, "Event.Logger")
self.assertEqual(result.levelno, logging.CRITICAL)
self.assertEqual(result.getMessage(),
"Zope.App.Event.ObjectEvent.ObjectAddedEvent: "
"XXX detail temporarily disabled\n")
self.assertEqual(result.exc_info, None)
def test_suite():
return unittest.TestSuite([
unittest.makeSuite(TestLogger1),
unittest.makeSuite(TestLogger2),
])
if __name__=='__main__':
unittest.main(defaultTest='test_suite')
=== Added File Zope3/lib/python/Zope/App/Event/tests/testObjectEvent.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: testObjectEvent.py,v 1.1 2002/12/21 15:32:45 poster Exp $
"""
import unittest, sys
from Zope.App.Event.ObjectEvent import ObjectAddedEvent
from Zope.App.Event.ObjectEvent import ObjectModifiedEvent
from Zope.App.Event.ObjectEvent import ObjectRemovedEvent
from Zope.App.Event.ObjectEvent import ObjectMovedEvent
from Zope.App.Event.ObjectEvent import ObjectAnnotationsModifiedEvent
from Zope.App.Event.ObjectEvent import ObjectContentModifiedEvent
class TestObjectAddedEvent(unittest.TestCase):
location = '/some/location'
object = object()
klass = ObjectAddedEvent
def setUp(self):
self.event = self.klass(self.object, self.location)
def testGetLocation(self):
self.assertEqual(self.event.location, self.location)
def testGetObject(self):
self.assertEqual(self.event.object, self.object)
class TestObjectModifiedEvent(TestObjectAddedEvent):
klass = ObjectModifiedEvent
class TestObjectAnnotationsModifiedEvent(TestObjectAddedEvent):
klass = ObjectAnnotationsModifiedEvent
class TestObjectContentModifiedEvent(TestObjectAddedEvent):
klass = ObjectContentModifiedEvent
class TestObjectRemovedEvent(TestObjectAddedEvent):
location = '/some/location'
def setUp(self):
self.event = ObjectRemovedEvent(self.object, self.location)
def testGetLocation(self):
self.assertEqual(self.event.location, self.location)
def testGetObject(self):
self.assertEqual(self.event.object, self.object)
class TestObjectMovedEvent(TestObjectAddedEvent):
from_location = '/some/other/location'
def setUp(self):
self.event = ObjectMovedEvent(self.object,
self.from_location, self.location)
def testFromLocation(self):
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)))
if __name__=='__main__':
unittest.main(defaultTest='test_suite')
=== Zope3/lib/python/Zope/App/Event/tests/testDirectives.py 1.1 => 1.2 ===
--- Zope3/lib/python/Zope/App/Event/tests/testDirectives.py:1.1 Wed Dec 18 18:36:58 2002
+++ Zope3/lib/python/Zope/App/Event/tests/testDirectives.py Sat Dec 21 10:32:45 2002
@@ -24,10 +24,10 @@
from Zope.Exceptions import NotFoundError
from Zope.Event import subscribe, unsubscribe, publish
-from Zope.Event.ObjectEvent import ObjectAddedEvent
-from Zope.Event.ObjectEvent import ObjectRemovedEvent
-from Zope.Event.ObjectEvent import ObjectModifiedEvent
-from Zope.Event.tests.testEventService \
+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.testEventService \
import DummySubscriber, DummyFilter, DummyEvent
from Zope.ComponentArchitecture.tests.PlacelessSetup import PlacelessSetup
from Zope.ComponentArchitecture import getServiceManager, getService
@@ -40,7 +40,7 @@
PlacelessSetup.setUp(self)
from Zope.Event.IEventService import IEventService
getServiceManager(None).defineService("Events", IEventService)
- from Zope.Event.GlobalEventService import eventService
+ from Zope.App.Event.GlobalEventService import eventService
getServiceManager(None).provideService("Events", eventService)
def testSubscribe(self):
@@ -56,8 +56,8 @@
'''<test:subscribe
subscriber="Zope.Event.tests.subscriber.subscriber"
event_types=
- "Zope.Event.IObjectEvent.IObjectAddedEvent
- Zope.Event.IObjectEvent.IObjectRemovedEvent"
+ "Zope.App.Event.IObjectEvent.IObjectAddedEvent
+ Zope.App.Event.IObjectEvent.IObjectRemovedEvent"
filter="Zope.Event.tests.subscriber.filter" />'''
))