[Zope-Checkins] CVS: Zope3/lib/python/Zope/Event/tests - __init__.py:1.2 subscriber.py:1.2 testDirectives.py:1.2 testEventService.py:1.2 testLogger.py:1.2 testObjectEvent.py:1.2

[Jim Fulton]

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

Added Files:
	__init__.py subscriber.py testDirectives.py 
	testEventService.py testLogger.py testObjectEvent.py 
Log Message:
Merged Zope-3x-branch into newly forked Zope3 CVS Tree.

=== Zope3/lib/python/Zope/Event/tests/__init__.py 1.1 => 1.2 ===
+#
+# 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$
+"""
+


=== Zope3/lib/python/Zope/Event/tests/subscriber.py 1.1 => 1.2 ===
+#
+# 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$
+"""
+
+
+class DummySubscriber:
+
+    def __init__(self):
+        self.notified = 0
+        
+    def notify(self, event):
+        self.notified += 1
+
+subscriber = DummySubscriber()
+
+class DummyFilter:
+    
+    def __init__(self,value=1):
+        self.value = value
+        
+    def __call__(self, event):
+        return self.value
+
+filter = DummyFilter


=== Zope3/lib/python/Zope/Event/tests/testDirectives.py 1.1 => 1.2 ===
+#
+# 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$
+"""
+
+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, publishEvent
+from Zope.Event.ObjectEvent import ObjectAddedEvent
+from Zope.Event.ObjectEvent import ObjectRemovedEvent
+from Zope.Event.ObjectEvent import ObjectModifiedEvent
+from testEventService import DummySubscriber, DummyFilter, DummyEvent
+from Zope.ComponentArchitecture.tests.PlacelessSetup import PlacelessSetup
+from Zope.ComponentArchitecture import getServiceManager, getService
+
+template = """<zopeConfigure
+  xmlns='http://namespaces.zope.org/zope'
+  xmlns:event='http://namespaces.zope.org/event'>
+  %s
+  </zopeConfigure>"""
+
+class Test(PlacelessSetup, TestCase):
+    
+    def setUp(self):
+        PlacelessSetup.setUp(self)
+        from Zope.Event.IEventService import IEventService
+        getServiceManager(None).defineService("Events", IEventService)
+        from Zope.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(StringIO(template % (
+            """
+            <directive name="subscribe"
+                       attributes="subscriber event_types filter"
+                       handler="Zope.Event.metaConfigure.subscribe"
+                       namespace="http://namespaces.zope.org/event"/>
+            <event:subscribe subscriber="Zope.Event.tests.subscriber.subscriber"
+                            
+ event_types="Zope.Event.IObjectEvent.IObjectAddedEvent
+                                         
+ Zope.Event.IObjectEvent.IObjectRemovedEvent"
+                             filter="Zope.Event.tests.subscriber.filter"/>
+            """)))
+
+        publishEvent(None,ObjectAddedEvent('foo'))
+        self.assertEqual(subscriber.notified,1)
+        publishEvent(None,ObjectRemovedEvent('foo', object()))
+        self.assertEqual(subscriber.notified,2)
+        publishEvent(None,ObjectModifiedEvent('foo'))
+        self.assertEqual(subscriber.notified,2) # NB: no increase ;-)
+        publishEvent(None,DummyEvent())
+        self.assertEqual(subscriber.notified,4) # NB: increased by 2 ;-)
+        
+        unsubscribe(subscriber)
+
+def test_suite():
+    return TestSuite((
+        makeSuite(Test),
+        ))
+
+if __name__=='__main__':
+    main(defaultTest='test_suite')


=== Zope3/lib/python/Zope/Event/tests/testEventService.py 1.1 => 1.2 ===
+#
+# 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$
+"""
+
+import unittest, sys
+from Zope.Testing.CleanUp import CleanUp # Base class w registry cleanup
+
+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, ObjectModifiedEvent
+from Zope.Event.GlobalEventService import GlobalEventService
+from Zope.Exceptions import NotFoundError
+from Zope.Event.IEvent import IEvent
+
+from 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('/foo')
+        self.subscriber = DummySubscriber()
+        
+    def testSubscribe1(self):
+        "Test subscribe method with one parameter"
+        self.service.subscribe(self.subscriber)
+        self.service.publishEvent(self.event)
+        self.assertEqual(self.subscriber.notified, 1)
+        
+    def testSubscribe2(self):
+        "Test subscribe method with two parameters"
+        self.service.subscribe(
+            self.subscriber,
+            event_type=IObjectAddedEvent
+            )
+        self.service.publishEvent(self.event)
+        self.assertEqual(self.subscriber.notified, 1)
+
+    def testSubscribe3(self):
+        "Test subscribe method with three parameters"
+        self.service.subscribe(
+            self.subscriber,
+            event_type=IObjectAddedEvent,
+            filter=DummyFilter()
+            )
+        self.service.publishEvent(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_type=IObjectAddedEvent,
+            filter=DummyFilter(0)
+            )
+        self.service.publishEvent(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_type=IObjectModifiedEvent,
+            filter=DummyFilter()
+            )
+        self.service.publishEvent(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 'publishEvent' method.
+        """
+        self.service.subscribe(
+            self.subscriber,
+            event_type=IObjectEvent
+            )
+        self.service.publishEvent(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 publishEvented event.
+        """
+        self.service.subscribe(
+            self.subscriber,
+            event_type=IObjectModifiedEvent
+            )
+        self.service.subscribe(
+            self.subscriber,
+            event_type=IObjectAddedEvent
+            )
+        self.service.publishEvent(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_type=IObjectAddedEvent,
+            filter=DummyFilter()
+            )
+        self.service.subscribe(
+            self.subscriber,
+            event_type=IObjectAddedEvent,
+            filter=DummyFilter()
+            )
+        self.service.subscribe(
+            self.subscriber,
+            event_type=IObjectAddedEvent,
+            filter=DummyFilter(0)
+            )
+        self.service.publishEvent(self.event)
+        self.assertEqual(self.subscriber.notified, 2)
+
+    def testUnsubscribe1(self):
+        "Test unsubscribe method"
+        subscriber = self.subscriber
+        self.service.subscribe(subscriber)
+        self.service.publishEvent(self.event)
+        self.assertEqual(self.subscriber.notified, 1)
+        self.service.unsubscribe(subscriber)
+        self.service.publishEvent(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('/foo')
+        self.service.subscribe(
+            self.subscriber)
+        self.service.subscribe(
+            self.subscriber,
+            event_type=IObjectAddedEvent,
+            filter=filter
+            )
+        self.service.subscribe(
+            self.subscriber,
+            event_type=IObjectAddedEvent
+            )
+        self.service.subscribe(
+            subscriber2,
+            event_type=IObjectAddedEvent
+            )
+        self.service.publishEvent(self.event)
+        self.assertEqual(self.subscriber.notified, 3)
+        self.assertEqual(subscriber2.notified, 1)
+        self.service.publishEvent(event2)
+        self.assertEqual(self.subscriber.notified, 4)
+        self.assertEqual(subscriber2.notified, 1)
+        self.service.unsubscribe(self.subscriber, IObjectAddedEvent)
+        self.service.publishEvent(self.event)
+        self.assertEqual(self.subscriber.notified, 6)
+        self.assertEqual(subscriber2.notified, 2)
+        self.service.unsubscribe(self.subscriber, IEvent)
+        self.service.publishEvent(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.publishEvent(self.event)
+        self.assertEqual(self.subscriber.notified, 6)
+        self.assertEqual(subscriber2.notified, 3)
+        self.service.unsubscribe(subscriber2, IObjectAddedEvent)
+        self.service.publishEvent(self.event)
+        self.assertEqual(self.subscriber.notified, 6)
+        self.assertEqual(subscriber2.notified, 3)
+
+    def testpublishEvent1(self):
+        "Test publishEvent method"
+        subscriber = self.subscriber
+        self.service.subscribe(subscriber)
+        self.assertEqual(self.subscriber.notified, 0)        
+        self.service.publishEvent(self.event)
+        self.assertEqual(self.subscriber.notified, 1)
+
+    def testpublishEvent2(self):
+        """Test publishEvent 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_type=IObjectEvent,
+            )
+        self.service.subscribe(
+            self.subscriber,
+            event_type=IObjectAddedEvent,
+            )
+        self.service.publishEvent(self.event)
+        self.assertEqual(self.subscriber.notified, 2) 
+
+    def testpublishEvent3(self):
+        """Test publishEvent 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_type=IObjectRemovedEvent
+            )
+        self.service.subscribe(
+            self.subscriber,
+            event_type=IObjectAddedEvent
+            )
+        self.service.publishEvent(DummyEvent())
+        self.assertEqual(self.subscriber.notified, 2)
+
+    def testpublishEvent4(self):
+        """Test publishEvent method to make sure that we don't
+        'leak registrations up' sez Jim.
+        """
+        subscriber = self.subscriber
+        self.service.subscribe(
+            self.subscriber,
+            event_type=IObjectEvent
+            )
+        self.service.subscribe(
+            self.subscriber,
+            event_type=IObjectAddedEvent
+            )
+        self.service.publishEvent(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.subscribe(
+            self.subscriber,
+            event_type=IObjectAddedEvent
+            )
+        self.assertEqual([(IObjectAddedEvent,None)], self.service.listSubscriptions(self.subscriber))
+    
+    def testListSubscriptions3(self):
+        "listing limited subscription"
+        self.service.subscribe(
+            self.subscriber,
+            event_type=IObjectAddedEvent
+            )
+        self.assertEqual([], self.service.listSubscriptions(self.subscriber, IObjectRemovedEvent))
+    
+def test_suite():
+    return unittest.TestSuite((
+        unittest.makeSuite(TestEventService),
+        ))
+
+if __name__=='__main__':
+    unittest.main(defaultTest='test_suite')


=== Zope3/lib/python/Zope/Event/tests/testLogger.py 1.1 => 1.2 ===
+#
+# 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$
+"""
+
+import unittest, sys
+from Zope.ComponentArchitecture.tests.PlacelessSetup import PlacelessSetup
+from Zope.ComponentArchitecture import getServiceManager, getService
+
+from Zope.Event import subscribe, unsubscribe, publishEvent
+from Zope.Event.ObjectEvent import ObjectAddedEvent
+from Zope.Event.Logger import Logger
+
+import zLOG
+from zLOG import BLATHER, PANIC
+
+from Zope.Event.GlobalEventService import GlobalEventService
+
+class DopeyLogger:
+
+    def __init__(self):
+        self.result=[]
+        
+    def log_write(self, subsystem, severity, summary, detail, error):
+        self.result.append((subsystem,severity,summary,detail,error))
+        
+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.Event.GlobalEventService import eventService
+        getServiceManager(None).provideService("Events", eventService)
+        # futz a logger in for testing
+        self.__old_log_write = zLOG.log_write
+        self.logger = DopeyLogger()
+        zLOG.log_write = self.logger.log_write
+        # register a logger
+        subscribe(self.eventlogger)
+        # send an event
+        publishEvent(None, ObjectAddedEvent('foo'))
+
+    def tearDown(self):
+        unsubscribe(self.eventlogger)
+        zLOG.log_write = self.__old_log_write
+        PlacelessSetup.tearDown(self)
+        
+    def testLogger(self):
+        "Test the logger logs appropriately"
+        # check the dopey logger
+        self.assertEqual(self.logger.result,
+                         [
+            (
+            'Event.Logger',
+            BLATHER,
+            'Zope.Event.ObjectEvent.ObjectAddedEvent',
+            "{'_ObjectAddedEvent__location': 'foo'}\n",
+            None,
+            )
+            ])
+
+class TestLogger2(TestLogger1):
+
+    eventlogger = Logger(PANIC)
+
+    def testLogger(self):
+        "Test the logger logs appropriately"
+        # check the dopey logger
+        self.assertEqual(self.logger.result,
+                         [
+            (
+            'Event.Logger',
+            PANIC,
+            'Zope.Event.ObjectEvent.ObjectAddedEvent',
+            "{'_ObjectAddedEvent__location': 'foo'}\n",
+            None,
+            )
+            ])
+
+def test_suite():
+    return unittest.TestSuite((
+        unittest.makeSuite(TestLogger1),
+        unittest.makeSuite(TestLogger2),
+        ))
+
+if __name__=='__main__':
+    unittest.main(defaultTest='test_suite')


=== Zope3/lib/python/Zope/Event/tests/testObjectEvent.py 1.1 => 1.2 ===
+#
+# 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$
+"""
+
+import unittest, sys
+
+from Zope.Event.ObjectEvent import ObjectAddedEvent, ObjectModifiedEvent
+from Zope.Event.ObjectEvent import ObjectRemovedEvent, ObjectMovedEvent
+
+class TestObjectAddedEvent(unittest.TestCase):
+    
+    location = '/some/location'
+    klass = ObjectAddedEvent
+    
+    def setUp(self):
+        self.event = self.klass(self.location)
+        
+    def testGetLocation(self):
+        "Test getLocation method"
+        self.assertEqual(self.event.getLocation(),self.location)
+
+class TestObjectModifiedEvent(TestObjectAddedEvent):
+
+    klass = ObjectModifiedEvent
+
+class TestObjectRemovedEvent(TestObjectAddedEvent):
+
+    
+    location = '/some/location'
+    obj = object()
+    
+    def setUp(self):
+        self.event = ObjectRemovedEvent(self.location, self.obj)
+        
+    def testGetLocation(self):
+        "Test getLocation method"
+        self.assertEqual(self.event.getLocation(),self.location)
+        
+    def testGetObject(self):
+        "Test getObject method"
+        self.assertEqual(self.event.getObject(), self.obj)
+
+
+
+class TestObjectMovedEvent(TestObjectAddedEvent):
+
+    from_location = '/some/other/location'
+    
+    def setUp(self):
+        self.event = ObjectMovedEvent(self.from_location, self.location)
+
+    def testFromLocation(self):
+        "Test getFromLocation method"
+        self.assertEqual(self.event.getFromLocation(),self.from_location)
+        
+def test_suite():
+    return unittest.TestSuite((
+        unittest.makeSuite(TestObjectAddedEvent),
+        unittest.makeSuite(TestObjectModifiedEvent),
+        unittest.makeSuite(TestObjectRemovedEvent),
+        unittest.makeSuite(TestObjectMovedEvent),
+        ))
+
+if __name__=='__main__':
+    unittest.main(defaultTest='test_suite')