[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)
+ ])