[Zope3-checkins] CVS: Zope3/lib/python/Zope/App/OFS/Services/CachingService/tests - testCacheConfiguration.py:1.1 testCachingService.py:1.6
Marius Gedminas
mgedmin@codeworks.lt
Thu, 12 Dec 2002 10:28:48 -0500
Update of /cvs-repository/Zope3/lib/python/Zope/App/OFS/Services/CachingService/tests
In directory cvs.zope.org:/tmp/cvs-serv24277/lib/python/Zope/App/OFS/Services/CachingService/tests
Modified Files:
testCachingService.py
Added Files:
testCacheConfiguration.py
Log Message:
Caching service now uses the new configuration infrastructure
Added invalidateAll to Zope.App.Caching.ICache
=== Added File Zope3/lib/python/Zope/App/OFS/Services/CachingService/tests/testCacheConfiguration.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.
#
##############################################################################
"""Unit test for CacheConfiguration.
$Id: testCacheConfiguration.py,v 1.1 2002/12/12 15:28:17 mgedmin Exp $
"""
from unittest import TestCase, TestSuite, main, makeSuite
from Interface import Interface
from Zope.App.OFS.Services.CachingService.CacheConfiguration \
import CacheConfiguration
from Zope.App.OFS.Services.ServiceManager.tests.PlacefulSetup \
import PlacefulSetup
from Zope.App.Traversing import traverse
from Zope.App.OFS.Services.ServiceManager.ServiceManager \
import ServiceManager
from Zope.App.OFS.Container.ZopeContainerAdapter import ZopeContainerAdapter
from Zope.App.OFS.Services.ConfigurationInterfaces \
import Active, Unregistered, Registered
from Zope.App.Caching.ICache import ICache
from Zope.App.DependencyFramework.IDependable import IDependable
from Zope.App.Caching.ICachingService import ICachingService
from Zope.App.OFS.Services.ConfigurationInterfaces import IConfigurable
from Zope.App.OFS.Services.Configuration import ConfigurationRegistry
from Zope.App.OFS.Services.ServiceManager.ServiceConfiguration \
import ServiceConfiguration
from Zope.ContextWrapper import ContextMethod
from Zope.Proxy.ContextWrapper import ContextWrapper
from Zope.Event.IObjectEvent import IObjectModifiedEvent
class DependableStub:
__implements__ = IDependable
def addDependent(self, location):
pass
def removeDependent(self, location):
pass
def dependents(self):
pass
class TestCache(DependableStub):
__implements__ = ICache, IDependable
def invalidateAll(self):
self.invalidated = True
class CachingServiceStub(DependableStub):
__implements__ = ICachingService, IConfigurable, IDependable
def __init__(self):
self.bindings = {}
self.subscriptions = {}
def queryConfigurationsFor(self, cfg, default=None):
return self.queryConfigurations(cfg.name)
queryConfigurationsFor = ContextMethod(queryConfigurationsFor)
def queryConfigurations(self, name, default=None):
registry = self.bindings.get(name, default)
return ContextWrapper(registry, self)
queryConfigurations = ContextMethod(queryConfigurations)
def createConfigurationsFor(self, cfg):
return self.createConfigurations(cfg.name)
createConfigurationsFor = ContextMethod(createConfigurationsFor)
def createConfigurations(self, name):
try:
registry = self.bindings[name]
except KeyError:
self.bindings[name] = registry = ConfigurationRegistry()
return ContextWrapper(registry, self)
createConfigurations = ContextMethod(createConfigurations)
def subscribe(self, obj, event):
self.subscriptions.setdefault(obj, []).append(event)
def unsubscribe(self, obj, event):
self.subscriptions.setdefault(obj, []).remove(event)
def listSubscriptions(self, obj):
return self.subscriptions.get(obj, [])
class TestConnectionConfiguration(PlacefulSetup, TestCase):
def setUp(self):
PlacefulSetup.setUp(self)
self.buildFolders()
self.rootFolder.setServiceManager(ServiceManager())
self.default = traverse(self.rootFolder,
'++etc++Services/Packages/default')
self.default.setObject('cch', TestCache())
self.cch = traverse(self.default, 'cch')
self.cm = ZopeContainerAdapter(traverse(self.default, "configure"))
self.cm.setObject('', CacheConfiguration('cache_name',
'/++etc++Services/Packages/default/cch'))
self.config = traverse(self.default, 'configure/1')
self.default.setObject('cache_srv', CachingServiceStub())
self.service = traverse(self.default, 'cache_srv')
self.cm.setObject('', ServiceConfiguration('Caching',
'/++etc++Services/Packages/default/cache_srv'))
traverse(self.default, 'configure/2').status = Active
def tearDown(self):
PlacefulSetup.tearDown(self)
def test_getComponent(self):
# This should be already tested by ComponentConfiguration tests, but
# let's doublecheck
self.assertEqual(self.config.getComponent(), self.cch)
def test_status(self):
self.assertEqual(self.config.status, Unregistered)
self.config.status = Active
self.assertEqual(self.config.status, Active)
cr = self.service.queryConfigurations('cache_name')
self.assertEqual(cr.active(), self.config)
def test_activated(self):
self.config.activated()
self.assertEqual(self.service.listSubscriptions(self.cch),
[IObjectModifiedEvent])
def test_deactivated(self):
self.service.subscribe(self.cch, IObjectModifiedEvent)
self.cch.invalidated = False
self.config.deactivated()
self.assertEqual(self.service.listSubscriptions(self.cch), [])
self.failIf(not self.cch.invalidated,
"deactivation should call invalidateAll")
def test_suite():
return makeSuite(TestConnectionConfiguration)
if __name__=='__main__':
main(defaultTest='test_suite')
=== Zope3/lib/python/Zope/App/OFS/Services/CachingService/tests/testCachingService.py 1.5 => 1.6 ===
--- Zope3/lib/python/Zope/App/OFS/Services/CachingService/tests/testCachingService.py:1.5 Wed Dec 11 04:04:08 2002
+++ Zope3/lib/python/Zope/App/OFS/Services/CachingService/tests/testCachingService.py Thu Dec 12 10:28:17 2002
@@ -17,19 +17,16 @@
"""
from unittest import TestCase, TestSuite, main, makeSuite
-from Interface.Verify import verifyClass, verifyObject
+from Interface.Verify import verifyObject
from Zope.App.Caching.ICache import ICache
from Zope.App.Caching.ICachingService import ICachingService
-from Zope.App.OFS.Services.ConfigurationInterfaces import Active
+from Zope.App.OFS.Services.CachingService.CacheConfiguration \
+ import CacheConfiguration
+from Zope.App.OFS.Services.ConfigurationInterfaces import Active, Registered
from Zope.App.OFS.Services.LocalEventService.tests.EventSetup import EventSetup
from Zope.App.OFS.Services.ServiceManager.ServiceConfiguration \
import ServiceConfiguration
from Zope.App.Traversing import getPhysicalPathString, traverse
-from Zope.App.tests.PlacelessSetup import PlacelessSetup
-from Zope.ComponentArchitecture import getServiceManager, getService
-from Zope.ComponentArchitecture.GlobalServiceManager \
- import serviceManager as sm
-from Zope.Event.IObjectEvent import IObjectModifiedEvent
def sort(list):
@@ -41,15 +38,14 @@
__implements__ = ICache
+ def __init__(self, name):
+ self.name = name
-class CachingServiceSetup(EventSetup):
+ def __repr__(self):
+ return "CacheStub(%r)" % self.name
- def setUp(self):
- EventSetup.setUp(self)
- global_service_manager = getServiceManager(None)
- global_service_manager.defineService("CachingService", ICachingService)
- self.createCachingService()
+class CachingServiceSetup(EventSetup):
def createCachingService(self, path=None):
from Zope.App.OFS.Services.CachingService.CachingService \
@@ -62,28 +58,43 @@
if not folder.hasServiceManager():
self.createServiceManager(folder)
- sm = traverse(folder, '++etc++Services')
- default = traverse(sm, 'Packages/default')
- default.setObject("myCachingService", CachingService())
+ default = traverse(folder, '++etc++Services/Packages/default')
+ key = default.setObject("myCachingService", CachingService())
+ service = traverse(default, key)
- path = "%s/Packages/default/myCachingService" % getPhysicalPathString(sm)
- configuration = ServiceConfiguration("CachingService", path)
-
- configure = traverse(default, 'configure')
+ path = getPhysicalPathString(service)
+ configuration = ServiceConfiguration("Caching", path)
configure = traverse(default, 'configure')
- key = configure.setObject("myCachingServiceDir", configuration)
+ key = configure.setObject(None, configuration)
traverse(configure, key).status = Active
+ return service
+
+ def addCache(self, name, cache=None, cname=None, status=Active, folder=''):
+ if not cache: cache = CacheStub("%s/%s" % (folder, name))
+ if not cname: cname = name
+ default = traverse(self.rootFolder,
+ folder +'/++etc++Services/Packages/default')
+ key = default.setObject(cname, cache)
+ cache = traverse(default, key)
+ configure = traverse(default, 'configure')
+ key = configure.setObject(None, CacheConfiguration(name,
+ getPhysicalPathString(cache)))
+ traverse(configure, key).status = status
+ return cache
+
class TestCachingService(CachingServiceSetup, TestCase):
def setUp(self):
CachingServiceSetup.setUp(self)
- self.cache1 = CacheStub()
- self.cache2 = CacheStub()
- self.service = getService(self.rootFolder, "CachingService")
- self.service.setObject('cache1', self.cache1)
- self.service.setObject('cache2', self.cache2)
+ self.service = self.createCachingService()
+ self.cache1 = self.addCache('cache1')
+ self.cache2 = self.addCache('cache2')
+ self.cache3 = self.addCache('cache3', status=Registered)
+ self.service_f1 = self.createCachingService('folder1')
+ self.cache1_f1 = self.addCache('cache1', folder='folder1')
+ self.cache4_f1 = self.addCache('cache4', folder='folder1')
def test_interface(self):
from Zope.App.OFS.Services.CachingService.CachingService \
@@ -92,45 +103,38 @@
verifyObject(ICachingService, self.service)
def test_getCache(self):
- self.assertEqual(self.cache1,
- self.service.getCache('cache1'))
+ self.assertEqual(self.cache1, self.service.getCache('cache1'))
+ self.assertEqual(self.cache2, self.service.getCache('cache2'))
self.assertRaises(KeyError, self.service.getCache, 'cache3')
+ self.assertRaises(KeyError, self.service.getCache, 'cache4')
+
+ self.assertEqual(self.cache1_f1, self.service_f1.getCache('cache1'))
+ self.assertEqual(self.cache2, self.service_f1.getCache('cache2'))
+ self.assertRaises(KeyError, self.service_f1.getCache, 'cache3')
+ self.assertEqual(self.cache4_f1, self.service_f1.getCache('cache4'))
+ self.assertRaises(KeyError, self.service_f1.getCache, 'cache5')
def test_queryCache(self):
- self.assertEqual(self.cache1,
- self.service.queryCache('cache1'))
- self.assertEqual(None,
- self.service.queryCache('cache3'))
- self.assertEqual('Error',
- self.service.queryCache('cache3', 'Error'))
+ self.assertEqual(self.cache1, self.service.queryCache('cache1'))
+ self.assertEqual(self.cache2, self.service.queryCache('cache2'))
+ self.assertEqual(None, self.service.queryCache('cache3'))
+ self.assertEqual('XX', self.service.queryCache('cache4', 'XX'))
+ self.assertEqual(None, self.service.queryCache('cache3'))
+ self.assertEqual('YY', self.service.queryCache('cache4', 'YY'))
+
+ self.assertEqual(self.cache1_f1, self.service_f1.queryCache('cache1'))
+ self.assertEqual(self.cache2, self.service_f1.queryCache('cache2'))
+ self.assertEqual(None, self.service_f1.queryCache('cache3'))
+ self.assertEqual('ZZ', self.service_f1.queryCache('cache3', 'ZZ'))
+ self.assertEqual(self.cache4_f1, self.service_f1.queryCache('cache4'))
+ self.assertEqual(None, self.service_f1.queryCache('cache5'))
+ self.assertEqual('12', self.service_f1.queryCache('cache5', '12'))
def test_getAvailableCaches(self):
self.assertEqual(['cache1', 'cache2'],
sort(self.service.getAvailableCaches()))
-
- def test_isAddable(self):
- self.assertEqual(1, self.service.isAddable(ICache))
- self.assertEqual(0, self.service.isAddable(ICachingService))
-
- def test_setObject(self):
- # setObject called in setUp... Ugh...
- self.assertEqual(self.service.listSubscriptions(self.cache1),
- [(IObjectModifiedEvent, None)])
- self.assertEqual(self.service.listSubscriptions(self.cache2),
- [(IObjectModifiedEvent, None)])
-
- def test__delitem__(self):
- self.assertEqual(self.service.listSubscriptions(self.cache1),
- [(IObjectModifiedEvent, None)])
- self.assertEqual(self.service.listSubscriptions(self.cache2),
- [(IObjectModifiedEvent, None)])
- del self.service['cache1']
- self.assertEqual(self.service.listSubscriptions(self.cache1), [])
- self.assertEqual(self.service.listSubscriptions(self.cache2),
- [(IObjectModifiedEvent, None)])
-
- del self.service['cache2']
- self.assertEqual(self.service.listSubscriptions(self.cache2), [])
+ self.assertEqual(['cache1', 'cache2', 'cache4'],
+ sort(self.service_f1.getAvailableCaches()))
def test_suite():