[Zope3-checkins] CVS: Zope3/src/zope/app/services/tests - eventsetup.py:1.4 objecthubsetup.py:1.3 placefulsetup.py:1.5 test_cachingservice.py:1.3 test_eventservice.py:1.3 test_hookedhubevent.py:1.3 test_objecthub.py:1.5
Steve Alexander
steve@cat-box.net
Mon, 30 Dec 2002 09:03:50 -0500
Update of /cvs-repository/Zope3/src/zope/app/services/tests
In directory cvs.zope.org:/tmp/cvs-serv23914/src/zope/app/services/tests
Modified Files:
eventsetup.py objecthubsetup.py placefulsetup.py
test_cachingservice.py test_eventservice.py
test_hookedhubevent.py test_objecthub.py
Log Message:
Large refactoring of the event service.
=== Zope3/src/zope/app/services/tests/eventsetup.py 1.3 => 1.4 ===
--- Zope3/src/zope/app/services/tests/eventsetup.py:1.3 Sat Dec 28 12:49:32 2002
+++ Zope3/src/zope/app/services/tests/eventsetup.py Mon Dec 30 09:03:17 2002
@@ -2,51 +2,34 @@
#
# 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 zope.app.services.tests.placefulsetup import PlacefulSetup
from zope.app.services.service import ServiceManager, ServiceConfiguration
-from zope.app.services.event import LocalEventService
+from zope.app.services.event import EventService
from zope.app.traversing import getPhysicalPathString, traverse
from zope.app.interfaces.services.configuration import Active
+from zope.app.interfaces.services.event import ISubscriptionService
class EventSetup(PlacefulSetup):
-
+
def setUp(self):
PlacefulSetup.setUp(self)
self.buildFolders()
- self.createEventService()
+ self.createStandardServices()
def createServiceManager(self, folder):
folder.setServiceManager(ServiceManager())
- def createEventService(self, path=None):
-
- folder = self.rootFolder
- if path is not None:
- folder = traverse(folder, path)
-
- if not folder.hasServiceManager():
- self.createServiceManager(folder)
-
- sm = traverse(folder, '++etc++Services')
- default = traverse(sm, 'Packages/default')
- default.setObject("myEventService", LocalEventService())
-
- path = "%s/Packages/default/myEventService" % getPhysicalPathString(sm)
- configuration = ServiceConfiguration("Events", path)
- default['configure'].setObject("myEventServiceDir", configuration)
- traverse(default, 'configure/1').status = Active
=== Zope3/src/zope/app/services/tests/objecthubsetup.py 1.2 => 1.3 ===
--- Zope3/src/zope/app/services/tests/objecthubsetup.py:1.2 Wed Dec 25 09:13:20 2002
+++ Zope3/src/zope/app/services/tests/objecthubsetup.py Mon Dec 30 09:03:17 2002
@@ -18,25 +18,94 @@
"""
from zope.app.services.tests.eventsetup import EventSetup
-from zope.component import getServiceManager
+from zope.component import getServiceManager, getService
from zope.app.services.service import ServiceConfiguration
from zope.app.traversing import getPhysicalPathString, traverse
+from zope.app.traversing import locationAsTuple
from zope.app.services.hub import ObjectHub
+from zope.app.interfaces.event import IObjectAddedEvent
from zope.app.interfaces.services.configuration import Active
+from zope.app.interfaces.event import ISubscriber
+
+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 = locationAsTuple(location)
+ testcase.assert_(interface.isImplementedBy(event),
+ 'Interface %s' % interface.getName())
+ testcase.assertEqual(event.location, 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)
+ if IObjectAddedEvent.isImplementedBy(event):
+ self.hub.register(event.location)
class ObjectHubSetup(EventSetup):
+ def setUpRegistrationSubscriber(self):
+ subscriber = RegistrationSubscriber(self.object_hub)
+ self.rootFolder.setObject('registration_subscriber', subscriber)
+ self.subscriber = traverse(self.rootFolder, 'registration_subscriber')
+ self.object_hub.subscribe(self.subscriber)
+
+ def setUpLoggingSubscriber(self):
+ subscriber = LoggingSubscriber()
+ self.rootFolder.setObject('logging_subscriber', subscriber)
+ self.subscriber = traverse(self.rootFolder, 'logging_subscriber')
+ self.object_hub.subscribe(self.subscriber)
+
def setUp(self):
EventSetup.setUp(self)
-
- from zope.app.interfaces.services.hub import IObjectHub
- global_service_manager = getServiceManager(None)
- global_service_manager.defineService("ObjectHub", IObjectHub)
- self.createObjectHub()
-
- def createObjectHub(self, path=None):
- folder = self.rootFolder
+ self.object_hub = getService(self.rootFolder, 'HubIds')
+ #from zope.app.interfaces.services.hub import IObjectHub
+ #global_service_manager = getServiceManager(None)
+ #global_service_manager.defineService("HubIds", IObjectHub)
+
+ def createObjectHub(self, path):
+ raise "HELL!"
+ #folder = self.rootFolder
if path is not None:
folder = traverse(folder, path)
@@ -48,7 +117,7 @@
default.setObject("myObjectHub", ObjectHub())
path = "%s/Packages/default/myObjectHub" % getPhysicalPathString(sm)
- configuration = ServiceConfiguration("ObjectHub", path)
+ configuration = ServiceConfiguration("HubIds", path)
configure = traverse(default, 'configure')
key = configure.setObject("myObjectHubDir", configuration)
=== Zope3/src/zope/app/services/tests/placefulsetup.py 1.4 => 1.5 ===
--- Zope3/src/zope/app/services/tests/placefulsetup.py:1.4 Sat Dec 28 12:49:32 2002
+++ Zope3/src/zope/app/services/tests/placefulsetup.py Mon Dec 30 09:03:17 2002
@@ -28,15 +28,15 @@
from zope.app.interfaces.content.folder import IRootFolder
from zope.app.interfaces.traversing import IContainmentRoot
from zope.app.interfaces.traversing import IPhysicallyLocatable
-from zope.app.interfaces.traversing import ITraverser
-from zope.app.interfaces.traversing import ITraversable
+from zope.app.interfaces.traversing import ITraverser, ITraversable
from zope.app.tests.placelesssetup import PlacelessSetup
+from zope.app.traversing.namespace import etc, provideNamespaceHandler
from zope.app.traversing.adapters import DefaultTraversable
-from zope.app.traversing.namespace import etc
-from zope.app.traversing.namespace import provideNamespaceHandler
from zope.app.traversing.adapters import WrapperPhysicallyLocatable
from zope.app.traversing.adapters import Traverser, RootPhysicallyLocatable
-
+from zope.app.traversing import traverse, getPhysicalPathString
+from zope.app.services.service import ServiceManager, ServiceConfiguration
+from zope.app.interfaces.services.configuration import Active
class PlacefulSetup(PlacelessSetup):
@@ -84,10 +84,9 @@
# | \ | |
# folder1_1_1 folder1_1_2 folder1_2_1 folder2_1_1
from zope.app.content.folder import Folder
- from zope.app.content.folder import RootFolder
from zope.proxy.context import ContextWrapper
# top
- self.rootFolder = RootFolder()
+ self.createRootFolder()
# level 1
self.folder1 = Folder()
self.rootFolder.setObject("folder1", self.folder1)
@@ -128,10 +127,83 @@
self.folder2_1_1 = ContextWrapper(self.folder2_1_1, self.folder2_1,
name = "folder2_1_1")
- def createServiceManager(self, folder = None):
+ def createServiceManager(self, folder=None):
if folder is None:
folder = self.rootFolder
from zope.app.services.tests.servicemanager \
import TestingServiceManager
-
folder.setServiceManager(TestingServiceManager())
+
+ def createRootFolder(self):
+ from zope.app.content.folder import RootFolder
+ self.rootFolder = RootFolder()
+
+ def getObjectHub(self):
+ from zope.app.services.hub import ObjectHub
+ return ObjectHub()
+
+ def createEventService(self, folder_path):
+ """Create an event service in 'folder', and configure it for
+ Events and Subscription services."""
+ folder = traverse(self.rootFolder, folder_path)
+ if not folder.hasServiceManager():
+ folder.setServiceManager(ServiceManager())
+ sm = traverse(folder, '++etc++Services')
+ default = traverse(sm, 'Packages/default')
+ service_name = 'anEventService'
+ from zope.app.services.event import EventService
+ default.setObject(service_name, EventService())
+
+ path = "%s/%s" % (getPhysicalPathString(default), service_name)
+ configuration = ServiceConfiguration("Events", path)
+ default['configure'].setObject(
+ "%sEventsDir" % service_name, configuration)
+ traverse(default, 'configure/1').status = Active
+
+ configuration = ServiceConfiguration("Subscription", path)
+ default['configure'].setObject(
+ "%sSubscriptionServiceDir" % service_name, configuration)
+ traverse(default, 'configure/2').status = Active
+
+ def createStandardServices(self):
+ '''Create a bunch of standard placeful services'''
+ if not hasattr(self, 'rootFolder'):
+ self.createRootFolder()
+ root = self.rootFolder
+ if root.hasServiceManager():
+ raise RuntimeError('ServiceManager already exists, so cannot '
+ 'create standard services')
+ root.setServiceManager(ServiceManager())
+ from zope.component import getServiceManager
+ defineService = getServiceManager(None).defineService
+
+ from zope.app.interfaces.services.hub import IObjectHub
+ from zope.app.interfaces.services.event import ISubscriptionService
+ from zope.app.services.event import EventService
+ defineService("Subscription", ISubscriptionService)
+
+ # Events service already defined by
+ # zope.app.events.tests.PlacelessSetup
+
+ defineService("HubIds", IObjectHub)
+
+ sm = traverse(root, '++etc++Services')
+ default = traverse(sm, 'Packages/default')
+ default.setObject("myEventService", EventService())
+ default.setObject("myObjectHub", self.getObjectHub())
+
+ path = "%s/Packages/default/myEventService" % getPhysicalPathString(sm)
+ configuration = ServiceConfiguration("Events", path)
+ default['configure'].setObject("myEventServiceDir", configuration)
+ traverse(default, 'configure/1').status = Active
+
+ configuration = ServiceConfiguration("Subscription", path)
+ default['configure'].setObject(
+ "mySubscriptionServiceDir", configuration)
+ traverse(default, 'configure/2').status = Active
+
+ path = "%s/Packages/default/myObjectHub" % getPhysicalPathString(sm)
+ configuration = ServiceConfiguration("HubIds", path)
+ default['configure'].setObject("myHubIdsServiceDir", configuration)
+ traverse(default, 'configure/3').status = Active
+
=== Zope3/src/zope/app/services/tests/test_cachingservice.py 1.2 => 1.3 ===
--- Zope3/src/zope/app/services/tests/test_cachingservice.py:1.2 Wed Dec 25 09:13:20 2002
+++ Zope3/src/zope/app/services/tests/test_cachingservice.py Mon Dec 30 09:03:17 2002
@@ -46,8 +46,7 @@
class CachingServiceSetup(EventSetup):
def createCachingService(self, path=None):
- from zope.app.services.cache \
- import CachingService
+ from zope.app.services.cache import CachingService
folder = self.rootFolder
if path is not None:
@@ -55,7 +54,7 @@
if not folder.hasServiceManager():
self.createServiceManager(folder)
-
+
default = traverse(folder, '++etc++Services/Packages/default')
key = default.setObject("myCachingService", CachingService())
service = traverse(default, key)
=== Zope3/src/zope/app/services/tests/test_eventservice.py 1.2 => 1.3 === (774/874 lines abridged)
--- Zope3/src/zope/app/services/tests/test_eventservice.py:1.2 Wed Dec 25 09:13:20 2002
+++ Zope3/src/zope/app/services/tests/test_eventservice.py Mon Dec 30 09:03:17 2002
@@ -18,42 +18,32 @@
"""
from unittest import TestCase, TestLoader, TextTestRunner
-
-from zope.app.services.service import ServiceManager
-
-from zope.app.services.event import LocalEventService
-
-from zope.app.services.service import ServiceConfiguration
-
+from zope.interface import Interface
+from zope.app.services.service import ServiceManager, ServiceConfiguration
+from zope.component import getServiceManager
+from zope.app.services.event import EventService
from zope.app.traversing import getPhysicalPathString, traverse
-
from zope.exceptions import NotFoundError
-
-from zope.event import subscribe, unsubscribe, listSubscriptions, publish
-from zope.event import getEventService
-from zope.event.tests.subscriber import DummySubscriber, DummyFilter
-from zope.app.interfaces.event import IObjectEvent
-from zope.app.interfaces.event import IObjectAddedEvent
-from zope.app.interfaces.event import IObjectRemovedEvent
-from zope.app.interfaces.event import IObjectModifiedEvent
+from zope.app.services.event import subscribe, unsubscribe
+from zope.app.services.event import listSubscriptions, getSubscriptionService
+from zope.app.event import getEventService, publish
+from zope.app.event.tests.subscriber import DummySubscriber, DummyFilter
+from zope.app.interfaces.event import IObjectEvent, IObjectModifiedEvent
+from zope.app.interfaces.event import IObjectRemovedEvent, IObjectAddedEvent
from zope.app.event.objectevent import ObjectAddedEvent, ObjectModifiedEvent
-from zope.app.event.objectevent import ObjectAddedEvent
-from zope.app.event.globaleventservice import GlobalEventService
-from zope.interfaces.event import IEvent
-from zope.interfaces.event import ISubscriptionAware
+from zope.app.event.globalservice import GlobalEventPublisher
+from zope.app.interfaces.event import IEvent, ISubscribingAware
from zope.app.interfaces.services.configuration import Active
from zope.app.interfaces.services.configuration import Unregistered
from zope.app.interfaces.services.configuration import Registered
-
from zope.proxy.context import ContextWrapper
-
from zope.app.services.tests.eventsetup import EventSetup
-class UnpromotingLocalEventService(LocalEventService):
[-=- -=- -=- 774 lines omitted -=- -=- -=-]
self.assertEqual(self.folder2Subscriber.notified, 0)
self.assertEqual(self.rootFolderSubscriber.notified, 1)
sm = traverse(self.rootFolder, "folder2/++etc++Services")
+ configuration = sm.queryConfigurations("Subscription").active()
+ # make sure it doesn't raise any errors
+ configuration.status = Registered
configuration = sm.queryConfigurations("Events").active()
# make sure it doesn't raise any errors
configuration.status = Registered
@@ -464,7 +869,7 @@
self.rootFolder.setObject(
"mySubscriber",
DummySubscriptionAwareSubscriber())
- self.mySubscriber=ContextWrapper(
+ self.mySubscriber = ContextWrapper(
self.rootFolder["mySubscriber"],
self.rootFolder,
name="mySubscriber")
@@ -477,6 +882,9 @@
self.mySubscriber.subscribable,
getEventService(self.rootFolder))
self.assertEqual(
+ self.mySubscriber.subscribable,
+ getSubscriptionService(self.rootFolder))
+ self.assertEqual(
self.mySubscriber.event_type,
IObjectAddedEvent)
self.assertEqual(
@@ -490,6 +898,9 @@
self.mySubscriber.un_subscribable,
getEventService(self.rootFolder))
self.assertEqual(
+ self.mySubscriber.un_subscribable,
+ getSubscriptionService(self.rootFolder))
+ self.assertEqual(
self.mySubscriber.un_event_type,
IObjectAddedEvent)
self.assertEqual(
@@ -498,8 +909,8 @@
def test_suite():
- loader=TestLoader()
- return loader.loadTestsFromTestCase(EventServiceTests)
+ loader = TestLoader()
+ return loader.loadTestsFromTestCase(TestEventPublisher)
if __name__=='__main__':
TextTestRunner().run(test_suite())
=== Zope3/src/zope/app/services/tests/test_hookedhubevent.py 1.2 => 1.3 ===
--- Zope3/src/zope/app/services/tests/test_hookedhubevent.py:1.2 Wed Dec 25 09:13:20 2002
+++ Zope3/src/zope/app/services/tests/test_hookedhubevent.py Mon Dec 30 09:03:17 2002
@@ -37,7 +37,6 @@
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 = getPhysicalPath(self.obj)
@@ -66,7 +65,6 @@
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 = getPhysicalPath(self.obj)
@@ -80,7 +78,6 @@
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 = getPhysicalPath(self.obj)
@@ -94,7 +91,6 @@
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 = getPhysicalPath(self.obj)
@@ -106,7 +102,6 @@
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 = getPhysicalPath(self.obj)
@@ -126,7 +121,6 @@
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 = getPhysicalPath(self.obj)
=== Zope3/src/zope/app/services/tests/test_objecthub.py 1.4 => 1.5 ===
--- Zope3/src/zope/app/services/tests/test_objecthub.py:1.4 Sat Dec 28 12:49:32 2002
+++ Zope3/src/zope/app/services/tests/test_objecthub.py Mon Dec 30 09:03:17 2002
@@ -26,14 +26,12 @@
from zope.app.event.objectevent\
import ObjectAddedEvent, ObjectModifiedEvent, ObjectRemovedEvent,\
ObjectMovedEvent, ObjectCreatedEvent
-from zope.interfaces.event import ISubscriber
from zope.app.interfaces.services.hub import ObjectHubError
from zope.app.interfaces.services.hub import \
IObjectRemovedHubEvent, IObjectModifiedHubEvent, \
IObjectMovedHubEvent, IObjectRegisteredHubEvent, \
IObjectUnregisteredHubEvent
-import zope.app.services.hub
from zope.app.services.hub \
import ObjectModifiedHubEvent, ObjectRemovedHubEvent, \
ObjectMovedHubEvent, ObjectRegisteredHubEvent, \
@@ -42,9 +40,7 @@
from zope.exceptions import NotFoundError
from types import StringTypes
-from zope.app.traversing import locationAsUnicode, locationAsTuple
-
-from zope.component import getService, getServiceManager
+from zope.app.traversing import locationAsUnicode, locationAsTuple, traverse
# while these tests don't really test much of the placeful aspect of the
# object hub, they do at least test basic functionality.
@@ -53,61 +49,6 @@
# test happens simply by virtue of the testHubEvent module in this
# directory
-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 = locationAsTuple(location)
- testcase.assert_(interface.isImplementedBy(event),
- 'Interface %s' % interface.getName())
- testcase.assertEqual(event.location, 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)
- if IObjectAddedEvent.isImplementedBy(event):
- self.hub.register(event.location)
-
-
class TransmitHubEventTest(ObjectHubSetup, unittest.TestCase):
hubid = 23
location = locationAsTuple('/foo/bar')
@@ -119,15 +60,12 @@
def setUp(self):
ObjectHubSetup.setUp(self)
- self.object_hub = getService(self.rootFolder, "ObjectHub")
+ self.setUpLoggingSubscriber()
self.hub_event = self.klass(self.object_hub,
self.hubid,
self.location,
self.obj)
- 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)
@@ -150,12 +88,10 @@
def setUp(self):
ObjectHubSetup.setUp(self)
- self.object_hub = getService(self.rootFolder, "ObjectHub")
+ self.setUpLoggingSubscriber()
self.hub_event = self.klass(
self.object_hub, self.hubid,
locationAsTuple('/old/location'), self.location, self.obj)
- self.subscriber = LoggingSubscriber()
- self.object_hub.subscribe(self.subscriber)
class TransmitObjectRegisteredHubEventTest(TransmitHubEventTest):
interface = IObjectRegisteredHubEvent
@@ -172,10 +108,8 @@
def setUp(self):
ObjectHubSetup.setUp(self)
- self.object_hub = getService(self.rootFolder, "ObjectHub")
+ self.setUpLoggingSubscriber()
self.setEvents()
- self.subscriber = LoggingSubscriber()
- self.object_hub.subscribe(self.subscriber)
def setEvents(self):
self.created_event = ObjectCreatedEvent(object())
@@ -236,6 +170,10 @@
'/foo/bas',
'/foo/bas/baz',
)
+
+ def setUp(self):
+ ObjectHubSetup.setUp(self)
+
def testSearchAll(self):
object_hub = self.object_hub
location_hubid = [(locationAsTuple(location),
@@ -266,8 +204,16 @@
self.object_hub.register, u'/foo/\uffffstuff')
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 'object at %s' % locationAsUnicode(location)
+ return FakeObject(locationAsUnicode(location))
from zope.app.interfaces.traversing import ITraverser
class DummyTraverser:
@@ -277,12 +223,10 @@
def traverse(self, location):
return fake_object_for_location(location)
- from zope.component.adapter\
- import provideAdapter
+ from zope.component.adapter import provideAdapter
provideAdapter(None, ITraverser, DummyTraverser)
object_hub = self.object_hub
-
location_hubid_object = [(locationAsTuple(location),
object_hub.register(location),
fake_object_for_location(location)
@@ -312,10 +256,8 @@
class TestObjectCreatedEvent(BasicHubTest):
def setUp(self):
ObjectHubSetup.setUp(self)
- self.object_hub = getService(self.rootFolder, "ObjectHub")
+ self.setUpRegistrationSubscriber()
self.setEvents()
- self.subscriber = RegistrationSubscriber(self.object_hub)
- self.object_hub.subscribe(self.subscriber)
def testLookingUpLocation(self):
hub = self.object_hub
@@ -346,10 +288,8 @@
class TestObjectAddedEvent(BasicHubTest):
def setUp(self):
ObjectHubSetup.setUp(self)
- self.object_hub = getService(self.rootFolder, "ObjectHub")
+ self.setUpRegistrationSubscriber()
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
@@ -405,10 +345,8 @@
class TestObjectRemovedEvent(BasicHubTest):
def setUp(self):
ObjectHubSetup.setUp(self)
- self.object_hub = getService(self.rootFolder, "ObjectHub")
+ self.setUpRegistrationSubscriber()
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.
@@ -458,10 +396,8 @@
class TestObjectModifiedEvent(BasicHubTest):
def setUp(self):
ObjectHubSetup.setUp(self)
- self.object_hub = getService(self.rootFolder, "ObjectHub")
+ self.setUpRegistrationSubscriber()
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.
@@ -517,10 +453,8 @@
class TestObjectMovedEvent(BasicHubTest):
def setUp(self):
ObjectHubSetup.setUp(self)
- self.object_hub = getService(self.rootFolder, "ObjectHub")
+ self.setUpRegistrationSubscriber()
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.