[Zope-Checkins] CVS: Zope3/lib/python/Zope/Event/tests - testEventService.py:1.1.2.1

[Chris Withers]

Update of /cvs-repository/Zope3/lib/python/Zope/Event/tests
In directory cvs.zope.org:/tmp/cvs-serv32429/tests

Added Files:
      Tag: Zope-3x-branch
	testEventService.py 
Log Message:
JF/CW -  It doesn't pass the tests (quite spectacularly ;-) but it smells okay.

=== Added File Zope3/lib/python/Zope/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.2.1 2002/02/22 21:26:15 chrisw Exp $
"""

import unittest, sys

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
from Zope.Event.EventService import EventService
from Zope.Exceptions import NotFoundError

class DummySubscriber:

    def __init__(self):
        self.notified = 0
        
    def notify(self, event):
        self.notified += 1
    
class DummyFilter:
    
    def __init__(self,value=1):
        self.value = value
        
    def __call__(self, event):
        return self.value

class DummyEvent:

    __implements__ = IObjectAddedEvent, IObjectRemovedEvent

class ObjectEvent:

    __implements__ = IObjectEvent
    
class TestEventService(unittest.TestCase):
    
    def setUp(self):
        self.service = EventService()
        self.event = ObjectAddedEvent('/foo')
        self.subscriber = DummySubscriber()
        
    def testSubscribe1(self):
        "Test subscribe method with one parameter"
        self.service.subscribe(self.subscriber)
        self.service.publish(self.event)
        self.assertEqual(self.subscriber.notified, 1)
        
    def testSubscribe2(self):
        "Test subscribe method with two parameters"
        self.service.subscribe(
            self.subscriber,
            event_types=[IObjectAddedEvent]
            )
        self.service.publish(self.event)
        self.assertEqual(self.subscriber.notified, 1)

    def testSubscribe3(self):
        "Test subscribe method with three parameters"
        self.service.subscribe(
            self.subscriber,
            event_types=[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.subscribe(
            self.subscriber,
            event_types=[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.subscribe(
            self.subscriber,
            event_types=[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.subscribe(
            self.subscriber,
            event_types=[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.subscribe(
            self.subscriber,
            event_types=[IObjectModifiedEvent]
            )
        self.service.subscribe(
            self.subscriber,
            event_types=[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.subscribe(
            self.subscriber,
            event_types=[IObjectAddedEvent],
            filter=DummyFilter()
            )
        self.service.subscribe(
            self.subscriber,
            event_types=[IObjectAddedEvent],
            filter=DummyFilter()
            )
        self.service.subscribe(
            self.subscriber,
            event_types=[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.subscribe(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)

    def testPublish1(self):
        "Test publish method"
        subscriber = self.subscriber
        self.service.subscribe(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.subscribe(
            self.subscriber,
            event_types=[IObjectEvent],
            )
        self.service.subscribe(
            self.subscriber,
            event_types=[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.subscribe(
            self.subscriber,
            event_types=[IObjectRemovedEvent,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.subscribe(
            self.subscriber,
            event_types=[IObjectEvent],
            )
        self.service.subscribe(
            self.subscriber,
            event_types=[IObjectAddedEvent],
            )
        self.service.publish(ObjectEvent())
        self.assertEqual(self.subscriber.notified, 1)        

def test_suite():
    return unittest.TestSuite((
        unittest.makeSuite(TestEventService),
        ))

if __name__=='__main__':
    unittest.main(defaultTest='test_suite')