[Zope3-checkins] CVS: Zope3/lib/python/Zope/App/OFS/Services/LocalEventService/tests - testEventService.py:1.5.6.1 EventSetup.py:NONE
Jim Fulton
jim@zope.com
Sat, 30 Nov 2002 07:44:58 -0500
Update of /cvs-repository/Zope3/lib/python/Zope/App/OFS/Services/LocalEventService/tests
In directory cvs.zope.org:/tmp/cvs-serv30047/lib/python/Zope/App/OFS/Services/LocalEventService/tests
Modified Files:
Tag: Zope3-Bangalore-TTW-Branch
testEventService.py
Removed Files:
Tag: Zope3-Bangalore-TTW-Branch
EventSetup.py
Log Message:
Refactored the way TTW component registration is done. There are now
separate registry objects that abstract the machinery for registering
multiple conflicting configurations and deciding which, if any are
active. Also provided a new field and widget for the status
information.
Along the way, cleaned up and streamlined placeful testing
infrastructure a bit.
Now checking into branch. Will give file-by-file (or at least more
specific logs) when the changes are merged into the head.
=== Zope3/lib/python/Zope/App/OFS/Services/LocalEventService/tests/testEventService.py 1.5 => 1.5.6.1 ===
--- Zope3/lib/python/Zope/App/OFS/Services/LocalEventService/tests/testEventService.py:1.5 Thu Oct 3 16:53:21 2002
+++ Zope3/lib/python/Zope/App/OFS/Services/LocalEventService/tests/testEventService.py Sat Nov 30 07:44:27 2002
@@ -19,19 +19,23 @@
from unittest import TestCase, TestLoader, TextTestRunner
-from Zope.ComponentArchitecture import getService, getServiceManager
-
-from Zope.Proxy.ContextWrapper import ContextWrapper
+from Zope.App.OFS.Services.ServiceManager.tests.PlacefulSetup \
+ import PlacefulSetup
-from Zope.Exceptions import NotFoundError
+from Zope.App.OFS.Services.ServiceManager.ServiceManager \
+ import ServiceManager
from Zope.App.OFS.Services.LocalEventService.LocalEventService \
import LocalEventService
-from Zope.App.OFS.Services.ServiceManager.ServiceDirective \
- import ServiceDirective
+from Zope.App.OFS.Services.ServiceManager.ServiceConfiguration \
+ import ServiceConfiguration
-from Zope.App.Traversing import getPhysicalPathString
+from Zope.App.Traversing import getPhysicalPathString, traverse
+
+from Zope.ComponentArchitecture import getService, getServiceManager
+
+from Zope.Exceptions import NotFoundError
from Zope.Event import subscribe, unsubscribe, listSubscriptions, publishEvent
from Zope.Event.tests.subscriber import DummySubscriber, DummyFilter
@@ -44,8 +48,35 @@
from Zope.Event.GlobalEventService import GlobalEventService
from Zope.Event.IEvent import IEvent
from Zope.Event.ISubscriptionAware import ISubscriptionAware
+from Zope.App.OFS.Services.ConfigurationInterfaces \
+ import Active, Unregistered, Registered
+
+from Zope.Proxy.ContextWrapper import ContextWrapper
+
+class EventSetup(PlacefulSetup):
+
+ def setUp(self):
+ PlacefulSetup.setUp(self)
+ self.buildFolders()
+ self.createEventService()
+
+ def createEventService(self, path=None):
-from EventSetup import EventSetup
+ folder = self.rootFolder
+ if path is not None:
+ folder = traverse(folder, path)
+
+ if not folder.hasServiceManager():
+ folder.setServiceManager(ServiceManager())
+
+ 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
class UnpromotingLocalEventService(LocalEventService):
@@ -79,9 +110,9 @@
class EventServiceTests(EventSetup, TestCase):
def _createNestedServices(self):
- self.createEventService(self.folder1)
- self.createEventService(self.folder1_1)
- self.createEventService(self.folder1_1_1)
+ self.createEventService('folder1')
+ self.createEventService('folder1/folder1_1')
+ self.createEventService('folder1/folder1_1/folder1_1_1')
def _createSubscribers(self):
self.rootFolder.setObject("rootFolderSubscriber", DummySubscriber())
@@ -106,8 +137,11 @@
def testListSubscriptions1(self):
"a non-subscribed subscriber gets an empty array"
self._createSubscribers()
- self.assertEqual([], self.rootFolder.getServiceManager().getService(
- "Events").listSubscriptions(self.rootFolderSubscriber))
+
+ events = getService(self.rootFolder, "Events")
+
+ self.assertEqual(events.listSubscriptions(self.rootFolderSubscriber),
+ [])
def testListSubscriptions2(self):
"one subscription"
@@ -383,19 +417,23 @@
def _createAlternateService(self, service):
self._createSubscribers()
self.folder2.setObject("folder2Subscriber", DummySubscriber())
- self.folder2Subscriber=ContextWrapper(
+ self.folder2Subscriber = ContextWrapper(
self.folder2["folder2Subscriber"],
self.folder2,
name="folder2Subscriber")
+
if not self.folder2.hasServiceManager():
- self.createServiceManager(self.folder2)
- sm=getServiceManager(self.folder2) # wrapped now
- sm.Packages['default'].setObject("myEventService", service)
+ self.folder2.setServiceManager(ServiceManager())
+
+ sm = traverse(self.rootFolder, 'folder2/++etc++Services')
+ default = traverse(sm, 'Packages/default')
+
+ default.setObject("myEventService", service)
path = "%s/Packages/default/myEventService" % getPhysicalPathString(sm)
- directive = ServiceDirective("Events", path)
- sm.Packages['default'].setObject("myEventServiceDir", directive)
- sm.bindService(directive)
+ configuration = ServiceConfiguration("Events", path)
+ default['configure'].setObject("myEventServiceDir", configuration)
+ traverse(default, 'configure/1').status = Active
subscribe(
self.rootFolderSubscriber,
@@ -439,9 +477,11 @@
self.folder1_1Subscriber,
event_type=IObjectAddedEvent
)
- sm=getServiceManager(self.folder1)
- directive = sm.getDirectives("Events")[0]
- sm.unbindService(directive)
+
+ sm = traverse(self.rootFolder, "folder1/++etc++Services")
+ configuration = sm.queryConfigurations("Events").active()
+ configuration.status = Registered
+
publishEvent(self.rootFolder, ObjectAddedEvent(None, '/foo'))
self.assertEqual(self.folder1Subscriber.notified, 0)
self.assertEqual(self.folder1_1Subscriber.notified, 1)
@@ -455,9 +495,11 @@
publishEvent(self.rootFolder, ObjectAddedEvent(None, '/foo'))
self.assertEqual(self.folder2Subscriber.notified, 0)
self.assertEqual(self.rootFolderSubscriber.notified, 1)
- sm=getServiceManager(self.folder2)
- directive = sm.getDirectives("Events")[0]
- sm.unbindService(directive) # make sure it doesn't raise any errors
+
+ sm = traverse(self.rootFolder, "folder2/++etc++Services")
+ configuration = sm.queryConfigurations("Events").active()
+ # make sure it doesn't raise any errors
+ configuration.status = Registered
def testSubscriptionAwareInteraction(self):
sub = DummySubscriptionAwareSubscriber()
=== Removed File Zope3/lib/python/Zope/App/OFS/Services/LocalEventService/tests/EventSetup.py ===