[Zope-Checkins] CVS: Zope3/lib/python/Zope/ObjectHub/tests - testRuidObjectEvent.py:1.1.2.1 testObjectHub.py:1.1.2.3

Steve Alexander steve@cat-box.net
Sun, 24 Feb 2002 16:35:39 -0500


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

Modified Files:
      Tag: Zope-3x-branch
	testObjectHub.py 
Added Files:
      Tag: Zope-3x-branch
	testRuidObjectEvent.py 
Log Message:
Added RuidObjectEvents and tests. Updated ObjectHub to act as a persistent EventChannel. Updated tests for ObjectHub to test that it sends out the correct kindsof event for the events it receives. That bit was fun! :-)



=== Added File Zope3/lib/python/Zope/ObjectHub/tests/testRuidObjectEvent.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: testRuidObjectEvent.py,v 1.1.2.1 2002/02/24 21:35:39 stevea Exp $
"""

import unittest, sys

from Zope.ObjectHub.RuidObjectEvent import RuidObjectRegisteredEvent
from Zope.ObjectHub.RuidObjectEvent import RuidObjectUnregisteredEvent
from Zope.ObjectHub.RuidObjectEvent import RuidObjectAddedEvent
from Zope.ObjectHub.RuidObjectEvent import RuidObjectModifiedEvent
from Zope.ObjectHub.RuidObjectEvent import RuidObjectContextChangedEvent
from Zope.ObjectHub.RuidObjectEvent import RuidObjectRemovedEvent

from Zope.Exceptions import NotFoundError

class DummyObjectHub:

    def __init__(self, ruid, obj):
        self.ruid = ruid
        self.obj = obj
    
    
    def getObject(self, ruid):
        if ruid==self.ruid:
            return self.obj
            
        raise NotFoundError
        
class TestRuidObjectAddedEvent(unittest.TestCase):
    
    location = '/some/location'
    ruid = 23
    obj = object()
    klass = RuidObjectAddedEvent
    
    def setUp(self):
        objecthub = DummyObjectHub(self.ruid, self.obj)
        self.event = self.klass(objecthub, self.ruid, self.location)
        
    def testGetLocation(self):
        "Test getLocation method"
        self.assertEqual(self.event.getLocation(), self.location)
        
    def testGetRuid(self):
        "Test getRuid method"
        self.assertEqual(self.event.getRuid(), self.ruid)
    
    def testGetObject(self):
        "Test getObject method"
        self.assertEqual(self.event.getObject(), self.obj)
    
class TestRuidObjectRegisteredEvent(TestRuidObjectAddedEvent):

    klass = RuidObjectRegisteredEvent

class TestRuidObjectUnregisteredEvent(TestRuidObjectAddedEvent):

    klass = RuidObjectUnregisteredEvent

class TestRuidObjectModifiedEvent(TestRuidObjectAddedEvent):

    klass = RuidObjectModifiedEvent

class TestRuidObjectContextChangedEvent(TestRuidObjectAddedEvent):

    klass = RuidObjectContextChangedEvent

class TestRuidObjectRemovedEvent(TestRuidObjectAddedEvent):

    klass = RuidObjectRemovedEvent

    def setUp(self):
        self.event = self.klass(self.obj, self.ruid, self.location)



def test_suite():
    return unittest.TestSuite((
        unittest.makeSuite(TestRuidObjectAddedEvent),
        unittest.makeSuite(TestRuidObjectRegisteredEvent),
        unittest.makeSuite(TestRuidObjectUnregisteredEvent),
        unittest.makeSuite(TestRuidObjectModifiedEvent),
        unittest.makeSuite(TestRuidObjectContextChangedEvent),
        unittest.makeSuite(TestRuidObjectRemovedEvent),
        ))

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


=== Zope3/lib/python/Zope/ObjectHub/tests/testObjectHub.py 1.1.2.2 => 1.1.2.3 ===
 from Zope.Event.ObjectEvent import ObjectAddedEvent, ObjectModifiedEvent
 from Zope.Event.ObjectEvent import ObjectRemovedEvent, ObjectMovedEvent
+from Zope.Event.ISubscriber import ISubscriber
+
 from Zope.ObjectHub.ObjectHub import ObjectHub, ObjectHubError
+from Zope.ObjectHub.IRuidObjectEvent import IRuidObjectAddedEvent
+from Zope.ObjectHub.IRuidObjectEvent import IRuidObjectRemovedEvent
+from Zope.ObjectHub.IRuidObjectEvent import IRuidObjectModifiedEvent
+from Zope.ObjectHub.IRuidObjectEvent import IRuidObjectContextChangedEvent
+
 from Zope.Exceptions import NotFoundError
+from types import StringTypes
+
+class LoggingSubscriber:
+
+    __implements__ = ISubscriber
+    
+    def __init__(self):
+        self.events_received = []
+    
+    def notify(self, event):
+        self.events_received.append(event)
+
 
+    # see ObjectHub._canonical
+    def _canonical(location):
+        """ returns a canonical traversal location for a location that is 
+        a string or a sequence of strings """
+        if not isinstance(location, StringTypes):
+            location='/'.join(location)
+        # URIs are ascii, right?
+        return str(location)
+    _canonical = staticmethod(_canonical)
+    
+    def verifyEventsReceived(self, testcase, event_spec_list):
+        # iterate through self.events_received and check
+        # that each one implements the interface that is
+        # in the same place, with the same location and ruid
+        
+        testcase.assertEqual(len(event_spec_list), len(self.events_received))
+        
+        for spec,event in zip(event_spec_list, self.events_received):
+            interface,ruid,location = spec
+            location = self._canonical(location)
+            testcase.assert_(interface.isImplementedBy(event))
+            testcase.assertEqual(event.getLocation(), location)
+            if ruid is not None:
+                testcase.assertEqual(event.getRuid(), ruid)
+            
+    
 class BasicHubTest(unittest.TestCase):
 
     location = '/foo/bar'
@@ -37,6 +82,10 @@
         self.moved_event = ObjectMovedEvent(self.location,
                                             self.new_location)
         self.object_hub = ObjectHub()
+        self.subscriber = LoggingSubscriber()
+        self.object_hub.subscribe(self.subscriber)
+
+        # TODO: test that ObjectHub acts as an EventChannel
 
 
 class TestObjectAddedEvent(BasicHubTest):
@@ -60,6 +109,10 @@
         location_from_hub = hub.lookupLocation(ruid)
 
         self.assertEqual(location_from_hub, location)
+        
+        self.subscriber.verifyEventsReceived(self, [
+                (IRuidObjectAddedEvent, ruid, location)
+            ])
 
         
     def testLookupUpAbsentLocation(self):
@@ -75,6 +128,9 @@
         
         self.assertRaises(NotFoundError, hub.lookupRuid, location)
 
+        self.subscriber.verifyEventsReceived(self, [])
+
+
         
     def testLookupUpAbsentRuid(self):
         """Test that we don't find a location for an ruid
@@ -89,6 +145,8 @@
         absent_ruid = 12
         
         self.assertRaises(NotFoundError, hub.lookupLocation, absent_ruid)
+        
+        self.subscriber.verifyEventsReceived(self, [])
 
 
 class TestObjectRemovedEvent(BasicHubTest):
@@ -114,6 +172,12 @@
         self.assertRaises(NotFoundError, hub.lookupRuid, location)
         self.assertRaises(NotFoundError, hub.lookupLocation, ruid)
         
+        self.subscriber.verifyEventsReceived(self, [
+                (IRuidObjectAddedEvent, ruid, location),
+                (IRuidObjectRemovedEvent, ruid, location)
+            ])
+        
+        
     def testRemovedAbsentLocation(self):
         """Test that removing an absent location is silently ignored.
         """
@@ -126,6 +190,8 @@
         # hub.notify(added_event)
                 
         hub.notify(removed_event)
+
+        self.subscriber.verifyEventsReceived(self, [])
                 
 
 class TestObjectModifiedEvent(BasicHubTest):
@@ -156,6 +222,12 @@
         self.assertEqual(location_from_hub, location_from_hub2)
         self.assertEqual(ruid, ruid2)
         
+        self.subscriber.verifyEventsReceived(self, [
+                (IRuidObjectAddedEvent, ruid, location),
+                (IRuidObjectModifiedEvent, ruid, location)
+            ])
+
+        
     def testModifiedAbsentLocation(self):
         """Test that lookup state does not change after an object
         modify event. In this case, modify of an absent location is
@@ -171,6 +243,9 @@
         
         hub.notify(modified_event)
         self.assertRaises(NotFoundError, hub.lookupRuid, location)
+        
+        self.subscriber.verifyEventsReceived(self, [])
+
 
 class TestObjectMovedEvent(BasicHubTest):
 
@@ -195,6 +270,12 @@
                 
         ruid2 = hub.lookupRuid(new_location)
         self.assertEqual(ruid2, ruid)
+        
+        self.subscriber.verifyEventsReceived(self, [
+                (IRuidObjectAddedEvent, ruid, location),
+                (IRuidObjectContextChangedEvent, ruid, new_location)
+            ])
+
 
     def testMovedAbsentLocation(self):
         """Test that moving an absent location is a noop.
@@ -211,6 +292,9 @@
         hub.notify(moved_event)
         self.assertRaises(NotFoundError, hub.lookupRuid, location)
         self.assertRaises(NotFoundError, hub.lookupRuid, new_location)
+        
+        self.subscriber.verifyEventsReceived(self, [])
+
 
     def testMovedToExistingLocation(self):
         """Test that moving to an existing location raises ObjectHubError.
@@ -218,9 +302,15 @@
         hub = self.object_hub
         added_event2 = self.added_new_location_event
         moved_event = self.moved_event
+        location = self.new_location
         
         hub.notify(added_event2)
+        
         self.assertRaises(ObjectHubError, hub.notify, moved_event)
+        
+        self.subscriber.verifyEventsReceived(self, [
+                (IRuidObjectAddedEvent, None, location)
+            ])