[Zope3-checkins] CVS: Zope3/src/zope/app/hub/tests -
__init__.py:1.1 objecthubsetup.py:1.1
test_hookedhubevent.py:1.1 test_hubevent.py:1.1
test_objecthub.py:1.1 test_registration.py:1.1
Stephan Richter
srichter at cosmos.phy.tufts.edu
Thu Mar 11 04:19:27 EST 2004
Update of /cvs-repository/Zope3/src/zope/app/hub/tests
In directory cvs.zope.org:/tmp/cvs-serv28114/src/zope/app/hub/tests
Added Files:
__init__.py objecthubsetup.py test_hookedhubevent.py
test_hubevent.py test_objecthub.py test_registration.py
Log Message:
Moved object hub to zope.app.hub. I did provide module aliases.
=== Added File Zope3/src/zope/app/hub/tests/__init__.py ===
# Import this.
=== Added File Zope3/src/zope/app/hub/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.
#
##############################################################################
"""
$Id: objecthubsetup.py,v 1.1 2004/03/11 09:19:25 srichter Exp $
"""
from zope.app import zapi
from zope.app.event.tests.eventsetup import EventSetup
from zope.app.services.servicenames import HubIds
from zope.app.traversing import traverse, canonicalPath
from zope.app.container.interfaces import IObjectAddedEvent, IObjectMovedEvent
from zope.app.event.interfaces import ISubscriber
from zope.interface import implements
class LoggingSubscriber:
# XXX Jim mentioned there is a new generic
# version of this in zope.app somewhere...
implements(ISubscriber)
def __init__(self):
self.events_received = []
def notify(self, event):
self.events_received.append(event)
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 hub id
testcase.assertEqual(len(event_spec_list), len(self.events_received))
for spec,event in zip(event_spec_list, self.events_received):
if len(spec)==4:
interface,hubid,location,obj = spec
elif len(spec)==3:
interface,hubid,location = spec
obj = None
elif len(spec)==2:
interface, location = spec
obj = None
hubid = None
location = canonicalPath(location)
testcase.assert_(interface.providedBy(event),
'Interface %s' % interface.getName())
testcase.assertEqual(canonicalPath(event.object), location)
if obj is not None:
testcase.assertEqual(event.object, obj)
# Sometimes, the test won't care about the hubid. In this case,
# it is passed into the spec as None.
if hubid is not None:
testcase.assertEqual(event.hubid, hubid)
self.events_received = []
class RegistrationSubscriber(LoggingSubscriber):
def __init__(self, objectHub):
LoggingSubscriber.__init__(self)
self.hub = objectHub
def notify(self, event):
LoggingSubscriber.notify(self, event)
# The policy is to register on object adds and object copies.
if IObjectAddedEvent.providedBy(event):
self.hub.register(event.object)
class ObjectHubSetup(EventSetup):
def setUpRegistrationSubscriber(self):
subscriber = RegistrationSubscriber(self.object_hub)
self.rootFolder['registration_subscriber'] = subscriber
self.subscriber = traverse(self.rootFolder, 'registration_subscriber')
self.object_hub.subscribe(self.subscriber)
def setUpLoggingSubscriber(self):
subscriber = LoggingSubscriber()
self.rootFolder['logging_subscriber'] = subscriber
self.subscriber = traverse(self.rootFolder, 'logging_subscriber')
self.object_hub.subscribe(self.subscriber)
def setUp(self):
EventSetup.setUp(self)
self.object_hub = zapi.getService(self.rootFolder, HubIds)
=== Added File Zope3/src/zope/app/hub/tests/test_hookedhubevent.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.
#
##############################################################################
"""
$Id: test_hookedhubevent.py,v 1.1 2004/03/11 09:19:25 srichter Exp $
"""
# in this 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 for now.
import unittest
from zope.app.hub.tests.objecthubsetup import ObjectHubSetup
from zope.app.hub import \
ObjectRegisteredHubEvent, ObjectUnregisteredHubEvent, \
ObjectModifiedHubEvent, ObjectMovedHubEvent, ObjectRemovedHubEvent
from zope.app.traversing import getPath
class AbstractTestHubEvent(ObjectHubSetup, unittest.TestCase):
klass = None
def setUp(self):
ObjectHubSetup.setUp(self)
self.obj = self.folder1_2_1
self.hubid = self.object_hub.register(self.obj)
self.location = getPath(self.obj)
self.event = self.klass(self.object_hub,
self.hubid,
self.location,
self.obj)
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 TestEmptyObjectRegisteredHubEvent(TestObjectRegisteredHubEvent):
def setUp(self):
ObjectHubSetup.setUp(self)
self.obj = self.folder1_2_1
self.hubid = self.object_hub.register(self.obj)
self.location = getPath(self.obj)
self.event = self.klass(self.object_hub, self.hubid)
class TestObjectUnregisteredHubEvent(AbstractTestHubEvent):
klass = ObjectUnregisteredHubEvent
class TestEmptyObjectUnregisteredHubEvent(TestObjectUnregisteredHubEvent):
def setUp(self):
ObjectHubSetup.setUp(self)
self.obj = self.folder1_2_1
self.hubid = self.object_hub.register(self.obj)
self.location = getPath(self.obj)
self.event = self.klass(self.object_hub, self.hubid, self.location)
class TestObjectModifiedHubEvent(AbstractTestHubEvent):
klass = ObjectModifiedHubEvent
class TestEmptyObjectModifiedHubEvent(TestObjectModifiedHubEvent):
def setUp(self):
ObjectHubSetup.setUp(self)
self.obj = self.folder1_2_1
self.hubid = self.object_hub.register(self.obj)
self.location = getPath(self.obj)
self.event = self.klass(self.object_hub, self.hubid)
class TestObjectMovedHubEvent(AbstractTestHubEvent):
fromLocation = '/old/location'
def setUp(self):
ObjectHubSetup.setUp(self)
self.obj = self.folder1_2_1
self.hubid = self.object_hub.register(self.obj)
self.location = getPath(self.obj)
self.event = self.klass(self.object_hub,
self.hubid,
self.fromLocation,
self.location,
self.obj)
def testGetFromLocation(self):
# Test from location
self.assertEqual(self.event.fromLocation, self.fromLocation)
klass = ObjectMovedHubEvent
class TestEmptyObjectMovedHubEvent(TestObjectMovedHubEvent):
def setUp(self):
ObjectHubSetup.setUp(self)
self.obj = self.folder1_2_1
self.hubid = self.object_hub.register(self.obj)
self.location = getPath(self.obj)
self.event = self.klass(self.object_hub,
self.hubid,
self.fromLocation)
class TestObjectRemovedHubEvent(AbstractTestHubEvent):
klass = ObjectRemovedHubEvent
# Hooked empty object removed not needed
def test_suite():
return unittest.TestSuite((
unittest.makeSuite(TestObjectRegisteredHubEvent),
unittest.makeSuite(TestEmptyObjectRegisteredHubEvent),
unittest.makeSuite(TestObjectUnregisteredHubEvent),
unittest.makeSuite(TestEmptyObjectUnregisteredHubEvent),
unittest.makeSuite(TestObjectModifiedHubEvent),
unittest.makeSuite(TestEmptyObjectModifiedHubEvent),
unittest.makeSuite(TestObjectMovedHubEvent),
unittest.makeSuite(TestEmptyObjectMovedHubEvent),
unittest.makeSuite(TestObjectRemovedHubEvent),
))
if __name__=='__main__':
unittest.main(defaultTest='test_suite')
=== Added File Zope3/src/zope/app/hub/tests/test_hubevent.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.
#
##############################################################################
"""
$Id: test_hubevent.py,v 1.1 2004/03/11 09:19:25 srichter Exp $
"""
import unittest
from zope.app.hub import \
ObjectRegisteredHubEvent, ObjectUnregisteredHubEvent, \
ObjectModifiedHubEvent, ObjectMovedHubEvent, \
ObjectRemovedHubEvent
from zope.exceptions import NotFoundError
class DummyObjectHub:
def __init__(self, ruid, obj, location):
self.ruid = ruid
self.obj = obj
self.location = location
def getObject(self, ruid):
if ruid==self.ruid:
return self.obj
raise NotFoundError
def getPath(self, ruid):
if ruid==self.ruid:
return self.location
raise NotFoundError
class AbstractTestHubEvent(unittest.TestCase):
location = '/some/location'
hubid = 23
obj = object()
klass = None
def setUp(self):
self.hub = DummyObjectHub(self.hubid, self.obj, self.location)
self.event = self.klass(self.hub, self.hubid, self.location, self.obj)
def testGetHub(self):
self.assertEqual(self.event.hub, self.hub)
def testGetLocation(self):
self.assertEqual(self.event.location, self.location)
def testGetHubId(self):
# Test hubid
self.assertEqual(self.event.hubid, self.hubid)
def testGetObject(self):
self.assertEqual(self.event.object, self.obj)
class TestObjectRegisteredHubEvent(AbstractTestHubEvent):
klass = ObjectRegisteredHubEvent
class TestEmptyObjectRegisteredHubEvent(TestObjectRegisteredHubEvent):
def setUp(self):
self.hub = DummyObjectHub(self.hubid, self.obj, self.location)
self.event = self.klass(self.hub, self.hubid)
class TestObjectUnregisteredHubEvent(AbstractTestHubEvent):
klass = ObjectUnregisteredHubEvent
class TestEmptyObjectUnregisteredHubEvent(unittest.TestCase):
location = '/some/location'
hubid = 23
obj = object()
klass = None
klass = ObjectUnregisteredHubEvent
def testRaisesTypeError(self):
self.assertRaises(TypeError,
self.klass,
DummyObjectHub(self.hubid,
self.obj,
self.location),
self.hubid)
class TestObjectModifiedHubEvent(AbstractTestHubEvent):
klass = ObjectModifiedHubEvent
class TestEmptyObjectModifiedHubEvent(TestObjectModifiedHubEvent):
def setUp(self):
self.hub = DummyObjectHub(self.hubid, self.obj, self.location)
self.event = self.klass(self.hub, self.hubid)
class TestObjectMovedHubEvent(AbstractTestHubEvent):
fromLocation = '/old/location'
def setUp(self):
self.hub = DummyObjectHub(self.hubid, self.obj, self.location)
self.event = self.klass(self.hub,
self.hubid,
self.fromLocation,
self.location,
self.obj)
def testGetFromLocation(self):
# Test from location
self.assertEqual(self.event.fromLocation, self.fromLocation)
klass = ObjectMovedHubEvent
class TestEmptyObjectMovedHubEvent(TestObjectMovedHubEvent):
def setUp(self):
self.hub = DummyObjectHub(self.hubid, self.obj, self.location)
self.event = self.klass(self.hub,
self.hubid,
self.fromLocation)
class TestObjectRemovedHubEvent(AbstractTestHubEvent):
klass = ObjectRemovedHubEvent
class TestEmptyObjectRemovedHubEvent(TestEmptyObjectUnregisteredHubEvent):
klass = ObjectRemovedHubEvent
def test_suite():
return unittest.TestSuite((
unittest.makeSuite(TestObjectRegisteredHubEvent),
unittest.makeSuite(TestEmptyObjectRegisteredHubEvent),
unittest.makeSuite(TestObjectUnregisteredHubEvent),
unittest.makeSuite(TestEmptyObjectUnregisteredHubEvent),
unittest.makeSuite(TestObjectModifiedHubEvent),
unittest.makeSuite(TestEmptyObjectModifiedHubEvent),
unittest.makeSuite(TestObjectMovedHubEvent),
unittest.makeSuite(TestEmptyObjectMovedHubEvent),
unittest.makeSuite(TestObjectRemovedHubEvent),
unittest.makeSuite(TestEmptyObjectRemovedHubEvent),
))
if __name__=='__main__':
unittest.main(defaultTest='test_suite')
=== Added File Zope3/src/zope/app/hub/tests/test_objecthub.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
$Id: test_objecthub.py,v 1.1 2004/03/11 09:19:25 srichter Exp $
"""
import unittest
from zope.app.tests import ztapi
from zope.app import zapi
from zope.app.hub.tests.objecthubsetup import ObjectHubSetup
from zope.app.event.interfaces import IObjectModifiedEvent
from zope.app.container.interfaces import IObjectAddedEvent
from zope.app.container.interfaces import IObjectRemovedEvent
from zope.app.container.interfaces import IObjectMovedEvent
from zope.app.container.contained import ObjectAddedEvent, ObjectRemovedEvent
from zope.app.container.contained import ObjectMovedEvent
from zope.app.event.objectevent import ObjectModifiedEvent, ObjectCreatedEvent
from zope.app.hub.interfaces import ObjectHubError
from zope.app.hub.interfaces import IObjectRemovedHubEvent
from zope.app.hub.interfaces import IObjectModifiedHubEvent
from zope.app.hub.interfaces import IObjectMovedHubEvent
from zope.app.hub.interfaces import IObjectRegisteredHubEvent
from zope.app.hub.interfaces import IObjectUnregisteredHubEvent
from zope.app.hub import ObjectModifiedHubEvent, ObjectRemovedHubEvent
from zope.app.hub import ObjectMovedHubEvent, ObjectRegisteredHubEvent
from zope.app.hub import ObjectUnregisteredHubEvent
from zope.app.hub import canonicalSlash, userPath
from zope.app.interfaces.traversing import IContainmentRoot
from zope.app.location import Location
from zope.exceptions import NotFoundError
from zope.app.traversing import canonicalPath, traverse
from zope.interface import implements, directlyProvides
from zope.app.container.contained import Contained
# while these tests don't really test much of the placeful aspect of the
# object hub, they do at least test basic functionality.
# we'll need real tests of the placeful aspects, 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 = '/folder1/folder1_1'
# Don't test the HubEvent base class.
# See below for testing subclasses / subinterfaces
# klass = HubEvent
# interface = IHubEvent
def setUp(self):
ObjectHubSetup.setUp(self)
self.setUpLoggingSubscriber()
self.obj = zapi.traverse(self.rootFolder, self.location)
self.hub_event = self.klass(self.object_hub,
self.hubid,
self.location,
self.obj)
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 = ObjectRemovedHubEvent
class TransmitObjectModifiedHubEventTest(TransmitHubEventTest):
interface = IObjectModifiedHubEvent
klass = ObjectModifiedHubEvent
class TransmitObjectMovedHubEventTest(TransmitHubEventTest):
interface = IObjectMovedHubEvent
klass = ObjectMovedHubEvent
def setUp(self):
ObjectHubSetup.setUp(self)
self.setUpLoggingSubscriber()
self.obj = zapi.traverse(self.rootFolder, self.location)
self.hub_event = self.klass(
self.object_hub, self.hubid,
'/old/location', self.location, self.obj)
class TransmitObjectRegisteredHubEventTest(TransmitHubEventTest):
interface = IObjectRegisteredHubEvent
klass = ObjectRegisteredHubEvent
class TransmitObjectUnregisteredHubEventTest(TransmitHubEventTest):
interface = IObjectUnregisteredHubEvent
klass = ObjectUnregisteredHubEvent
class Folder(Location):
pass
class BasicHubTest(ObjectHubSetup, unittest.TestCase):
location = '/folder1/folder1_1'
new_location = '/folder2/folder2_1'
def setUp(self):
ObjectHubSetup.setUp(self)
self.setUpLoggingSubscriber()
self.obj = zapi.traverse(self.rootFolder, self.location)
self.setEvents()
def setEvents(self):
obj = self.obj
self.created_event = ObjectCreatedEvent(obj)
self.added_event = ObjectAddedEvent(obj)
newobj = zapi.traverse(self.rootFolder, self.new_location)
self.added_new_location_event = ObjectAddedEvent(newobj)
self.removed_event = ObjectRemovedEvent(obj)
self.modified_event = ObjectModifiedEvent(obj)
self.moved_event = ObjectMovedEvent(
obj,
obj.__parent__, obj.__name__,
newobj.__parent__, newobj.__name__,
)
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 TestSearchRegistrations(BasicHubTest):
locations = (
'/',
'/foo',
'/foo/baq',
'/foo/baq/baz',
'/foo/bar',
'/foo/bar/baz',
'/foo/bar2/baz',
'/foo/bar/baz2',
'/foo/bar/baz3',
'/foo/bas',
'/foo/bas/baz',
)
def setUp(self):
ObjectHubSetup.setUp(self)
def testSearchAll(self):
object_hub = self.object_hub
location_hubid = [(location,
object_hub.register(location))
for location in self.locations]
location_hubid.sort()
r = list(object_hub.iterRegistrations())
self.assertEqual(r, location_hubid)
def testSearchSome(self):
object_hub = self.object_hub
location_hubid= [(location,
object_hub.register(location))
for location in self.locations
if location.startswith('/foo/bar/')]
location_hubid.sort()
r = list(object_hub.iterRegistrations('/foo/bar'))
self.assertEqual(r, location_hubid)
def testIterObjectRegistrations(self):
class FakeObject:
def __init__(self, location):
self.location = location
def __str__(self):
return 'FakeObject at %s' % self.location
def __eq__(self, other):
return self.location == other.location
def fake_object_for_location(location):
return FakeObject(canonicalPath(location))
from zope.app.interfaces.traversing import ITraverser
from zope.app.traversing.adapters import Traverser
class DummyTraverser(Traverser):
implements(ITraverser)
def traverse(self, location, *args, **kw):
if location in TestSearchRegistrations.locations:
return fake_object_for_location(location)
else:
return Traverser.traverse(self, location, *args, **kw)
ztapi.provideAdapter(None, ITraverser, DummyTraverser)
object_hub = self.object_hub
location_hubid_object = [(location,
object_hub.register(location),
fake_object_for_location(location)
)
for location in self.locations]
location_hubid_object.sort()
r = [loc_id for loc_id in object_hub.iterObjectRegistrations()]
r.sort()
self.assertEqual(r, location_hubid_object)
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 TestObjectCreatedEvent(BasicHubTest):
def setUp(self):
ObjectHubSetup.setUp(self)
self.setUpRegistrationSubscriber()
self.setEvents()
def testLookingUpLocation(self):
hub = self.object_hub
event = self.created_event
location = None
hub.notify(event)
self.assertEqual(location_from_hub, location)
self.subscriber.verifyEventsReceived(self, [
(IObjectCreatedEvent, 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.getHubId, location)
self.subscriber.verifyEventsReceived(self, [])
class TestObjectAddedEvent(BasicHubTest):
def setUp(self):
ObjectHubSetup.setUp(self)
self.setUpRegistrationSubscriber()
self.obj = zapi.traverse(self.rootFolder, self.location)
self.setEvents()
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.getHubId(location)
# check that hub id is an int
self.failUnless(isinstance(hubid, int)) # int(hubid)
location_from_hub = hub.getPath(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.getHubId, 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.getPath, absent_hubid)
self.subscriber.verifyEventsReceived(self, [])
class TestObjectRemovedEvent(BasicHubTest):
def setUp(self):
ObjectHubSetup.setUp(self)
self.obj = zapi.traverse(self.rootFolder, self.location)
self.setUpRegistrationSubscriber()
self.setEvents()
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.getHubId(location)
# check that hubid is an int
self.failUnless(isinstance(hubid, int)) # int(hubid)
hub.notify(removed_event)
self.assertRaises(NotFoundError, hub.getHubId, location)
self.assertRaises(NotFoundError, hub.getPath, 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),
])
def testRemoveDescendants(self):
# Test that removing an object also removes its descendants
locations = (
'/',
'/1',
'/1/1_1',
'/1/1_1/1_1_1',
'/2',
'/2/2_1',
'/2/2_2' )
hub = self.object_hub
for path in locations:
hub.register(path)
removed_event = self.removed_event
removed_event.oldParent = '/'
removed_event.oldName = '1'
hub.notify(removed_event)
newLocations = [path for (path, oid) in hub.iterRegistrations()]
newLocations.sort()
self.assertEqual([
'/',
'/2',
'/2/2_1',
'/2/2_2'],
newLocations)
class TestObjectModifiedEvent(BasicHubTest):
def setUp(self):
ObjectHubSetup.setUp(self)
self.setUpRegistrationSubscriber()
self.obj = zapi.traverse(self.rootFolder, self.location)
self.setEvents()
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.getHubId(location)
# check that hubid is an int
self.failUnless(isinstance(hubid, int)) # int(hubid)
location_from_hub = hub.getPath(hubid)
self.assertEqual(location_from_hub, location)
hub.notify(modified_event)
hubid2 = hub.getHubId(location)
location_from_hub2 = hub.getPath(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.getHubId, location)
self.subscriber.verifyEventsReceived(self, [
(IObjectModifiedEvent, location),
])
class TestObjectMovedEvent(BasicHubTest):
def setUp(self):
ObjectHubSetup.setUp(self)
self.setUpRegistrationSubscriber()
self.obj = zapi.traverse(self.rootFolder, self.location)
self.setEvents()
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.getHubId(location)
self.subscriber.verifyEventsReceived(self, [
(IObjectAddedEvent, location),
(IObjectRegisteredHubEvent, hubid, location),
])
# simulate moving the object
self.obj.__parent__ = zapi.traverse(self.rootFolder, "folder2")
self.obj.__name__ = "folder2_1"
hub.notify(moved_event)
location_from_hub = hub.getPath(hubid)
self.assertEqual(location_from_hub, new_location)
self.assertRaises(NotFoundError, hub.getHubId, location)
hubid2 = hub.getHubId(new_location)
self.assertEqual(hubid2, hubid)
self.subscriber.verifyEventsReceived(self, [
(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.getHubId, location)
self.assertRaises(NotFoundError, hub.getHubId, new_location)
# simulate moving the object
self.obj.__parent__ = zapi.traverse(self.rootFolder, "folder2")
self.obj.__name__ = "folder2_1"
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.subscriber.verifyEventsReceived(self, [
(IObjectAddedEvent, location),
(IObjectRegisteredHubEvent, None, location),
(IObjectAddedEvent, new_location),
(IObjectRegisteredHubEvent, None, new_location),
])
# simulate moving the object
self.obj.__parent__ = zapi.traverse(self.rootFolder, "folder2")
self.obj.__name__ = "folder2_1"
self.assertRaises(ObjectHubError, hub.notify, moved_event)
self.subscriber.verifyEventsReceived(self, [
(IObjectMovedEvent, new_location),
])
def testRepathDescendantsOnRename(self):
# Test that the paths of descendants of renamed objects are updated
locations = (
'/',
'/1',
'/1/1_1',
'/1/1_1/1_1_1',
'/2',
'/2/2_1',
'/2/2_2' )
hub = self.object_hub
for path in locations:
hub.register(path)
moved_event = self.moved_event
moved_event.oldParent = '/'
moved_event.oldName = '1'
moved_event.newParent = '/'
moved_event.newName = '3'
hub.notify(moved_event)
newLocations = [path for (path, oid) in hub.iterRegistrations()]
newLocations.sort()
self.assertEqual([
'/',
'/2',
'/2/2_1',
'/2/2_2',
'/3',
'/3/1_1',
'/3/1_1/1_1_1'],
newLocations)
def testRepathDescendantsOnMove(self):
# Test that the paths of descendants of moved objects are updated
locations = (
'/',
'/1',
'/1/1_1',
'/1/1_1/1_1_1',
'/2',
'/2/2_1',
'/2/2_2' )
hub = self.object_hub
for path in locations:
hub.register(path)
moved_event = self.moved_event
moved_event.oldParent = '/'
moved_event.oldName = '1'
moved_event.newParent = '/2/2_1'
moved_event.newName = '1'
hub.notify(moved_event)
newLocations = [path for (path, oid) in hub.iterRegistrations()]
newLocations.sort()
self.assertEqual([
'/',
'/2',
'/2/2_1',
'/2/2_1/1',
'/2/2_1/1/1_1',
'/2/2_1/1/1_1/1_1_1',
'/2/2_2'],
newLocations)
class TestPathFunctions(BasicHubTest):
def testCanonicalSlash(self):
self.assertEqual(canonicalSlash('/'), '/')
self.assertEqual(canonicalSlash('/', 'bar'), '/bar/')
self.assertEqual(canonicalSlash('/foo'), '/foo/')
self.assertEqual(canonicalSlash('/foo', 'bar'), '/foo/bar/')
self.assertRaises(ValueError, canonicalSlash, '\\')
self.assertRaises(ValueError, canonicalSlash, '')
def testUserPath(self):
self.assertEqual(userPath('/'), '/')
self.assertEqual(userPath('/foo'), '/foo')
self.assertEqual(userPath('/foo/'), '/foo')
self.assertEqual(userPath('/foo/bar'), '/foo/bar')
self.assertEqual(userPath('/foo/bar/'), '/foo/bar')
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(TestSearchRegistrations),
unittest.makeSuite(TestObjectAddedEvent),
unittest.makeSuite(TestObjectRemovedEvent),
unittest.makeSuite(TestObjectModifiedEvent),
unittest.makeSuite(TestObjectMovedEvent),
unittest.makeSuite(TestPathFunctions),
))
if __name__=='__main__':
unittest.main(defaultTest='test_suite')
=== Added File Zope3/src/zope/app/hub/tests/test_registration.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.
##############################################################################
"""
$Id: test_registration.py,v 1.1 2004/03/11 09:19:25 srichter Exp $
"""
from unittest import makeSuite, main, TestCase
from zope.app import zapi
from zope.app.hub import Registration
from zope.app.services.tests.placefulsetup import PlacefulSetup
from zope.app.traversing import traverse, canonicalPath
from zope.app.container.contained import ObjectAddedEvent
from zope.component import getService
from zope.app.services.servicenames import EventPublication, HubIds
class TestRegistration(PlacefulSetup, TestCase):
def setUp(self):
PlacefulSetup.setUp(self, site=True)
self.createStandardServices()
r = Registration()
default = traverse(self.rootFolder, '++etc++site/default')
default['registrar'] = r
self.registrar = traverse(default, 'registrar')
self.hub = getService(self.rootFolder, HubIds)
self.events = getService(self.rootFolder, EventPublication)
def testSubscribeUnsubscribe(self):
r = self.registrar
self.assertEqual(r.isSubscribed(), False)
r.subscribe()
self.assertEqual(r.isSubscribed(), True)
self.assertRaises(RuntimeError, r.subscribe)
r.unsubscribe()
self.assertEqual(r.isSubscribed(), False)
self.assertRaises(RuntimeError, r.unsubscribe)
def testRegister(self):
self.registrar.subscribe()
self.assertEqual(self.hub.numRegistrations(), 0)
content = zapi.traverse(self.rootFolder, "folder1/folder1_1")
event = ObjectAddedEvent(content)
self.events.publish(event)
self.assertEqual(self.hub.numRegistrations(), 1)
def testRegisterExisting(self):
self.registrar.subscribe()
self.registrar.registerExisting()
# there are ten folders set up by PlacefulSetup that
# should get registered
self.assertEqual(self.hub.numRegistrations(), 10)
def test_suite():
return makeSuite(TestRegistration)
if __name__=='__main__':
main(defaultTest='test_suite')
More information about the Zope3-Checkins
mailing list