[Zope3-checkins] CVS: Zope3/lib/python/Zope/App/OFS/Services/LocalObjectHub/tests - ObjectHubSetup.py:1.1 __init__.py:1.1 testHubEvent.py:1.1 testObjectHub.py:1.1
Gary Poster
gary@modernsongs.com
Mon, 21 Oct 2002 02:14:48 -0400
Update of /cvs-repository/Zope3/lib/python/Zope/App/OFS/Services/LocalObjectHub/tests
In directory cvs.zope.org:/tmp/cvs-serv13005/Zope/App/OFS/Services/LocalObjectHub/tests
Added Files:
ObjectHubSetup.py __init__.py testHubEvent.py testObjectHub.py
Log Message:
sorry for the huge honking checkin.
Adds a simple local objecthub implementation and made ObjectHub a service
Modifies the main objecthub as we have been discussing:
* objecthub attribute is hubid, not hid (sorry Jim, I'll change it back if you want but there were a lot of "yay"s and no "nay"s :-)
* no more IObjectAddedHubEvent
* IObjectRemovedEvent now (hopefully) actually has the effect on the ObjectHub that is described in the interface, and that we agreed upon, namely (assuming removed object was cataloged in objecthub) removing catalog of object in objecthub and sending out an IObjectRemovedHubEvent, subclass of IObjectUnregisteredHubEvent, to the ObjectHub subscribers
I tried to spruce up the LocalEventService a bit but the code still looks as opaque as ever, I'm afraid. Among other small improvements, though, at least you actually can see the silly "user interface" now without a traceback. Now for a *real* user interface sometime. :-)
Fixed a few typos while I was at it as well...and I'm sure made my share of new ones :-)
=== Added File Zope3/lib/python/Zope/App/OFS/Services/LocalObjectHub/tests/ObjectHubSetup.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: ObjectHubSetup.py,v 1.1 2002/10/21 06:14:47 poster Exp $
"""
from Zope.App.OFS.Services.LocalEventService.tests.EventSetup import \
EventSetup
from Zope.ComponentArchitecture import getService, getServiceManager
from Zope.App.OFS.Services.ServiceManager.ServiceDirective \
import ServiceDirective
from Zope.App.Traversing import getPhysicalPathString
from Zope.App.OFS.Services.LocalObjectHub.LocalObjectHub import LocalObjectHub
class ObjectHubSetup(EventSetup):
def setUp(self):
EventSetup.setUp(self)
from Zope.ObjectHub.IObjectHub import IObjectHub
globsm=getServiceManager(None)
globsm.defineService("ObjectHub", IObjectHub)
self.createObjectHub()
def createObjectHub(self, folder=None):
if folder is None: folder=self.rootFolder
if not folder.hasServiceManager():
self.createServiceManager(folder)
sm=getServiceManager(folder) # wrapped now
sm.Packages['default'].setObject("myObjectHub",LocalObjectHub())
path = "%s/Packages/default/myObjectHub" % getPhysicalPathString(sm)
directive = ServiceDirective("ObjectHub", path)
sm.Packages['default'].setObject("myObjectHubDir", directive)
sm.bindService(directive)
=== Added File Zope3/lib/python/Zope/App/OFS/Services/LocalObjectHub/tests/__init__.py ===
=== Added File Zope3/lib/python/Zope/App/OFS/Services/LocalObjectHub/tests/testHubEvent.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.
#
##############################################################################
"""
Revision information:
$Id: testHubEvent.py,v 1.1 2002/10/21 06:14:47 poster Exp $
"""
# in the local version of these tests, we are no longer using a fake
# ObjectHub, which makes these tests less pure...but still useful
# as a test for both the events and the object hub now.
import unittest, sys
from ObjectHubSetup import ObjectHubSetup
from Zope.App.OFS.Services.LocalObjectHub.LocalHubEvent import \
ObjectRegisteredHubEvent, ObjectUnregisteredHubEvent, \
ObjectModifiedHubEvent, ObjectMovedHubEvent, \
ObjectRemovedHubEvent
from Zope.App.Traversing import getPhysicalPathString
from Zope.Exceptions import NotFoundError
from Zope.ComponentArchitecture import getService
class AbstractTestHubEvent(ObjectHubSetup, unittest.TestCase):
klass = None
def setUp(self):
ObjectHubSetup.setUp(self)
self.object_hub = getService(self.rootFolder, "ObjectHub")
self.obj = self.folder1_2_1
self.hubid = self.object_hub.register(self.obj)
self.location = getPhysicalPathString(self.obj)
self.event = self.klass(self.object_hub, self.hubid, self.location)
def testGetLocation(self):
"Test getLocation method"
self.assertEqual(self.event.location, self.location)
def testGetHubId(self):
"Test getHubId method"
self.assertEqual(self.event.hubid, self.hubid)
def testGetObject(self):
"Test getObject method"
self.assertEqual(self.event.object, self.obj)
class TestObjectRegisteredHubEvent(AbstractTestHubEvent):
klass = ObjectRegisteredHubEvent
class TestObjectUnregisteredHubEvent(AbstractTestHubEvent):
klass = ObjectUnregisteredHubEvent
class TestObjectModifiedHubEvent(AbstractTestHubEvent):
klass = ObjectModifiedHubEvent
class TestObjectMovedHubEvent(AbstractTestHubEvent):
klass = ObjectMovedHubEvent
class TestObjectRemovedHubEvent(AbstractTestHubEvent):
klass = ObjectRemovedHubEvent
def setUp(self):
AbstractTestHubEvent.setUp(self)
self.event = self.klass(self.obj, self.hubid, self.location)
def test_suite():
return unittest.TestSuite((
unittest.makeSuite(TestObjectRegisteredHubEvent),
unittest.makeSuite(TestObjectUnregisteredHubEvent),
unittest.makeSuite(TestObjectModifiedHubEvent),
unittest.makeSuite(TestObjectMovedHubEvent),
unittest.makeSuite(TestObjectRemovedHubEvent),
))
if __name__=='__main__':
unittest.main(defaultTest='test_suite')
=== Added File Zope3/lib/python/Zope/App/OFS/Services/LocalObjectHub/tests/testObjectHub.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.
#
##############################################################################
"""testObjectHub
Revision information:
$Id: testObjectHub.py,v 1.1 2002/10/21 06:14:47 poster Exp $
"""
import unittest, sys
from ObjectHubSetup import ObjectHubSetup
from Zope.Event.IObjectEvent import IObjectAddedEvent, IObjectRemovedEvent
from Zope.Event.IObjectEvent import IObjectModifiedEvent, IObjectMovedEvent
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 ObjectHubError
from Zope.ObjectHub.IHubEvent import IObjectRemovedHubEvent
from Zope.ObjectHub.IHubEvent import IObjectModifiedHubEvent
from Zope.ObjectHub.IHubEvent import IObjectMovedHubEvent
from Zope.ObjectHub.IHubEvent import IObjectRegisteredHubEvent
from Zope.ObjectHub.IHubEvent import IObjectUnregisteredHubEvent
import Zope.ObjectHub.HubEvent as HubIdObjectEvent
from Zope.Exceptions import NotFoundError
from types import StringTypes
from Zope.App.Traversing import locationAsUnicode
from Zope.ObjectHub.tests.testObjectHub import LoggingSubscriber, \
RegistrationSubscriber
from Zope.ComponentArchitecture import getService, getServiceManager
# while these tests don't really test much of the placeful aspect of the
# object hub, they (as duplicates of the global versions) do at least
# test basic functionality.
# we'll need real tests of the placeful aspects, as well as of the new
# features (being able to supply wrapped objects) but for now a basic
# test happens simply by virtue of the testHubEvent module in this
# directory
class TransmitHubEventTest(ObjectHubSetup, unittest.TestCase):
hubid = 23
location = '/foo/bar'
# Don't test the HubEvent base class.
# See below for testing subclasses / subinterfaces
# klass = HubEvent
# interface = IHubEvent
def setUp(self):
ObjectHubSetup.setUp(self)
self.object_hub = getService(self.rootFolder, "ObjectHub")
self.hub_event = self.klass(self.object_hub,
self.hubid,
self.location)
self.subscriber = LoggingSubscriber()
self.object_hub.subscribe(self.subscriber)
def testTransmittedEvent(self):
"""Test that the HubEvents are transmitted by the notify method
"""
self.object_hub.notify(self.hub_event)
self.subscriber.verifyEventsReceived(self, [
(self.interface, self.hubid, self.location)
])
class TransmitObjectRemovedHubEventTest(TransmitHubEventTest):
interface = IObjectRemovedHubEvent
klass = HubIdObjectEvent.ObjectRemovedHubEvent
class TransmitObjectModifiedHubEventTest(TransmitHubEventTest):
interface = IObjectModifiedHubEvent
klass = HubIdObjectEvent.ObjectModifiedHubEvent
class TransmitObjectMovedHubEventTest(TransmitHubEventTest):
interface = IObjectMovedHubEvent
klass = HubIdObjectEvent.ObjectMovedHubEvent
class TransmitObjectRegisteredHubEventTest(TransmitHubEventTest):
interface = IObjectRegisteredHubEvent
klass = HubIdObjectEvent.ObjectRegisteredHubEvent
class TransmitObjectUnregisteredHubEventTest(TransmitHubEventTest):
interface = IObjectUnregisteredHubEvent
klass = HubIdObjectEvent.ObjectUnregisteredHubEvent
class BasicHubTest(ObjectHubSetup, unittest.TestCase):
location = '/foo/bar'
obj = object()
new_location = '/baz/spoo'
def setUp(self):
ObjectHubSetup.setUp(self)
self.object_hub = getService(self.rootFolder, "ObjectHub")
self.setEvents()
self.subscriber = LoggingSubscriber()
self.object_hub.subscribe(self.subscriber)
def setEvents(self):
self.added_event = ObjectAddedEvent(self.obj, self.location)
self.added_new_location_event = ObjectAddedEvent(
self.obj, self.new_location)
self.removed_event = ObjectRemovedEvent(self.obj, self.location)
self.modified_event = ObjectModifiedEvent(self.obj, self.location)
self.moved_event = ObjectMovedEvent(self.obj,
self.location,
self.new_location)
class TestRegistrationEvents(BasicHubTest):
def testRegistration(self):
self.assertRaises(NotFoundError,
self.object_hub.unregister,
self.location)
self.assertRaises(NotFoundError, self.object_hub.unregister, 42)
hubid = self.object_hub.register(self.location)
hubid2 = self.object_hub.register(self.new_location)
self.subscriber.verifyEventsReceived(self, [
(IObjectRegisteredHubEvent, hubid, self.location),
(IObjectRegisteredHubEvent, hubid2, self.new_location)
])
# register again and check for error
self.assertRaises(ObjectHubError,
self.object_hub.register,
self.location)
# unregister first object by location
self.object_hub.unregister(self.location)
self.subscriber.verifyEventsReceived(self, [
(IObjectUnregisteredHubEvent, hubid, self.location)
])
# unregister second object by hub id
self.object_hub.unregister(hubid2)
self.subscriber.verifyEventsReceived(self, [
(IObjectUnregisteredHubEvent, hubid2, self.new_location)
])
def testRegistrationRelativeLocation(self):
self.assertRaises(ValueError, self.object_hub.register, 'foo/bar')
class TestNoRegistration(BasicHubTest):
def testAddWithoutRegistration(self):
"""Test that no HubIdEvents are generated
if there is no registration
"""
hub = self.object_hub
event = self.added_event
location = self.location
hub.notify(event)
self.subscriber.verifyEventsReceived(self, [
(IObjectAddedEvent, location),
])
class TestObjectAddedEvent(BasicHubTest):
def setUp(self):
ObjectHubSetup.setUp(self)
self.object_hub = getService(self.rootFolder, "ObjectHub")
self.setEvents()
self.subscriber = RegistrationSubscriber(self.object_hub)
self.object_hub.subscribe(self.subscriber)
def testLookingUpLocation(self):
"""Test that the location is in the lookup
Compare getHubIdForLocation and getLocationForHubId
Checks the sequence of events
"""
hub = self.object_hub
event = self.added_event
location = self.location
hub.notify(event)
hubid = hub.lookupHubId(location)
# check that hub id is an int
self.failUnless(isinstance(hubid, int)) # int(hubid)
location_from_hub = hub.lookupLocation(hubid)
self.assertEqual(location_from_hub, location)
self.subscriber.verifyEventsReceived(self, [
(IObjectAddedEvent, location),
(IObjectRegisteredHubEvent, hubid, location),
])
def testLookupUpAbsentLocation(self):
"""Test that we don't find an hub id for location
that we haven't added.
"""
hub = self.object_hub
event = self.added_event
location = self.location
# Do not add the location to the hub
# hub.notify(event)
self.assertRaises(NotFoundError, hub.lookupHubId, location)
self.subscriber.verifyEventsReceived(self, [])
def testLookupUpAbsentHubId(self):
"""Test that we don't find a location for an hub id
that isn't there.
"""
hub = self.object_hub
event = self.added_event
# Do not add the location to the hub
# hub.notify(event)
absent_hubid = 12
self.assertRaises(NotFoundError, hub.lookupLocation, absent_hubid)
self.subscriber.verifyEventsReceived(self, [])
class TestObjectRemovedEvent(BasicHubTest):
def setUp(self):
ObjectHubSetup.setUp(self)
self.object_hub = getService(self.rootFolder, "ObjectHub")
self.setEvents()
self.subscriber = RegistrationSubscriber(self.object_hub)
self.object_hub.subscribe(self.subscriber)
def testRemovedLocation(self):
"""Test that a location that is added then removed is
actually gone.
"""
hub = self.object_hub
added_event = self.added_event
removed_event = self.removed_event
location = self.location
obj = self.obj
hub.notify(added_event)
hubid = hub.lookupHubId(location)
# check that hubid is an int
self.failUnless(isinstance(hubid, int)) # int(hubid)
hub.notify(removed_event)
self.assertRaises(NotFoundError, hub.lookupHubId, location)
self.assertRaises(NotFoundError, hub.lookupLocation, hubid)
self.subscriber.verifyEventsReceived(self, [
(IObjectAddedEvent, location),
(IObjectRegisteredHubEvent, hubid, location),
(IObjectRemovedEvent, location),
(IObjectRemovedHubEvent, hubid, location, obj),
])
def testRemovedAbsentLocation(self):
"""Test that removing an absent location is silently ignored.
"""
hub = self.object_hub
added_event = self.added_event
removed_event = self.removed_event
location = self.location
# Do not add location
# hub.notify(added_event)
hub.notify(removed_event)
self.subscriber.verifyEventsReceived(self, [
(IObjectRemovedEvent, location),
])
class TestObjectModifiedEvent(BasicHubTest):
def setUp(self):
ObjectHubSetup.setUp(self)
self.object_hub = getService(self.rootFolder, "ObjectHub")
self.setEvents()
self.subscriber = RegistrationSubscriber(self.object_hub)
self.object_hub.subscribe(self.subscriber)
def testModifiedLocation(self):
"""Test that lookup state does not change after an object
modify event.
"""
hub = self.object_hub
added_event = self.added_event
modified_event = self.modified_event
location = self.location
hub.notify(added_event)
hubid = hub.lookupHubId(location)
# check that hubid is an int
self.failUnless(isinstance(hubid, int)) # int(hubid)
location_from_hub = hub.lookupLocation(hubid)
self.assertEqual(location_from_hub, location)
hub.notify(modified_event)
hubid2 = hub.lookupHubId(location)
location_from_hub2 = hub.lookupLocation(hubid2)
self.assertEqual(location_from_hub, location_from_hub2)
self.assertEqual(hubid, hubid2)
self.subscriber.verifyEventsReceived(self, [
(IObjectAddedEvent, location),
(IObjectRegisteredHubEvent, hubid, location),
(IObjectModifiedEvent, location),
(IObjectModifiedHubEvent, hubid, 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
a noop.
"""
hub = self.object_hub
added_event = self.added_event
modified_event = self.modified_event
location = self.location
# Do not add location
# hub.notify(added_event)
hub.notify(modified_event)
self.assertRaises(NotFoundError, hub.lookupHubId, location)
self.subscriber.verifyEventsReceived(self, [
(IObjectModifiedEvent, location),
])
class TestObjectMovedEvent(BasicHubTest):
def setUp(self):
ObjectHubSetup.setUp(self)
self.object_hub = getService(self.rootFolder, "ObjectHub")
self.setEvents()
self.subscriber = RegistrationSubscriber(self.object_hub)
self.object_hub.subscribe(self.subscriber)
def testMovedLocation(self):
"""Test that the location does indeed change after a move.
"""
hub = self.object_hub
added_event = self.added_event
moved_event = self.moved_event
location = self.location
new_location = self.new_location
hub.notify(added_event)
hubid = hub.lookupHubId(location)
hub.notify(moved_event)
location_from_hub = hub.lookupLocation(hubid)
self.assertEqual(location_from_hub, new_location)
self.assertRaises(NotFoundError, hub.lookupHubId, location)
hubid2 = hub.lookupHubId(new_location)
self.assertEqual(hubid2, hubid)
self.subscriber.verifyEventsReceived(self, [
(IObjectAddedEvent, location),
(IObjectRegisteredHubEvent, hubid, location),
(IObjectMovedEvent, new_location),
(IObjectMovedHubEvent, hubid, new_location)
])
def testMovedAbsentLocation(self):
"""Test that moving an absent location is a no-op.
"""
hub = self.object_hub
added_event = self.added_event
moved_event = self.moved_event
location = self.location
new_location = self.new_location
# Do not add location
# hub.notify(added_event)
hub.notify(moved_event)
self.assertRaises(NotFoundError, hub.lookupHubId, location)
self.assertRaises(NotFoundError, hub.lookupHubId, new_location)
self.subscriber.verifyEventsReceived(self, [
(IObjectMovedEvent, new_location),
])
def testMovedToExistingLocation(self):
"""Test that moving to an existing location raises ObjectHubError.
"""
hub = self.object_hub
added_event = self.added_event
added_event2 = self.added_new_location_event
moved_event = self.moved_event
location = self.location
new_location = self.new_location
hub.notify(added_event)
hub.notify(added_event2)
self.assertRaises(ObjectHubError, hub.notify, moved_event)
self.subscriber.verifyEventsReceived(self, [
(IObjectAddedEvent, location),
(IObjectRegisteredHubEvent, None, location),
(IObjectAddedEvent, new_location),
(IObjectRegisteredHubEvent, None, new_location),
(IObjectMovedEvent, new_location),
])
def test_suite():
return unittest.TestSuite((
unittest.makeSuite(TransmitObjectRemovedHubEventTest),
unittest.makeSuite(TransmitObjectModifiedHubEventTest),
unittest.makeSuite(TransmitObjectMovedHubEventTest),
unittest.makeSuite(TransmitObjectRegisteredHubEventTest),
unittest.makeSuite(TransmitObjectUnregisteredHubEventTest),
unittest.makeSuite(TestRegistrationEvents),
unittest.makeSuite(TestNoRegistration),
unittest.makeSuite(TestObjectAddedEvent),
unittest.makeSuite(TestObjectRemovedEvent),
unittest.makeSuite(TestObjectModifiedEvent),
unittest.makeSuite(TestObjectMovedEvent),
))
if __name__=='__main__':
unittest.main(defaultTest='test_suite')