[Zope3-checkins] CVS: Zope3/src/zope/app/event/tests - __init__.py:1.1.2.1 placelesssetup.py:1.1.2.1 test_directives.py:1.1.2.1 test_eventservice.py:1.1.2.1 test_logger.py:1.1.2.1 test_objectevent.py:1.1.2.1
Jim Fulton
jim@zope.com
Mon, 23 Dec 2002 14:31:36 -0500
Update of /cvs-repository/Zope3/src/zope/app/event/tests
In directory cvs.zope.org:/tmp/cvs-serv19908/zope/app/event/tests
Added Files:
Tag: NameGeddon-branch
__init__.py placelesssetup.py test_directives.py
test_eventservice.py test_logger.py test_objectevent.py
Log Message:
Initial renaming before debugging
=== Added File Zope3/src/zope/app/event/tests/__init__.py ===
#
# This file is necessary to make this directory a package.
=== Added File Zope3/src/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.2.1 2002/12/23 19:31:34 jim Exp $
"""
from zope.component import getServiceManager
from zope.interfaces.event import IEventService
from zope.app.event.globaleventservice import eventService
from zope.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/src/zope/app/event/tests/test_directives.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_directives.py,v 1.1.2.1 2002/12/23 19:31:34 jim Exp $
"""
from unittest import TestCase, TestSuite, main, makeSuite
from StringIO import StringIO
from zope.configuration.xmlconfig import xmlconfig
from zope.exceptions import NotFoundError
from zope.event import subscribe, unsubscribe, 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.component.tests.placelesssetup import PlacelessSetup
from zope.component import getServiceManager, getService
from zope.configuration.tests.basetestdirectivesxml import makeconfig
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)
def testSubscribe(self):
from zope.event.tests.subscriber import 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"
event_types=
"Zope.App.Event.IObjectEvent.IObjectAddedEvent
Zope.App.Event.IObjectEvent.IObjectRemovedEvent"
filter="Zope.Event.tests.subscriber.filter" />'''
))
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)
def test_suite():
return makeSuite(Test)
if __name__=='__main__':
main(defaultTest='test_suite')
=== Added File Zope3/src/zope/app/event/tests/test_eventservice.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_eventservice.py,v 1.1.2.1 2002/12/23 19:31:34 jim Exp $
"""
import unittest, sys
from zope.testing.cleanup import CleanUp # Base class w registry cleanup
from zope.app.interfaces.event import IObjectEvent
from zope.app.interfaces.event import 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.globaleventservice import GlobalEventService
from zope.exceptions import NotFoundError
from zope.interfaces.event 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/src/zope/app/event/tests/test_logger.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: test_logger.py,v 1.1.2.1 2002/12/23 19:31:35 jim Exp $
"""
import unittest
import logging
from zope.component.tests.placelesssetup import PlacelessSetup
from zope.component 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.interfaces.event 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/src/zope/app/event/tests/test_objectevent.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_objectevent.py,v 1.1.2.1 2002/12/23 19:31:35 jim 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')