[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')