[Zope3-checkins] CVS: Zope3/src/zope/app/services/tests - __init__.py:1.1.2.1 eventsetup.py:1.1.2.1 objecthubsetup.py:1.1.2.1 placefulsetup.py:1.1.2.1 sample1.py:1.1.2.1 sample2.py:1.1.2.1 test_adapter.py:1.1.2.1 test_auth.py:1.1.2.1 test_cacheconfiguration.py:1.1.2.1 test_cachingservice.py:1.1.2.1 test_configurationmanager.py:1.1.2.1 test_configurationregistry.py:1.1.2.1 test_configurations.py:1.1.2.1 test_configurationstatusproperty.py:1.1.2.1 test_connectionconfiguration.py:1.1.2.1 test_connectionservice.py:1.1.2.1 test_errorreportingservice.py:1.1.2.1 test_eventservice.py:1.1.2.1 test_field.py:1.1.2.1 test_hookedhubevent.py:1.1.2.1 test_hubevent.py:1.1.2.1 test_nameconfigurable.py:1.1.2.1 test_objecthub.py:1.1.2.1 test_principalannotation.py:1.1.2.1 test_roleservice.py:1.1.2.1 test_serviceconfiguration.py:1.1.2.1 test_servicemanager.py:1.1.2.1 test_user.py:1.1.2.1 test_view.py:1.1.2.1 test_viewpackage.py:1.1.2.1 testingconfigurationregistry.py:1.1.2.1 testingiconfigurable.py:1.1!
.2.1 testingservicemanager.py:1.1.2.1
Jim Fulton
jim@zope.com
Mon, 23 Dec 2002 14:32:28 -0500
Update of /cvs-repository/Zope3/src/zope/app/services/tests
In directory cvs.zope.org:/tmp/cvs-serv19908/zope/app/services/tests
Added Files:
Tag: NameGeddon-branch
__init__.py eventsetup.py objecthubsetup.py placefulsetup.py
sample1.py sample2.py test_adapter.py test_auth.py
test_cacheconfiguration.py test_cachingservice.py
test_configurationmanager.py test_configurationregistry.py
test_configurations.py test_configurationstatusproperty.py
test_connectionconfiguration.py test_connectionservice.py
test_errorreportingservice.py test_eventservice.py
test_field.py test_hookedhubevent.py test_hubevent.py
test_nameconfigurable.py test_objecthub.py
test_principalannotation.py test_roleservice.py
test_serviceconfiguration.py test_servicemanager.py
test_user.py test_view.py test_viewpackage.py
testingconfigurationregistry.py testingiconfigurable.py
testingservicemanager.py
Log Message:
Initial renaming before debugging
=== Added File Zope3/src/zope/app/services/tests/__init__.py ===
#
# This file is necessary to make this directory a package.
=== Added File Zope3/src/zope/app/services/tests/eventsetup.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: eventsetup.py,v 1.1.2.1 2002/12/23 19:32:23 jim Exp $
"""
from zope.app.services.tests.placefulsetup \
import PlacefulSetup
from zope.app.services.service \
import ServiceManager
from zope.app.services.event \
import LocalEventService
from zope.app.services.service \
import ServiceConfiguration
from Zope.App.Traversing import getPhysicalPathString, traverse
from zope.app.interfaces.services.configuration import Active
class EventSetup(PlacefulSetup):
def setUp(self):
PlacefulSetup.setUp(self)
self.buildFolders()
self.createEventService()
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
=== Added File Zope3/src/zope/app/services/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.2.1 2002/12/23 19:32:23 jim Exp $
"""
from zope.app.services.tests.eventsetup import \
EventSetup
from zope.component import getServiceManager
from zope.app.services.service \
import ServiceConfiguration
from Zope.App.Traversing import getPhysicalPathString, traverse
from zope.app.services.hub import ObjectHub
from zope.app.interfaces.services.configuration import Active
class ObjectHubSetup(EventSetup):
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
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("myObjectHub", ObjectHub())
path = "%s/Packages/default/myObjectHub" % getPhysicalPathString(sm)
configuration = ServiceConfiguration("ObjectHub", path)
configure = traverse(default, 'configure')
key = configure.setObject("myObjectHubDir", configuration)
traverse(configure, key).status = Active
=== Added File Zope3/src/zope/app/services/tests/placefulsetup.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: placefulsetup.py,v 1.1.2.1 2002/12/23 19:32:23 jim Exp $
"""
from zope.app.tests.placelesssetup import PlacelessSetup
from Zope import ComponentArchitecture as CA
from Zope.App.ComponentArchitecture import hooks
from zope.component.adapter import provideAdapter
from Zope.App.Traversing import getPhysicalPathString
from zope.app.traversing.defaulttraversable import DefaultTraversable
from zope.app.traversing.etcnamespace import etc
from zope.app.interfaces.traversing.containmentroot import IContainmentRoot
from zope.app.interfaces.traversing.physicallylocatable import IPhysicallyLocatable
from zope.app.interfaces.traversing.traverser import ITraverser
from zope.app.interfaces.traversing.traversable import ITraversable
from zope.app.traversing.namespaces import provideNamespaceHandler
from zope.app.traversing.physicallocationadapters \
import WrapperPhysicallyLocatable, RootPhysicallyLocatable
from zope.app.traversing.traverser import Traverser
from zope.app.content.folder import IRootFolder
from zope.app.interfaces.container import ISimpleReadContainer
from zope.app.container.traversal import ContainerTraversable
from zope.component.view import provideView
from zope.publisher.interfaces.browser import IBrowserPresentation
from zope.app.browser.absoluteurl \
import SiteAbsoluteURL, AbsoluteURL
class PlacefulSetup(PlacelessSetup):
def setUp(self):
PlacelessSetup.setUp(self)
# set up placeful hooks, saving originals for tearDown
self.__old_getServiceManager_hook = CA.getServiceManager_hook
CA.getServiceManager_hook = hooks.getServiceManager_hook
self.setUpTraversal()
def setUpTraversal(self):
provideAdapter(None, ITraverser, Traverser)
provideAdapter(None, ITraversable, DefaultTraversable)
provideAdapter(
ISimpleReadContainer, ITraversable, ContainerTraversable)
provideAdapter(
None, IPhysicallyLocatable, WrapperPhysicallyLocatable)
provideAdapter(
IContainmentRoot, IPhysicallyLocatable, RootPhysicallyLocatable)
# set up etc namespace
provideNamespaceHandler("etc", etc)
provideView(None, "absolute_url", IBrowserPresentation,
AbsoluteURL)
provideView(IRootFolder, "absolute_url", IBrowserPresentation,
SiteAbsoluteURL)
def tearDown(self):
# clean up folders and placeful service managers and services too?
CA.getServiceManager_hook = self.__old_getServiceManager_hook
PlacelessSetup.tearDown(self)
def buildFolders(self):
# set up a reasonably complex folder structure
#
# ____________ rootFolder ____________
# / \
# folder1 __________________ folder2
# | \ |
# folder1_1 ____ folder1_2 folder2_1
# | \ | |
# 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.context import ContextWrapper
# top
self.rootFolder = RootFolder()
# level 1
self.folder1 = Folder()
self.rootFolder.setObject("folder1", self.folder1)
self.folder1 = ContextWrapper(self.folder1, self.rootFolder,
name = "folder1")
self.folder2 = Folder()
self.rootFolder.setObject("folder2", self.folder2)
self.folder2 = ContextWrapper(self.folder2, self.rootFolder,
name = "folder2")
# level 2
self.folder1_1 = Folder()
self.folder1.setObject("folder1_1", self.folder1_1)
self.folder1_1 = ContextWrapper(self.folder1_1, self.folder1,
name = "folder1_1")
self.folder1_2 = Folder()
self.folder1.setObject("folder1_2", self.folder1_2)
self.folder1_2 = ContextWrapper(self.folder1_2, self.folder1,
name = "folder1_2")
self.folder2_1 = Folder()
self.folder2.setObject("folder2_1", self.folder2_1)
self.folder2_1 = ContextWrapper(self.folder2_1, self.folder2,
name = "folder2_1")
# level 3
self.folder1_1_1 = Folder()
self.folder1_1.setObject("folder1_1_1", self.folder1_1_1)
self.folder1_1_1 = ContextWrapper(self.folder1_1_1, self.folder1_1,
name = "folder1_1_1")
self.folder1_1_2 = Folder()
self.folder1_1.setObject("folder1_1_2", self.folder1_1_2)
self.folder1_1_2 = ContextWrapper(self.folder1_1_2, self.folder1_1,
name = "folder1_1_2")
self.folder1_2_1 = Folder()
self.folder1_2.setObject("folder1_2_1", self.folder1_2_1)
self.folder1_2_1 = ContextWrapper(self.folder1_2_1, self.folder1_2,
name = "folder1_2_1")
self.folder2_1_1 = Folder()
self.folder2_1.setObject("folder2_1_1", self.folder2_1_1)
self.folder2_1_1 = ContextWrapper(self.folder2_1_1, self.folder2_1,
name = "folder2_1_1")
def createServiceManager(self, folder = None):
if folder is None:
folder = self.rootFolder
from zope.app.services.tests.testingservicemanager \
import TestingServiceManager
folder.setServiceManager(TestingServiceManager())
=== Added File Zope3/src/zope/app/services/tests/sample1.py ===
x = 'sample 1'
=== Added File Zope3/src/zope/app/services/tests/sample2.py ===
y = 'sample 2'
=== Added File Zope3/src/zope/app/services/tests/test_adapter.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.
#
##############################################################################
"""Test the adapter module
$Id: test_adapter.py,v 1.1.2.1 2002/12/23 19:32:23 jim Exp $
"""
from unittest import TestCase, TestSuite, main, makeSuite
from zope.app.services.tests.testingiconfigurable import TestingIConfigurable
from zope.app.services.adapter import AdapterService
from zope.interface import Interface
from zope.proxy.context.context import ContextWrapper
from zope.component.exceptions import ComponentLookupError
from zope.app.services.tests.placefulsetup \
import PlacefulSetup
from zope.app.services.service import ServiceManager
from zope.app.services.adapter import AdapterConfiguration
from zope.app.content.folder import RootFolder
from zope.component import getServiceManager
from Zope.App.Traversing import traverse
from zope.component.interfaces import IServiceService
from zope.component.adapter import provideAdapter
class I1(Interface):
pass
class I1E(I1):
pass
class I2B(Interface):
pass
class I2(I2B):
pass
class I3(Interface):
pass
class I4(Interface):
pass
class Configuration:
forInterface = I1
providedInterface = I2
def getAdapter(self, object):
return self.factory(object)
def activated(self): pass
def deactivated(self): pass
class C: pass
class A:
def __init__(self, object):
self.context = object
class TestAdapterService(PlacefulSetup, TestingIConfigurable, TestCase):
def setUp(self):
PlacefulSetup.setUp(self)
self.buildFolders()
self.rootFolder.setServiceManager(ServiceManager())
self._service = ContextWrapper(AdapterService(), self.rootFolder)
def test_implements_IAdapterService(self):
from zope.component.interfaces import IAdapterService
from zope.interface.verify import verifyObject
verifyObject(IAdapterService, self._service)
def createTestingConfigurable(self):
return ContextWrapper(AdapterService(), C())
def createTestingConfiguration(self):
return Configuration()
def test_queryAdapter_no_adapter(self):
service = self._service
class O:
__implements__ = I1
o = O()
self.assertEqual(service.queryAdapter(o, I2), None)
self.assertEqual(service.queryAdapter(o, I2, 42), 42)
self.assertEqual(service.queryAdapter(o, I1), o)
self.assertEqual(service.queryAdapter(o, I1, 42), o)
def test_getAdapter_no_adapter(self):
service = self._service
class O:
__implements__ = I1
o = O()
self.assertRaises(ComponentLookupError, service.getAdapter, O(), I2)
self.assertEqual(service.getAdapter(o, I1), o)
def test_queryAdapter_and_getAdapter(self):
service = self._service
sm = traverse(self.rootFolder, '++etc++Services')
configure = traverse(sm, 'Packages/default/configure')
configuration = Configuration()
configure.setObject('', configuration)
configuration = traverse(configure, '1')
class O:
__implements__ = I1
configuration.factory = A
registry = service.createConfigurationsFor(configuration)
registry.register(configuration)
registry.activate(configuration)
o = O()
for m in 'queryAdapter', 'getAdapter':
for r in I1, I1E:
for p in I2B, I2:
o = O()
o.__implements__ = r
adapter = getattr(service, m)(o, p)
self.assertEqual(adapter.__class__, A)
self.assertEqual(adapter.context, o)
self.assertEqual(getattr(service, m)(o, I1), o)
self.assertEqual(service.queryAdapter(o, I3), None)
self.assertEqual(service.queryAdapter(o, I3, 42), 42)
self.assertRaises(ComponentLookupError, service.getAdapter, O(), I3)
def test_queryAdapter_delegation(self):
service = self._service
self.buildFolders()
self.rootFolder.setServiceManager(ServiceManager())
sm = traverse(self.rootFolder, '++etc++Services')
configure = traverse(sm, 'Packages/default/configure')
configuration = Configuration()
configure.setObject('', configuration)
configuration = traverse(configure, '1')
class O:
__implements__ = I1
configuration.factory = A
registry = service.createConfigurationsFor(configuration)
registry.register(configuration)
registry.activate(configuration)
o = O()
class A2(A): pass
provideAdapter(I1, I4, A2)
adapter = service.queryAdapter(o, I4)
self.assertEqual(adapter.__class__, A2)
self.assertEqual(adapter.context, o)
def test_queryAdapter_delegation_w_no_adapters_locally(self):
service = self._service
class O:
__implements__ = I1
o = O()
class A2(A): pass
provideAdapter(I1, I4, A2)
adapter = service.queryAdapter(o, I4)
self.assertEqual(adapter.__class__, A2)
self.assertEqual(adapter.context, o)
def test_getRegisteredMatching(self):
self.test_queryAdapter_and_getAdapter()
registry = self._service.queryConfigurations(I1, I2, '')
for args in ((), (I1E, ), (None, I2), (I1E, I2), ):
r = self._service.getRegisteredMatching(*args)
self.assertEqual(list(r), [(I1, I2, registry)])
class PhonyServiceManager:
__implements__ = IServiceService
def resolve(self, name):
if name == 'Foo.Bar.A':
return A
class TestAdapterConfiguration(PlacefulSetup, TestCase):
def setUp(self):
PlacefulSetup.setUp(self)
rootFolder = RootFolder()
rootFolder.setServiceManager(PhonyServiceManager())
self.configuration = ContextWrapper(
AdapterConfiguration(I1, I2, "Foo.Bar.A"),
rootFolder,
)
def test_getAdapter(self):
c = C()
adapter = self.configuration.getAdapter(c)
self.assertEqual(adapter.__class__, A)
self.assertEqual(adapter.context, c)
self.assertEqual(self.configuration.forInterface, I1)
self.assertEqual(self.configuration.providedInterface, I2)
def test_suite():
return TestSuite((
makeSuite(TestAdapterService),
makeSuite(TestAdapterConfiguration),
))
if __name__=='__main__':
main(defaultTest='test_suite')
=== Added File Zope3/src/zope/app/services/tests/test_auth.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_auth.py,v 1.1.2.1 2002/12/23 19:32:23 jim Exp $
"""
from unittest import TestCase, TestSuite, main, makeSuite
from zope.app.services.auth \
import AuthenticationService, DuplicateLogin, DuplicateId
from zope.app.services.auth import User
from zope.app.interfaces.services.auth import IUser
from zope.exceptions import NotFoundError
from zope.publisher.interfaces.http import IHTTPCredentials
from zope.app.services.tests.placefulsetup \
import PlacefulSetup
class Request:
__implements__ = IHTTPCredentials
def __init__(self, lpw):
self.__lpw = lpw
def _authUserPW(self):
return self.__lpw
challenge = None
def unauthorized(self, challenge):
self.challenge = challenge
class AuthServiceTest(TestCase, PlacefulSetup):
def setUp(self):
PlacefulSetup.setUp(self)
from zope.component import getService
from zope.app.security.basicauthadapter import BasicAuthAdapter
from zope.app.interfaces.security import ILoginPassword
getService(None, "Adapters").provideAdapter(
IHTTPCredentials, ILoginPassword, BasicAuthAdapter)
auth = AuthenticationService()
auth.setObject('srichter', User('srichter', 'Stephan', 'Richter',
'srichter', 'hello'))
auth.setObject('jim', User('jim', 'Jim', 'Foulton',
'jim', 'hello2'))
auth.setObject('stevea', User('stevea', 'Steve', 'Alexander',
'stevea', 'hello3'))
self._auth = auth
def testGetPrincipalByLogin(self):
auth = self._auth
self.assertEqual(auth['srichter'],
auth.getPrincipalByLogin('srichter'))
def testAuthenticate(self):
auth = self._auth
req = Request(('srichter', 'hello'))
pid = auth.authenticate(req).getId()
self.assertEquals(pid, 'srichter')
req = Request(('srichter', 'hello2'))
p = auth.authenticate(req)
self.assertEquals(p, None)
req = Request(('doesnotexit', 'hello'))
principal = auth.authenticate(req)
self.assertEquals(principal, None)
def testUnauthenticatedPrincipal(self):
auth = self._auth
self.assertEqual(None, auth.unauthenticatedPrincipal())
def testUnauthorized(self):
auth = self._auth
request = Request(None)
auth.unauthorized(auth.unauthenticatedPrincipal(), request)
self.assertEquals(request.challenge, "basic realm=zope")
request = Request(None)
auth.unauthorized(None, request)
self.assertEquals(request.challenge, "basic realm=zope")
request = Request(None)
auth.unauthorized("srichter", request)
self.assertEquals(request.challenge, None)
def testGetPrincipal(self):
auth = self._auth
self.assertEqual(auth['srichter'], auth.getPrincipal('srichter'))
self.assertEqual(None, auth.getPrincipal('srichter2'))
def testGetPrincipals(self):
auth = self._auth
self.assertEqual([auth['srichter']], auth.getPrincipals('srichter'))
def test_suite():
return makeSuite(AuthServiceTest)
if __name__=='__main__':
main(defaultTest='test_suite')
=== Added File Zope3/src/zope/app/services/tests/test_cacheconfiguration.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: test_cacheconfiguration.py,v 1.1.2.1 2002/12/23 19:32:23 jim Exp $
"""
from unittest import TestCase, main, makeSuite
from zope.app.services.cache \
import CacheConfiguration
from zope.app.services.tests.placefulsetup \
import PlacefulSetup
from Zope.App.Traversing import traverse
from zope.app.services.service \
import ServiceManager
from zope.app.container.zope import ZopeContainerAdapter
from zope.app.interfaces.services.configuration \
import Active, Unregistered
from zope.app.interfaces.cache.cache import ICache
from zope.app.interfaces.dependable import IDependable
from zope.app.interfaces.cache.cache import ICachingService
from zope.app.interfaces.services.configuration import IConfigurable
from zope.app.services.configuration import ConfigurationRegistry
from zope.app.services.service \
import ServiceConfiguration
from zope.proxy.context import ContextMethod
from zope.proxy.context.context import ContextWrapper
from zope.app.interfaces.event 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')
=== Added File Zope3/src/zope/app/services/tests/test_cachingservice.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.
#
##############################################################################
"""CachingService tests.
$Id: test_cachingservice.py,v 1.1.2.1 2002/12/23 19:32:23 jim Exp $
"""
from unittest import TestCase, TestSuite, main, makeSuite
from zope.interface.verify import verifyObject
from zope.app.interfaces.cache.cache import ICache
from zope.app.interfaces.cache.cache import ICachingService
from zope.app.services.cache \
import CacheConfiguration
from zope.app.interfaces.services.configuration import Active, Registered
from zope.app.services.tests.eventsetup import EventSetup
from zope.app.services.service \
import ServiceConfiguration
from Zope.App.Traversing import getPhysicalPathString, traverse
def sort(list):
list.sort()
return list
class CacheStub:
__implements__ = ICache
def __init__(self, name):
self.name = name
def __repr__(self):
return "CacheStub(%r)" % self.name
class CachingServiceSetup(EventSetup):
def createCachingService(self, path=None):
from zope.app.services.cache \
import CachingService
folder = self.rootFolder
if path is not None:
folder = traverse(folder, path)
if not folder.hasServiceManager():
self.createServiceManager(folder)
default = traverse(folder, '++etc++Services/Packages/default')
key = default.setObject("myCachingService", CachingService())
service = traverse(default, key)
path = getPhysicalPathString(service)
configuration = ServiceConfiguration("Caching", path)
configure = traverse(default, 'configure')
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.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.services.cache \
import ILocalCachingService
verifyObject(ILocalCachingService, self.service)
verifyObject(ICachingService, self.service)
def test_getCache(self):
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(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()))
self.assertEqual(['cache1', 'cache2', 'cache4'],
sort(self.service_f1.getAvailableCaches()))
def test_suite():
return TestSuite((
makeSuite(TestCachingService),
))
if __name__=='__main__':
main(defaultTest='test_suite')
=== Added File Zope3/src/zope/app/services/tests/test_configurationmanager.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_configurationmanager.py,v 1.1.2.1 2002/12/23 19:32:23 jim Exp $
"""
from zope.app.services.configurationmanager \
import ConfigurationManager
from zope.app.interfaces.services.configurationmanager \
import IConfigurationManager
from unittest import TestCase, TestSuite, main, makeSuite
from zope.interface.common.tests.basemapping import BaseTestIEnumerableMapping
from zope.interface.verify import verifyObject
from zope.app.interfaces.container import IDeleteNotifiable
from zope.app.interfaces.container import IContainer
from zope.app.interfaces.container import IZopeContainer
from zope.app.container.zope import ZopeContainerAdapter
from zope.app.tests.placelesssetup import PlacelessSetup
from zope.component.adapter import provideAdapter
class Undeletable:
__implements__ = IDeleteNotifiable
def manage_beforeDelete(self, object, container):
self.was_called = 1
class Test(BaseTestIEnumerableMapping, PlacelessSetup, TestCase):
"""Testing for Configuration Manager """
def setUp(self):
PlacelessSetup.setUp(self)
self.__manager = manager = ConfigurationManager()
for l in 'abcdefghijklmnop':
manager.setObject('', l)
del manager['8']
del manager['10']
def test_implements_IConfigurationManager(self):
verifyObject(IConfigurationManager, self.__manager)
def _IEnumerableMapping__stateDict(self):
# Hook needed by BaseTestIEnumerableMapping
# also, effectively test setObject and __delitem__.
return {
'1': 'a', '2': 'b', '3': 'c', '4': 'd', '5': 'e',
'6': 'f', '7': 'g', '9': 'i', '11': 'k', '12': 'l',
'13': 'm', '14': 'n', '15': 'o', '16': 'p',
}
def _IEnumerableMapping__sample(self):
# Hook needed by BaseTestIEnumerableMapping
# also, effectively test setObject and __delitem__.
return self.__manager
def _IEnumerableMapping__absentKeys(self):
# Hook needed by BaseTestIEnumerableMapping
# also, effectively test setObject and __delitem__.
return ['-1', '8', '10', '17', '100', '10000']
#########################################################
# Move Top
def test_moveTop_nothing(self):
self.__manager.moveTop([])
self.assertEqual(
list(self.__manager.keys()),
['1', '2', '3', '4', '5', '6', '7', '9',
'11', '12', '13', '14', '15', '16'],
)
# Make sure we still have thye right items
self.test_items()
def test_moveTop_1_no_effect(self):
self.__manager.moveTop(['1'])
self.assertEqual(
list(self.__manager.keys()),
['1', '2', '3', '4', '5', '6', '7', '9',
'11', '12', '13', '14', '15', '16'],
)
# Make sure we still have thye right items
self.test_items()
def test_moveTop_many_no_effect(self):
self.__manager.moveTop(['1', '88', '3', '2', '99'])
self.assertEqual(
list(self.__manager.keys()),
['1', '2', '3', '4', '5', '6', '7', '9',
'11', '12', '13', '14', '15', '16'],
)
# Make sure we still have thye right items
self.test_items()
def test_moveTop_1(self):
self.__manager.moveTop(['3'])
self.assertEqual(
list(self.__manager.keys()),
['3', '1', '2', '4', '5', '6', '7', '9',
'11', '12', '13', '14', '15', '16'],
)
# Make sure we still have thye right items
self.test_items()
def test_moveTop_many(self):
self.__manager.moveTop(['1', '3', '88', '4', '11', '15', '16', '99'])
self.assertEqual(
list(self.__manager.keys()),
['1', '3', '4', '11', '15', '16', '2', '5', '6', '7', '9',
'12', '13', '14'],
)
# Make sure we still have thye right items
self.test_items()
def test_moveTop_one_element_container(self):
manager = ConfigurationManager()
manager.setObject('', 'a')
manager.moveTop(['1'])
self.assertEqual(list(manager.items()), [('1', 'a')])
#########################################################
# Move Bottom
def test_moveBottom_nothing(self):
self.__manager.moveBottom([])
self.assertEqual(
list(self.__manager.keys()),
['1', '2', '3', '4', '5', '6', '7', '9',
'11', '12', '13', '14', '15', '16'],
)
# Make sure we still have thye right items
self.test_items()
def test_moveBottom_1_no_effect(self):
self.__manager.moveBottom(['16'])
self.assertEqual(
list(self.__manager.keys()),
['1', '2', '3', '4', '5', '6', '7', '9',
'11', '12', '13', '14', '15', '16'],
)
# Make sure we still have thye right items
self.test_items()
def test_moveBottom_many_no_effect(self):
self.__manager.moveBottom(['14', '88', '16', '15', '99'])
self.assertEqual(
list(self.__manager.keys()),
['1', '2', '3', '4', '5', '6', '7', '9',
'11', '12', '13', '14', '15', '16'],
)
# Make sure we still have thye right items
self.test_items()
def test_moveBottom_1(self):
self.__manager.moveBottom(['3'])
self.assertEqual(
list(self.__manager.keys()),
['1', '2', '4', '5', '6', '7', '9',
'11', '12', '13', '14', '15', '16', '3'],
)
# Make sure we still have thye right items
self.test_items()
def test_moveBottom_many(self):
self.__manager.moveBottom(
['1', '3', '88', '4', '11', '16', '15', '99'])
self.assertEqual(
list(self.__manager.keys()),
['2', '5', '6', '7', '9',
'12', '13', '14', '1', '3', '4', '11', '15', '16'],
)
# Make sure we still have thye right items
self.test_items()
def test_moveBottom_one_element_container(self):
manager = ConfigurationManager()
manager.setObject('', 'a')
manager.moveBottom(['1'])
self.assertEqual(list(manager.items()), [('1', 'a')])
#########################################################
# Move Up
def test_moveUp_nothing(self):
self.__manager.moveUp([])
self.assertEqual(
list(self.__manager.keys()),
['1', '2', '3', '4', '5', '6', '7', '9',
'11', '12', '13', '14', '15', '16'],
)
# Make sure we still have thye right items
self.test_items()
def test_moveUp_1_no_effect(self):
self.__manager.moveUp(['1'])
self.assertEqual(
list(self.__manager.keys()),
['1', '2', '3', '4', '5', '6', '7', '9',
'11', '12', '13', '14', '15', '16'],
)
# Make sure we still have thye right items
self.test_items()
def test_moveUp_many_no_effect(self):
self.__manager.moveUp(['1', '88', '3', '2', '99'])
self.assertEqual(
list(self.__manager.keys()),
['1', '2', '3', '4', '5', '6', '7', '9',
'11', '12', '13', '14', '15', '16'],
)
# Make sure we still have thye right items
self.test_items()
def test_moveUp_1(self):
self.__manager.moveUp(['3'])
self.assertEqual(
list(self.__manager.keys()),
['1', '3', '2', '4', '5', '6', '7', '9',
'11', '12', '13', '14', '15', '16'],
)
# Make sure we still have thye right items
self.test_items()
def test_moveUp_many(self):
self.__manager.moveUp(
['1', '3', '88', '4', '11', '16', '15', '99'])
self.assertEqual(
list(self.__manager.keys()),
['1', '3', '4', '2', '5', '6', '7', '11', '9',
'12', '13', '15', '16', '14'],
)
# Make sure we still have thye right items
self.test_items()
def test_moveUp_one_element_container(self):
manager = ConfigurationManager()
manager.setObject('', 'a')
manager.moveUp(['1'])
self.assertEqual(list(manager.items()), [('1', 'a')])
#########################################################
# Move Down
def test_moveDown_nothing(self):
self.__manager.moveDown([])
self.assertEqual(
list(self.__manager.keys()),
['1', '2', '3', '4', '5', '6', '7', '9',
'11', '12', '13', '14', '15', '16'],
)
# Make sure we still have thye right items
self.test_items()
def test_moveDown_1_no_effect(self):
self.__manager.moveDown(['16'])
self.assertEqual(
list(self.__manager.keys()),
['1', '2', '3', '4', '5', '6', '7', '9',
'11', '12', '13', '14', '15', '16'],
)
# Make sure we still have thye right items
self.test_items()
def test_moveDown_many_no_effect(self):
self.__manager.moveDown(['16', '88', '14', '15', '99'])
self.assertEqual(
list(self.__manager.keys()),
['1', '2', '3', '4', '5', '6', '7', '9',
'11', '12', '13', '14', '15', '16'],
)
# Make sure we still have thye right items
self.test_items()
def test_moveDown_1(self):
self.__manager.moveDown(['3'])
self.assertEqual(
list(self.__manager.keys()),
['1', '2', '4', '3', '5', '6', '7', '9',
'11', '12', '13', '14', '15', '16'],
)
# Make sure we still have thye right items
self.test_items()
def test_moveDown_many(self):
self.__manager.moveDown(
['1', '3', '88', '4', '11', '16', '15', '99'])
self.assertEqual(
list(self.__manager.keys()),
['2', '1', '5', '3', '4', '6', '7', '9',
'12', '11', '13', '14', '15', '16'],
)
# Make sure we still have thye right items
self.test_items()
def test_moveDown_one_element_container(self):
manager = ConfigurationManager()
manager.setObject('', 'a')
manager.moveDown(['1'])
self.assertEqual(list(manager.items()), [('1', 'a')])
#########################################################
def test_manageBeforeDelete(self):
provideAdapter(IContainer, IZopeContainer, ZopeContainerAdapter)
container = []
manager = ConfigurationManager()
thingy = Undeletable()
manager.setObject('xyzzy', thingy)
manager.manage_beforeDelete(manager, container)
self.failUnless(thingy.was_called)
def test_suite():
return makeSuite(Test)
if __name__=='__main__':
main(defaultTest='test_suite')
=== Added File Zope3/src/zope/app/services/tests/test_configurationregistry.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.
#
##############################################################################
"""XXX short summary goes here.
XXX longer description goes here.
$Id: test_configurationregistry.py,v 1.1.2.1 2002/12/23 19:32:23 jim Exp $
"""
from unittest import TestCase, TestSuite, main, makeSuite
from zope.app.services.tests.placefulsetup \
import PlacefulSetup
from zope.proxy.context.context import ContextWrapper, getItem
from zope.app.services.configuration import ConfigurationRegistry
from zope.app.services.service import ServiceManager
from Zope.App.Traversing import traverse
class Configuration:
active = 0
def activated(self):
self.active += 1
def deactivated(self):
self.active -= 1
class Test(PlacefulSetup, TestCase):
def setUp(self):
PlacefulSetup.setUp(self)
self.buildFolders()
root = self.rootFolder
root.setServiceManager(ServiceManager())
self.__default = traverse(root, "++etc++Services/Packages/default")
self.__registry = ContextWrapper(ConfigurationRegistry(), root)
def __config(self, name):
self.__default.setObject(name, Configuration())
return getItem(self.__default, name)
def test_register_and_registered_and_nonzero_and_active(self):
registry = self.__registry
self.assertEqual(registry.active(), None)
self.failIf(registry)
self.__c1 = c1 = self.__config("1")
registry.register(c1)
self.failUnless(registry)
self.failUnless(registry.registered(c1))
self.assertEqual(c1.active, 0)
self.assertEqual(registry.active(), None)
self.__c2 = c2 = self.__config("2")
self.failIf(registry.registered(c2))
registry.register(c2)
self.failUnless(registry)
self.failUnless(registry.registered(c2))
self.assertEqual(c2.active, 0)
def test_unregister_and_registered_and_nonzero(self):
# reuse registration test to set things up (more)
self.test_register_and_registered_and_nonzero_and_active()
registry = self.__registry
c1 = self.__c1
registry.unregister(c1)
self.failIf(registry.registered(c1))
self.assertEqual(c1.active, 0)
c2 = self.__c2
registry.unregister(c2)
self.failIf(registry.registered(c2))
self.assertEqual(c2.active, 0)
self.failIf(registry)
def test_activate_and_active(self):
# reuse registration test to set things up (more)
self.test_register_and_registered_and_nonzero_and_active()
registry = self.__registry
self.assertEqual(registry.active(), None)
c1 = self.__c1
c2 = self.__c2
registry.activate(c2)
self.assertEqual(c1.active, 0)
self.failUnless(registry.registered(c1))
self.assertEqual(c2.active, 1)
self.failUnless(registry.registered(c2))
self.assertEqual(registry.active(), c2)
registry.activate(c2)
self.assertEqual(c1.active, 0)
self.failUnless(registry.registered(c1))
self.assertEqual(c2.active, 1)
self.failUnless(registry.registered(c2))
self.assertEqual(registry.active(), c2)
registry.activate(c1)
self.assertEqual(c1.active, 1)
self.failUnless(registry.registered(c1))
self.assertEqual(c2.active, 0)
self.failUnless(registry.registered(c2))
self.assertEqual(registry.active(), c1)
def test_activate_unregistered(self):
registry = self.__registry
self.assertRaises(ValueError, registry.activate, self.__config('3'))
self.test_activate_and_active()
self.assertRaises(ValueError, registry.activate, self.__config('4'))
def test_deactivate(self):
self.test_activate_and_active()
registry = self.__registry
c1 = self.__c1
c2 = self.__c2
self.assertEqual(registry.active(), c1)
registry.deactivate(c2)
self.assertEqual(c2.active, 0)
self.assertEqual(registry.active(), c1)
registry.deactivate(c1)
self.assertEqual(c2.active, 0)
self.assertEqual(c1.active, 0)
self.assertEqual(registry.active(), None)
self.failUnless(registry.registered(c1))
self.failUnless(registry.registered(c2))
def test_unregister_active(self):
self.test_activate_and_active()
registry = self.__registry
c1 = self.__c1
c2 = self.__c2
self.assertEqual(registry.active(), c1)
registry.unregister(c1)
self.assertEqual(c2.active, 0)
self.assertEqual(c1.active, 0)
self.assertEqual(registry.active(), None)
self.failIf(registry.registered(c1))
self.failUnless(registry.registered(c2))
def test_deactivate_unregistered(self):
registry = self.__registry
self.assertRaises(ValueError, registry.deactivate, self.__config('3'))
def test_info(self):
self.test_activate_and_active()
registry = self.__registry
c1 = self.__c1
c2 = self.__c2
info = registry.info()
info.sort(lambda a, b: cmp(a['id'], b['id']))
self.assertEqual(
info,
[
{'id': 'default/1',
'active': True,
'configuration': c1,
},
{'id': 'default/2',
'active': False,
'configuration': c2,
},
])
registry.deactivate(c1)
info = registry.info()
info.sort(lambda a, b: cmp(a['id'], b['id']))
self.assertEqual(
info,
[
{'id': 'default/1',
'active': False,
'configuration': c1,
},
{'id': 'default/2',
'active': False,
'configuration': c2,
},
])
def test_suite():
return TestSuite((
makeSuite(Test),
))
if __name__=='__main__':
main(defaultTest='test_suite')
=== Added File Zope3/src/zope/app/services/tests/test_configurations.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 tests for configuration classes defined in
Zope.App.OFS.Services.Configuration
$Id: test_configurations.py,v 1.1.2.1 2002/12/23 19:32:23 jim Exp $
"""
from unittest import TestCase, TestSuite, main, makeSuite
from zope.interface import Interface
from zope.app.interfaces.services.configuration \
import Active, Registered, Unregistered
from zope.app.interfaces.dependable import DependencyError
from zope.app.services.configuration import SimpleConfiguration
from zope.app.services.configuration import NamedComponentConfiguration
from zope.app.services.tests.placefulsetup \
import PlacefulSetup
from zope.app.services.tests.testingservicemanager \
import TestingServiceManager
from zope.proxy.context.context import ContextWrapper
from zope.app.interfaces.dependable import IDependable
from Zope.App.Traversing import traverse
from zope.security.proxy import Proxy
class ITestComponent(Interface):
pass
class ComponentStub:
__implements__ = IDependable
_dependents = ()
def addDependent(self, location):
self._dependents = tuple(
[d for d in self._dependents if d != location]
+
[location]
)
def removeDependent(self, location):
self._dependents = tuple(
[d for d in self._dependents if d != location]
)
def dependents(self):
return self._dependents
class TestSimpleConfiguration(TestCase):
def test_manage_beforeDelete(self):
container = object()
cfg = SimpleConfiguration()
# cannot delete an active configuration
cfg.status = Active
self.assertRaises(DependencyError, cfg.manage_beforeDelete, cfg,
container)
# deletion of a registered configuration causes it to become
# unregistered
cfg.status = Registered
cfg.manage_beforeDelete(cfg, container)
self.assertEquals(cfg.status, Unregistered)
class TestNamedComponentConfiguration(TestSimpleConfiguration, PlacefulSetup):
def setUp(self):
PlacefulSetup.setUp(self)
self.buildFolders()
self.__sm = TestingServiceManager()
self.rootFolder.setServiceManager(self.__sm)
self.name = 'foo'
def test_getComponent(self):
# set up a component
path, component = 'foo', object()
self.rootFolder.setObject(path, component)
# set up a configuration
cfg = NamedComponentConfiguration(self.name, path)
cfg = ContextWrapper(cfg, self.rootFolder)
# check that getComponent finds the configuration
self.assertEquals(cfg.getComponent(), component)
def test_getComponent_permission(self):
# set up a component
path, component = 'foo', object()
self.rootFolder.setObject(path, component)
# set up a configuration
cfg = NamedComponentConfiguration(self.name, path, 'Zope.TopSecret')
cfg.getInterface = lambda: ITestComponent
cfg = ContextWrapper(cfg, self.rootFolder)
# check that getComponent finds the configuration
result = cfg.getComponent()
self.assertEquals(result, component)
self.failUnless(type(result) is Proxy)
def test_manage_afterAdd(self):
# set up a component
path, component = 'foo', ComponentStub()
self.rootFolder.setObject(path, component)
# set up a configuration
cfg = NamedComponentConfiguration(self.name, path)
self.rootFolder.setObject('cfg', cfg)
cfg = traverse(self.rootFolder, 'cfg')
# simulate IAddNotifiable
cfg.manage_afterAdd(cfg, self.rootFolder)
# check that the dependency tracking works
self.assertEquals(component.dependents(), ('/cfg',))
def test_manage_beforeDelete_dependents(self):
# set up a component
path, component = 'foo', ComponentStub()
self.rootFolder.setObject(path, component)
component.addDependent('/cfg')
# set up a configuration
cfg = NamedComponentConfiguration(self.name, path)
cfg.status = Unregistered
self.rootFolder.setObject('cfg', cfg)
cfg = traverse(self.rootFolder, 'cfg')
# simulate IDeleteNotifiable
cfg.manage_beforeDelete(cfg, self.rootFolder)
# check that the dependency tracking works
self.assertEquals(component.dependents(), ())
# NamedConfiguration is too simple to need testing at the moment
def test_suite():
return TestSuite((
makeSuite(TestSimpleConfiguration),
makeSuite(TestNamedComponentConfiguration),
))
if __name__=='__main__':
main(defaultTest='test_suite')
=== Added File Zope3/src/zope/app/services/tests/test_configurationstatusproperty.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.
#
##############################################################################
"""XXX short summary goes here.
XXX longer description goes here.
$Id: test_configurationstatusproperty.py,v 1.1.2.1 2002/12/23 19:32:23 jim Exp $
"""
from unittest import TestCase, TestSuite, main, makeSuite
from zope.component.interfaces import IServiceService
from zope.app.services.tests.placefulsetup \
import PlacefulSetup
from zope.app.services.role import RoleService
from zope.app.services.tests.testingconfigurationregistry \
import TestingConfigurationRegistry, TestingConfiguration
from zope.app.services.configuration import ConfigurationStatusProperty
from zope.app.interfaces.services.configuration \
import Active, Unregistered, Registered
from zope.proxy.context.context import ContextWrapper
from zope.component.exceptions import ComponentLookupError
class TestingConfiguration(TestingConfiguration):
status = ConfigurationStatusProperty("Services")
service_type = "Test"
class PassiveConfiguration(TestingConfiguration):
status = ConfigurationStatusProperty("NoSuchService")
class TestingConfigurationRegistry(TestingConfigurationRegistry):
class_ = TestingConfiguration
class TestingServiceManager:
__implements__ = IServiceService # I lied
registry = None
def getService(self, name):
if name == "Services":
return self
raise ComponentLookupError("Wrong service name", name)
def queryService(self, name, default=None):
if name == "Services":
return self
else:
return default
def queryConfigurationsFor(self, configuration, default=None):
if configuration.service_type != "Test":
raise ValueError("Bad service type", configuration.service_type)
return self.registry
def createConfigurationsFor(self, configuration):
if configuration.service_type != "Test":
raise ValueError("Bad service type", configuration.service_type)
self.registry = TestingConfigurationRegistry()
return self.registry
class Test(PlacefulSetup, TestCase):
def setUp(self):
PlacefulSetup.setUp(self)
self.buildFolders()
self.__sm = TestingServiceManager()
self.rootFolder.setServiceManager(self.__sm)
def test_property(self):
configa = ContextWrapper(TestingConfiguration('a'), self.rootFolder)
self.assertEqual(configa.status, Unregistered)
configa.status = Registered
self.assertEqual(self.__sm.registry._data, (None, 'a'))
self.assertEqual(configa.status, Registered)
configa.status = Active
self.assertEqual(self.__sm.registry._data, ('a', ))
self.assertEqual(configa.status, Active)
configb = ContextWrapper(TestingConfiguration('b'), self.rootFolder)
self.assertEqual(self.__sm.registry._data, ('a', ))
self.assertEqual(configb.status, Unregistered)
configb.status = Registered
self.assertEqual(self.__sm.registry._data, ('a', 'b'))
self.assertEqual(configb.status, Registered)
configc = ContextWrapper(TestingConfiguration('c'), self.rootFolder)
self.assertEqual(configc.status, Unregistered)
self.assertEqual(self.__sm.registry._data, ('a', 'b'))
configc.status = Registered
self.assertEqual(self.__sm.registry._data, ('a', 'b', 'c'))
self.assertEqual(configc.status, Registered)
configc.status = Active
self.assertEqual(self.__sm.registry._data, ('c', 'a', 'b'))
self.assertEqual(configc.status, Active)
configc.status = Unregistered
self.assertEqual(self.__sm.registry._data, (None, 'a', 'b'))
self.assertEqual(configc.status, Unregistered)
self.assertEqual(configb.status, Registered)
self.assertEqual(configa.status, Registered)
def test_passive(self):
# scenario:
# 1. create and configure an SQLConnectionService
# 2. create and configure a database adapter&connection
# 3. disable SQLConnectionService
# now the ConnectionConfiguration.status cannot access the
# SQLConnectionService
configa = ContextWrapper(PassiveConfiguration('a'), self.rootFolder)
self.assertEqual(configa.status, Unregistered)
try:
configa.status = Registered
except ComponentLookupError:
self.assertEqual(configa.status, Unregistered)
else:
self.fail("should complain about missing service")
try:
configa.status = Active
except ComponentLookupError:
self.assertEqual(configa.status, Unregistered)
else:
self.fail("should complain about missing service")
def test_suite():
return TestSuite((
makeSuite(Test),
))
if __name__=='__main__':
main(defaultTest='test_suite')
=== Added File Zope3/src/zope/app/services/tests/test_connectionconfiguration.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 ConnectionConfiguration.
$Id: test_connectionconfiguration.py,v 1.1.2.1 2002/12/23 19:32:23 jim Exp $
"""
from unittest import TestCase, TestSuite, main, makeSuite
from zope.interface import Interface
from zope.app.services.connection \
import ConnectionConfiguration
from zope.app.services.tests.placefulsetup \
import PlacefulSetup
from Zope.App.Traversing import traverse
from zope.app.services.service \
import ServiceManager
from zope.app.container.zope import ZopeContainerAdapter
from zope.app.interfaces.services.configuration \
import Active, Unregistered, Registered
from zope.app.interfaces.rdb import IZopeDatabaseAdapter
from zope.app.interfaces.dependable import IDependable
from zope.app.interfaces.rdb import IConnectionService
from zope.app.interfaces.services.configuration import IConfigurable
from zope.app.services.configuration import ConfigurationRegistry
from zope.app.services.service \
import ServiceConfiguration
from zope.proxy.context import ContextMethod
from zope.proxy.context.context import ContextWrapper
class DependableStub:
__implements__ = IDependable
def addDependent(self, location):
pass
def removeDependent(self, location):
pass
def dependents(self):
pass
class TestDA(DependableStub):
__implements__ = IZopeDatabaseAdapter, IDependable
class ConnectionServiceStub(DependableStub):
__implements__ = IConnectionService, IConfigurable, IDependable
def __init__(self):
self.bindings = {}
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)
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('da', TestDA())
self.da = traverse(self.default, 'da')
self.cm = ZopeContainerAdapter(traverse(self.default, "configure"))
self.cm.setObject('', ConnectionConfiguration('conn_name',
'/++etc++Services/Packages/default/da'))
self.config = traverse(self.default, 'configure/1')
self.default.setObject('conn_srv', ConnectionServiceStub())
self.service = traverse(self.default, 'conn_srv')
self.cm.setObject('', ServiceConfiguration('SQLDatabaseConnections',
'/++etc++Services/Packages/default/conn_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.da)
def test_status(self):
self.assertEqual(self.config.status, Unregistered)
self.config.status = Active
self.assertEqual(self.config.status, Active)
cr = self.service.queryConfigurations('conn_name')
self.assertEqual(cr.active(), self.config)
def test_suite():
return makeSuite(TestConnectionConfiguration)
if __name__=='__main__':
main(defaultTest='test_suite')
=== Added File Zope3/src/zope/app/services/tests/test_connectionservice.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
#
##############################################################################
"""DT_SQLVar Tests
$Id: test_connectionservice.py,v 1.1.2.1 2002/12/23 19:32:23 jim Exp $
"""
import unittest
from Zope.App.Traversing import traverse
from zope.app.interfaces.rdb import IZopeConnection
from zope.app.interfaces.rdb import IZopeDatabaseAdapter
from zope.app.interfaces.rdb import \
IConnectionService
from zope.app.container.zope import ZopeContainerAdapter
from zope.app.services.connection import \
ConnectionService
from zope.app.services.service \
import ServiceManager
from zope.app.services.service \
import ServiceConfiguration
from zope.app.interfaces.dependable import IDependable
from zope.app.services.tests.placefulsetup \
import PlacefulSetup
from zope.app.interfaces.annotation import IAnnotatable
from zope.app.interfaces.annotation import IAttributeAnnotatable
from zope.app.attributeannotations import AttributeAnnotations
from zope.app.interfaces.annotation import IAnnotations
from zope.app.interfaces.dependable import IDependable
from zope.app.dependable import Dependable
from zope.component.adapter import provideAdapter
from zope.app.interfaces.services.configuration \
import Active, Unregistered, Registered
from zope.app.services.connection \
import ConnectionConfiguration
class ConnectionServiceForTests(ConnectionService):
__implements__ = ConnectionService.__implements__, IAttributeAnnotatable
class DAStub:
__implements__ = IZopeDatabaseAdapter, IAttributeAnnotatable
def __init__(self, n):
self.n = n
def __call__(self):
return 'DA #%d' % self.n
def sort(list):
list.sort()
return list
class TestConnectionService(unittest.TestCase, PlacefulSetup):
def setUp(self):
PlacefulSetup.setUp(self)
provideAdapter(IAttributeAnnotatable,
IAnnotations, AttributeAnnotations)
provideAdapter(IAnnotatable, IDependable, Dependable)
# Set up a local connection service
self.buildFolders()
self.rootFolder.setServiceManager(ServiceManager())
self.default = traverse(self.rootFolder,
'++etc++Services/Packages/default')
self.default.setObject('conn_srv', ConnectionServiceForTests())
self.service = traverse(self.default, 'conn_srv')
self.cm = ZopeContainerAdapter(traverse(self.default, "configure"))
self.cm.setObject('', ServiceConfiguration('SQLDatabaseConnections',
'/++etc++Services/Packages/default/conn_srv'))
traverse(self.default, 'configure/1').status = Active
self.default.setObject('da1', DAStub(1))
self.default.setObject('da2', DAStub(2))
self.cm.setObject('', ConnectionConfiguration('conn1',
'/++etc++Services/Packages/default/da1'))
traverse(self.default, 'configure/2').status = Active
self.cm.setObject('', ConnectionConfiguration('conn2',
'/++etc++Services/Packages/default/da2'))
traverse(self.default, 'configure/3').status = Active
self.cm.setObject('', ConnectionConfiguration('conn3',
'/++etc++Services/Packages/default/da1'))
traverse(self.default, 'configure/4').status = Registered
# Now self.service has conn1 and conn2 available and knows about conn3
# Set up a more local connection service
folder1 = traverse(self.rootFolder, 'folder1')
folder1.setServiceManager(ServiceManager())
default1 = traverse(folder1, '++etc++Services/Packages/default')
default1.setObject('conn_srv1', ConnectionServiceForTests())
self.service1 = traverse(default1, 'conn_srv1')
cm1 = ZopeContainerAdapter(traverse(default1, "configure"))
cm1.setObject('', ServiceConfiguration('SQLDatabaseConnections',
'/folder1/++etc++Services/Packages/default/conn_srv1'))
traverse(default1, 'configure/1').status = Active
default1.setObject('da3', DAStub(3))
default1.setObject('da4', DAStub(4))
cm1.setObject('', ConnectionConfiguration('conn1',
'/folder1/++etc++Services/Packages/default/da3'))
traverse(default1, 'configure/2').status = Active
cm1.setObject('', ConnectionConfiguration('conn4',
'/folder1/++etc++Services/Packages/default/da4'))
traverse(default1, 'configure/3').status = Active
# Now self.service1 overrides conn1, adds new conn4 available, and
# inherits conn2 from self.service
def testGetConnection(self):
self.assertEqual('DA #1', self.service.getConnection('conn1'))
self.assertEqual('DA #2', self.service.getConnection('conn2'))
self.assertRaises(KeyError, self.service.getConnection, 'conn3')
self.assertRaises(KeyError, self.service.getConnection, 'conn4')
self.assertEqual('DA #3', self.service1.getConnection('conn1'))
self.assertEqual('DA #2', self.service1.getConnection('conn2'))
self.assertRaises(KeyError, self.service1.getConnection, 'conn3')
self.assertEqual('DA #4', self.service1.getConnection('conn4'))
self.assertRaises(KeyError, self.service1.getConnection, 'conn5')
def testQueryConnection(self):
self.assertEqual('DA #1', self.service.queryConnection('conn1'))
self.assertEqual('DA #2', self.service.queryConnection('conn2'))
self.assertEqual(None, self.service.queryConnection('conn3'))
self.assertEqual('xx', self.service.queryConnection('conn3', 'xx'))
self.assertEqual(None, self.service.queryConnection('conn4'))
self.assertEqual('xx', self.service.queryConnection('conn4', 'xx'))
self.assertEqual('DA #3', self.service1.queryConnection('conn1'))
self.assertEqual('DA #2', self.service1.queryConnection('conn2'))
self.assertEqual(None, self.service1.queryConnection('conn3'))
self.assertEqual('xx', self.service1.queryConnection('conn3', 'xx'))
self.assertEqual('DA #4', self.service1.queryConnection('conn4'))
self.assertEqual(None, self.service1.queryConnection('conn5'))
self.assertEqual('xx', self.service1.queryConnection('conn5', 'xx'))
def testGetAvailableConnections(self):
self.assertEqual(['conn1', 'conn2'],
sort(self.service.getAvailableConnections()))
self.assertEqual(['conn1', 'conn2', 'conn4'],
sort(self.service1.getAvailableConnections()))
def test_suite():
suite = unittest.TestSuite()
suite.addTest(unittest.makeSuite(TestConnectionService))
return suite
if __name__ == '__main__':
unittest.TextTestRunner().run(test_suite())
=== Added File Zope3/src/zope/app/services/tests/test_errorreportingservice.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: test_errorreportingservice.py,v 1.1.2.1 2002/12/23 19:32:23 jim Exp $
"""
import sys
from unittest import TestCase, TestLoader, TextTestRunner
from zope.app.services.errorr \
import ErrorReportingService
from zope.testing.cleanup import CleanUp
from zope.exceptions.exceptionformatter import format_exception
class C1:
def getAnErrorInfo(self):
exc_info = None
try:
someerror()
except:
exc_info = sys.exc_info()
return exc_info
class ErrorReportingServiceTests(CleanUp, TestCase):
def test_checkForEmpryLog(self):
# Test Check Empty Log
errService = ErrorReportingService()
getProp = errService.getLogEntries()
self.failIf(getProp)
def test_checkProperties(self):
# Test Properties test
errService = ErrorReportingService()
setProp = {
'keep_entries':10,
'copy_to_zlog':1,
'ignored_exceptions':()
}
errService.setProperties(**setProp)
getProp = errService.getProperties()
self.assertEqual(setProp, getProp)
def test_ErrorLog(self):
# Test for Logging Error. Create one error and check whether its
# logged or not.
errService = ErrorReportingService()
exc_info = C1().getAnErrorInfo()
errService.raising(exc_info)
getErrLog = errService.getLogEntries()
self.assertEquals(1, len(getErrLog))
tb_text = ''.join(format_exception(*exc_info, **{'as_html': 0}))
err_id = getErrLog[0]['id']
self.assertEquals(tb_text,
errService.getLogEntryById(err_id)['tb_text'])
def test_suite():
loader=TestLoader()
return loader.loadTestsFromTestCase(ErrorReportingServiceTests)
if __name__=='__main__':
TextTestRunner().run(test_suite())
=== Added File Zope3/src/zope/app/services/tests/test_eventservice.py === (407/507 lines abridged)
##############################################################################
#
# 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: test_eventservice.py,v 1.1.2.1 2002/12/23 19:32:23 jim Exp $
"""
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.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.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.interfaces.services.configuration \
import Active, Unregistered, Registered
from zope.proxy.context.context import ContextWrapper
[-=- -=- -=- 407 lines omitted -=- -=- -=-]
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()
self.rootFolder.setObject(
"mySubscriber",
DummySubscriptionAwareSubscriber())
self.mySubscriber=ContextWrapper(
self.rootFolder["mySubscriber"],
self.rootFolder,
name="mySubscriber")
filter = DummyFilter()
subscribe(
self.mySubscriber,
event_type=IObjectAddedEvent,
filter=filter)
self.assertEqual(
self.mySubscriber.subscribable,
getEventService(self.rootFolder))
self.assertEqual(
self.mySubscriber.event_type,
IObjectAddedEvent)
self.assertEqual(
self.mySubscriber.filter,
filter)
unsubscribe(
self.mySubscriber,
event_type=IObjectAddedEvent,
filter=filter)
self.assertEqual(
self.mySubscriber.un_subscribable,
getEventService(self.rootFolder))
self.assertEqual(
self.mySubscriber.un_event_type,
IObjectAddedEvent)
self.assertEqual(
self.mySubscriber.un_filter,
filter)
def test_suite():
loader=TestLoader()
return loader.loadTestsFromTestCase(EventServiceTests)
if __name__=='__main__':
TextTestRunner().run(test_suite())
=== Added File Zope3/src/zope/app/services/tests/test_field.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.
#
##############################################################################
"""Tests for ComponentLocation field.
$Id: test_field.py,v 1.1.2.1 2002/12/23 19:32:23 jim Exp $
"""
from unittest import TestCase, TestSuite, main, makeSuite
from zope.app.services.tests.placefulsetup \
import PlacefulSetup
from Zope.App.Traversing import traverse
from zope.schema.interfaces import ValidationError
from zope.interface import Interface
from zope.app.services.field import ComponentLocation
class I1(Interface): pass
class C:
__implements__ = I1
class D:
pass
class Test(PlacefulSetup, TestCase):
def test__validate(self):
self.buildFolders()
self.folder1.setObject('c', C())
self.folder1.setObject('d', D())
folder2 = traverse(self.rootFolder, 'folder2')
field = ComponentLocation(type=I1)
field = field.bind(folder2)
field.validate(u'/folder1/c')
self.assertRaises(ValidationError, field.validate, u'/folder1/d')
self.assertRaises(ValidationError, field.validate, u'/folder1/e')
def test_suite():
return TestSuite((
makeSuite(Test),
))
if __name__=='__main__':
main(defaultTest='test_suite')
=== Added File Zope3/src/zope/app/services/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.
#
##############################################################################
"""
Revision information:
$Id: test_hookedhubevent.py,v 1.1.2.1 2002/12/23 19:32:23 jim 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, sys
from zope.app.services.tests.objecthubsetup import ObjectHubSetup
from zope.app.services.hub import \
ObjectRegisteredHubEvent, ObjectUnregisteredHubEvent, \
ObjectModifiedHubEvent, ObjectMovedHubEvent, \
ObjectRemovedHubEvent
from Zope.App.Traversing import getPhysicalPath
from zope.component 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 = getPhysicalPath(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.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)
self.event = self.klass(self.object_hub, self.hubid)
class TestObjectUnregisteredHubEvent(AbstractTestHubEvent):
klass = ObjectUnregisteredHubEvent
class TestEmptyObjectUnregisteredHubEvent(TestObjectUnregisteredHubEvent):
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)
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.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)
self.event = self.klass(self.object_hub, self.hubid)
class TestObjectMovedHubEvent(AbstractTestHubEvent):
fromLocation = '/old/location'
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)
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.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)
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/services/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.
#
##############################################################################
"""
Revision information:
$Id: test_hubevent.py,v 1.1.2.1 2002/12/23 19:32:23 jim Exp $
"""
import unittest, sys
from zope.app.services.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 getLocation(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/services/tests/test_nameconfigurable.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.
#
##############################################################################
"""NameConfigurable tests
$Id: test_nameconfigurable.py,v 1.1.2.1 2002/12/23 19:32:23 jim Exp $
"""
from unittest import TestCase, TestSuite, main, makeSuite
from zope.app.services.configuration import NameConfigurable
from zope.app.services.configuration import NameComponentConfigurable
from zope.proxy.context.context import ContextWrapper
from zope.proxy.context.context import getWrapperContainer
class ConfigurationStub:
def __init__(self, **kw):
self.__dict__.update(kw)
def getComponent(self):
return self.component
class RegistryStub:
def __init__(self, nonzero=0, active=None):
self._nonzero = nonzero or (active and 1 or 0)
self._active = active
def __nonzero__(self):
return self._nonzero
def active(self):
return self._active
class TestNameConfigurable(TestCase):
def setUp(self):
self.container = object()
self.subject = ContextWrapper(NameConfigurable(), self.container)
def test_queryConfigurationsFor(self):
subject = self.subject
cfg = ConfigurationStub(name="Foo")
self.assertEquals(subject.queryConfigurationsFor(cfg), None)
self.assertEquals(subject.queryConfigurationsFor(cfg, 42), 42)
registry = RegistryStub()
subject._bindings["Foo"] = registry
result = subject.queryConfigurationsFor(cfg)
self.assertEquals(result, registry)
self.assertEquals(getWrapperContainer(result), subject)
self.assertEquals(getWrapperContainer(getWrapperContainer(result)),
self.container)
def test_queryConfigurations(self):
subject = self.subject
self.assertEquals(subject.queryConfigurations("Foo"), None)
self.assertEquals(subject.queryConfigurations("Foo", 42), 42)
registry = RegistryStub()
subject._bindings["Foo"] = registry
result = subject.queryConfigurations("Foo")
self.assertEquals(result, registry)
self.assertEquals(getWrapperContainer(result), subject)
self.assertEquals(getWrapperContainer(getWrapperContainer(result)),
self.container)
def test_createConfigurationsFor(self):
subject = self.subject
cfg1 = ConfigurationStub(name='Foo')
cfg2 = ConfigurationStub(name='Bar')
r1 = subject.createConfigurationsFor(cfg1)
r2 = subject.createConfigurationsFor(cfg2)
r3 = subject.createConfigurationsFor(cfg1)
self.assertEquals(r1, r3)
self.assertNotEquals(r1, r2)
self.assertNotEquals(r2, r3)
self.assertEquals(r3, subject._bindings['Foo'])
self.assertEquals(getWrapperContainer(r3), subject)
self.assertEquals(getWrapperContainer(getWrapperContainer(r3)),
self.container)
self.failUnless(subject._p_changed)
def test_createConfigurations(self):
subject = self.subject
r1 = subject.createConfigurations('Foo')
r2 = subject.createConfigurations('Bar')
r3 = subject.createConfigurations('Foo')
self.assertEquals(r1, r3)
self.assertNotEquals(r1, r2)
self.assertNotEquals(r2, r3)
self.assertEquals(r3, subject._bindings['Foo'])
self.assertEquals(getWrapperContainer(r3), subject)
self.assertEquals(getWrapperContainer(getWrapperContainer(r3)),
self.container)
self.failUnless(subject._p_changed)
def test_listConfigurationNames(self):
subject = self.subject
self.assertEquals(tuple(subject.listConfigurationNames()), ())
subject._bindings['Foo'] = 1
self.assertEquals(tuple(subject.listConfigurationNames()), ('Foo',))
subject._bindings['Bar'] = 0 # false values should be filtered out
self.assertEquals(tuple(subject.listConfigurationNames()), ('Foo',))
class TestNameComponentConfigurable(TestNameConfigurable):
def setUp(self):
self.container = object()
self.subject = ContextWrapper(NameComponentConfigurable(),
self.container)
def test_queryActiveComponent(self):
subject = self.subject
self.assertEquals(subject.queryActiveComponent('xyzzy'), None)
self.assertEquals(subject.queryActiveComponent('xyzzy', 'No'), 'No')
subject._bindings['xyzzy'] = RegistryStub()
self.assertEquals(subject.queryActiveComponent('xyzzy'), None)
subject._bindings['xyzzy'] = RegistryStub(nonzero=1)
self.assertEquals(subject.queryActiveComponent('xyzzy'), None)
cfg = ConfigurationStub(component='X')
subject._bindings['xyzzy'] = RegistryStub(active=cfg)
self.assertEquals(subject.queryActiveComponent('xyzzy'), 'X')
def test_suite():
return TestSuite((
makeSuite(TestNameConfigurable),
makeSuite(TestNameComponentConfigurable),
))
if __name__=='__main__':
main(defaultTest='test_suite')
=== Added File Zope3/src/zope/app/services/tests/test_objecthub.py === (509/609 lines abridged)
##############################################################################
#
# 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: test_objecthub.py,v 1.1.2.1 2002/12/23 19:32:24 jim Exp $
"""
import unittest, sys
from zope.app.services.tests.objecthubsetup import ObjectHubSetup
from zope.app.interfaces.event\
import IObjectAddedEvent, IObjectRemovedEvent, IObjectModifiedEvent,\
IObjectMovedEvent
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.exceptions import NotFoundError
from types import StringTypes
from Zope.App.Traversing import locationAsUnicode, locationAsTuple
from zope.component import getService, getServiceManager
# 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
[-=- -=- -=- 509 lines omitted -=- -=- -=-]
hub.notify(moved_event)
self.assertRaises(NotFoundError, hub.getHubId, location)
self.assertRaises(NotFoundError, hub.getHubId, 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(TestSearchRegistrations),
unittest.makeSuite(TestObjectAddedEvent),
unittest.makeSuite(TestObjectRemovedEvent),
unittest.makeSuite(TestObjectModifiedEvent),
unittest.makeSuite(TestObjectMovedEvent),
))
if __name__=='__main__':
unittest.main(defaultTest='test_suite')
=== Added File Zope3/src/zope/app/services/tests/test_principalannotation.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: test_principalannotation.py,v 1.1.2.1 2002/12/23 19:32:24 jim Exp $
"""
from unittest import TestCase, TestLoader, TextTestRunner
from zope.app.services.tests.placefulsetup \
import PlacefulSetup
from zope.component import getServiceManager, getService
from zope.app.services.principalannotation import PrincipalAnnotationService, Annotations, AnnotationsForPrincipal
from zope.app.interfaces.services.principalannotation import IPrincipalAnnotationService
from zope.component.adapter import provideAdapter
from zope.component import getAdapter
from zope.app.interfaces.annotation import IAnnotations
from zope.app.interfaces.security import IPrincipal
class Principal:
__implements__ = IPrincipal
def __init__(self, id):
self.id = id
def getId(self):
return self.id
class PrincipalAnnotationTests(PlacefulSetup, TestCase):
def setUp(self):
PlacefulSetup.setUp(self)
self.buildFolders()
root_sm = getServiceManager(None)
svc = PrincipalAnnotationService()
root_sm.defineService("PrincipalAnnotation", IPrincipalAnnotationService)
root_sm.provideService("PrincipalAnnotation", svc)
self.createServiceManager()
sm = getServiceManager(self.rootFolder)
sm.PrincipalAnnotation = svc
self.svc = getService(self.rootFolder, "PrincipalAnnotation")
def testGetSimple(self):
prince = Principal('somebody')
self.assert_(not self.svc.hasAnnotation(prince.getId()))
princeAnnotation = self.svc.getAnnotation(prince.getId())
self.assert_(self.svc.hasAnnotation(prince.getId()))
princeAnnotation['something'] = 'whatever'
def testGetFromLayered(self):
princeSomebody = Principal('somebody')
self.createServiceManager(self.folder1)
sm1 = getServiceManager(self.folder1)
sm1.PrincipalAnnotation = PrincipalAnnotationService()
subService = getService(self.folder1, "PrincipalAnnotation")
parentAnnotation = self.svc.getAnnotation(princeSomebody.getId())
self.assert_(self.svc.hasAnnotation(princeSomebody.getId()))
self.assert_(not subService.hasAnnotation(princeSomebody.getId()))
parentAnnotation['hair_color'] = 'blue'
subAnnotation = subService.getAnnotation(princeSomebody.getId())
self.assertEquals(subAnnotation['hair_color'], 'blue')
subAnnotation['foo'] = 'bar'
self.assertEquals(parentAnnotation.get("foo"), None)
def testAdapter(self):
p = Principal('somebody')
provideAdapter(IPrincipal, IAnnotations, AnnotationsForPrincipal(self.svc))
annotations = getAdapter(p, IAnnotations)
annotations["test"] = "bar"
annotations = getAdapter(p, IAnnotations)
self.assertEquals(annotations["test"], "bar")
def test_suite():
loader=TestLoader()
return loader.loadTestsFromTestCase(PrincipalAnnotationTests)
if __name__=='__main__':
TextTestRunner().run(test_suite())
=== Added File Zope3/src/zope/app/services/tests/test_roleservice.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: test_roleservice.py,v 1.1.2.1 2002/12/23 19:32:24 jim Exp $
"""
from unittest import TestCase, TestLoader, TextTestRunner
from zope.app.services.tests.placefulsetup \
import PlacefulSetup
from zope.component import getServiceManager, getService
from zope.app.interfaces.security import IRoleService
from zope.app.security.registries.roleregistry import roleRegistry
from zope.app.services.role import RoleService
from zope.app.services.role import Role
class RoleServiceTests(PlacefulSetup, TestCase):
def _Test__new(self):
return RoleService()
def setUp(self):
PlacefulSetup.setUp(self)
self.buildFolders()
root_sm = getServiceManager(None)
root_sm.defineService("Roles", IRoleService)
self.roleRegistry = roleRegistry
root_sm.provideService("Roles", roleRegistry)
self.createServiceManager()
sm = getServiceManager(self.rootFolder)
rs = RoleService()
sm.Roles = rs
self.rs = getService(self.rootFolder,"Roles")
def testGetRole(self):
self.roleRegistry.defineRole('Manager', 'Manager', '')
r = Role("Hacker","","")
self.rs.setObject("Hacker", r)
self.assertEqual(self.rs.getRole('Hacker').getId(), 'Hacker')
self.assertEqual(self.rs.getRole('Manager').getId(), 'Manager')
roles = [role.getId() for role in self.rs.getRoles()]
roles.sort()
self.assertEqual(roles, ['Anonymous', 'Hacker', 'Manager'])
def testGetRoleFromLayeredServices(self):
self.roleRegistry.defineRole('Manager', 'Manager', '')
r = Role("Hacker","","")
self.rs.setObject("Hacker", r)
self.createServiceManager(self.folder1)
sm1 = getServiceManager(self.folder1)
sm1.Roles = RoleService()
rs1 = getService(self.folder1, "Roles")
r1 = Role("Reviewer",'','')
rs1.setObject("Reviewer", r1)
self.assertEqual(rs1.getRole('Hacker').getId(), 'Hacker')
self.assertEqual(rs1.getRole('Manager').getId(), 'Manager')
self.assertEqual(rs1.getRole('Reviewer').getId(), 'Reviewer')
roles = [role.getId() for role in rs1.getRoles()]
roles.sort()
self.assertEqual(roles, ['Anonymous', 'Hacker', 'Manager','Reviewer'])
def test_suite():
loader=TestLoader()
return loader.loadTestsFromTestCase(RoleServiceTests)
if __name__=='__main__':
TextTestRunner().run(test_suite())
=== Added File Zope3/src/zope/app/services/tests/test_serviceconfiguration.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.
#
##############################################################################
"""XXX short summary goes here.
XXX longer description goes here.
$Id: test_serviceconfiguration.py,v 1.1.2.1 2002/12/23 19:32:24 jim Exp $
"""
from unittest import TestCase, TestSuite, main, makeSuite
from zope.interface import Interface
from zope.component import getServiceManager
from Zope.App.Traversing import traverse
from zope.app.services.service \
import ServiceConfiguration
from zope.app.services.tests.placefulsetup \
import PlacefulSetup
from zope.app.services.service \
import ServiceManager
from zope.component.service \
import serviceManager
from zope.app.interfaces.services.service \
import IBindingAware
from zope.app.interfaces.services.configuration \
import Active, Unregistered, Registered
from zope.app.interfaces.dependable import IDependable
from zope.app.interfaces.dependable import DependencyError
from zope.app.container.zope import ZopeContainerAdapter
class ITestService(Interface):
pass
class TestService:
__implements__ = ITestService, IBindingAware, IDependable
_bound = _unbound = ()
def bound(self, name):
self._bound += (name, )
def unbound(self, name):
self._unbound += (name, )
_dependents = ()
def addDependent(self, location):
self._dependents = tuple(
[d for d in self._dependents if d != location]
+
[location]
)
def removeDependent(self, location):
self._dependents = tuple(
[d for d in self._dependents if d != location]
)
def dependents(self):
return self._dependents
class Test(PlacefulSetup, TestCase):
def setUp(self):
PlacefulSetup.setUp(self)
self.buildFolders()
self.rootFolder.setServiceManager(ServiceManager())
serviceManager.defineService('test_service', ITestService)
default = traverse(self.rootFolder,
'++etc++Services/Packages/default')
self.__default = default
default.setObject('c', TestService())
configuration = ServiceConfiguration(
'test_service', '/++etc++Services/Packages/default/c')
self.__c = traverse(default, 'c')
self.__cm = ZopeContainerAdapter(traverse(default, "configure"))
self.__cm.setObject('', configuration)
self.__config = traverse(default, 'configure/1')
def test_activated(self):
old = self.__c._bound
self.__config.activated()
self.assertEqual(self.__c._bound, old+('test_service',))
def test_deactivated(self):
old = self.__c._unbound
self.__config.deactivated()
self.assertEqual(self.__c._unbound, old+('test_service',))
def test_getInterface(self):
self.assertEquals(self.__config.getInterface(), ITestService)
# XXX the following tests check the same things as
# Zope.App.OFS.Services.tests.testConfigurations, but in a different way
def test_getComponent(self):
self.assertEqual(self.__config.getComponent(), self.__c)
def test_manage_afterAdd(self):
self.assertEqual(self.__c._dependents,
('/++etc++Services/Packages/default/configure/1', ))
def test_manage_beforeDelete_and_unregistered(self):
self.__config.status = Registered
sm = getServiceManager(self.__default)
registry = sm.queryConfigurationsFor(self.__config)
self.failUnless(registry, "The components should be registered")
del self.__cm['1']
self.assertEqual(self.__c._dependents, ())
self.failIf(registry, "The components should not be registered")
def test_disallow_delete_when_active(self):
self.__config.status = Active
try:
del self.__cm['1']
except DependencyError:
pass # OK
else:
self.failUnless(0, "Should have gotten a depency error")
def test_suite():
return makeSuite(Test)
if __name__=='__main__':
main(defaultTest='test_suite')
=== Added File Zope3/src/zope/app/services/tests/test_servicemanager.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: test_servicemanager.py,v 1.1.2.1 2002/12/23 19:32:24 jim Exp $
"""
from unittest import TestCase, TestLoader, TextTestRunner
from zope.interface import Interface
from zope.app.content.folder import RootFolder
from zope.app.content.folder import Folder
from zope.proxy.context.context import getWrapperContext, getWrapperContainer
from zope.app.services.service import ServiceManager
from zope.app.services.service \
import ServiceConfiguration
from zope.component import getService, getServiceManager
from zope.exceptions import ZopeError
from zope.app.services.tests.placefulsetup import PlacefulSetup
from Zope.App.Traversing import traverse
from zope.app.interfaces.services.configuration \
import Active, Unregistered, Registered
from zope.component.service \
import serviceManager
class ITestService(Interface):
pass
class TestService:
__implements__ = ITestService
class ServiceManagerTests(PlacefulSetup, TestCase):
def setUp(self):
PlacefulSetup.setUp(self)
self.buildFolders()
serviceManager.defineService('test_service', ITestService)
def _Test__new(self):
return ServiceManager()
def createServiceManager(self):
self.rootFolder.setServiceManager(ServiceManager())
def testGetService(self):
self.createServiceManager()
sm = traverse(self.rootFolder, '++etc++Services')
default = traverse(sm, 'Packages/default')
ts = TestService()
default.setObject('test_service1', ts)
configuration = ServiceConfiguration(
'test_service',
'/++etc++Services/Packages/default/test_service1')
default['configure'].setObject('', configuration)
traverse(default, 'configure/1').status = Active
testOb = getService(self.rootFolder, 'test_service')
c = getWrapperContainer
self.assertEqual(c(c(c(c(testOb)))), self.rootFolder)
self.assertEqual(testOb, ts)
def test_get(self):
self.createServiceManager()
sm = traverse(self.rootFolder, '++etc++Services')
default = traverse(sm, 'Packages/default')
ts = TestService()
default.setObject('test_service1', ts)
configuration = ServiceConfiguration(
'test_service',
'/++etc++Services/Packages/default/test_service1')
default['configure'].setObject('', configuration)
traverse(default, 'configure/1').status = Active
testOb = sm.get('test_service')
self.assertEqual(testOb, ts)
testOb = sm.get('test_service2')
self.assertEqual(testOb, None)
def testAddService(self):
self.createServiceManager()
sm = traverse(self.rootFolder, '++etc++Services')
default = traverse(sm, 'Packages/default')
ts1 = TestService()
default.setObject('test_service1', ts1)
configuration = ServiceConfiguration(
'test_service',
'/++etc++Services/Packages/default/test_service1')
default['configure'].setObject('', configuration)
traverse(default, 'configure/1').status = Active
ts2 = TestService()
default.setObject('test_service2', ts2)
configuration = ServiceConfiguration(
'test_service',
'/++etc++Services/Packages/default/test_service2')
default['configure'].setObject('', configuration)
traverse(default, 'configure/2').status = Registered
testOb = getService(self.rootFolder, 'test_service')
self.assertEqual(testOb, ts1)
def testUnbindService(self):
root_ts = TestService()
gsm = getServiceManager(None)
gsm.provideService('test_service', root_ts)
self.testGetService() # set up localservice
sm = traverse(self.rootFolder, '++etc++Services')
traverse(sm, 'Packages/default/configure/1').status = Unregistered
self.assertEqual(getService(self.rootFolder, 'test_service'), root_ts)
def testContextServiceLookup(self):
self.testGetService() # set up localservice
sm=getServiceManager(self.rootFolder)
self.assertEqual(getService(self.folder1_1, 'test_service'),
sm.Packages['default']['test_service1'])
def testContextServiceLookupWithMultipleServiceManagers(self):
self.testGetService() # set up root localservice
sm=getServiceManager(self.rootFolder)
self.folder1.setServiceManager(ServiceManager())
sm2=getServiceManager(self.folder1)
self.assertEqual(getService(self.folder1, 'test_service'),
sm.Packages['default']['test_service1'])
def testComponentArchitectureServiceLookup(self):
self.rootFolder.setServiceManager(ServiceManager())
self.folder1.setServiceManager(ServiceManager())
ts = TestService()
globsm=getServiceManager(None)
globsm.provideService('test_service', ts)
service = getService(self.folder1, 'test_service')
self.assertEqual(service, ts)
def test_resolve(self):
from zope.proxy.context.context import ContextWrapper as cw
from zope.app.services.module import Manager
import Zope.App.OFS.Services.ServiceManager.tests.Sample1
import Zope.App.OFS.Services.ServiceManager.tests.Sample2
self.rootFolder.setServiceManager(ServiceManager())
sm=getServiceManager(self.rootFolder)
Packages = cw(sm.Packages, sm, name='Packages')
default = cw(Packages['default'], Packages, name='Packages')
default.setObject('m1', Manager())
manager = cw(default['m1'], default, name='m1')
manager.new('Zope.App.OFS.Services.ServiceManager.tests.Sample1',
'x = "root m1"\n')
default.setObject('m2', Manager())
manager = cw(default['m2'], default, name='m1')
manager.new('XXX.ZZZ', 'x = "root m2"\nZZZ = 42\n')
self.folder1.setServiceManager(ServiceManager())
sm2=getServiceManager(self.folder1)
Packages = cw(sm2.Packages, sm2, name='Packages')
default = cw(Packages['default'], Packages, name='Packages')
default.setObject('m1', Manager())
manager = cw(default['m1'], default, name='m1')
manager.new('Zope.App.OFS.Services.ServiceManager.tests.Sample1',
'x = "folder1 m1 1"')
self.assertEqual(
sm2.resolve("Zope.App.OFS.Services.ServiceManager.tests.Sample1.x"),
"folder1 m1 1")
self.assertEqual(
sm.resolve("Zope.App.OFS.Services.ServiceManager.tests.Sample1.x"),
"root m1")
self.assertEqual(
sm2.resolve("Zope.App.OFS.Services.ServiceManager.tests.Sample2.y"),
"sample 2")
self.assertEqual(
sm.resolve("Zope.App.OFS.Services.ServiceManager.tests.Sample2.y"),
"sample 2")
self.assertEqual(sm.resolve("XXX.ZZZ.ZZZ"), 42)
self.assertEqual(sm.resolve("XXX.ZZZ."), 42)
self.assertEqual(sm.resolve("XXX.ZZZ.x"), "root m2")
self.assertEqual(sm2.resolve("XXX.ZZZ.ZZZ"), 42)
self.assertEqual(sm2.resolve("XXX.ZZZ."), 42)
self.assertEqual(sm2.resolve("XXX.ZZZ.x"), "root m2")
def test_suite():
loader=TestLoader()
return loader.loadTestsFromTestCase(ServiceManagerTests)
if __name__=='__main__':
TextTestRunner().run(test_suite())
=== Added File Zope3/src/zope/app/services/tests/test_user.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_user.py,v 1.1.2.1 2002/12/23 19:32:24 jim Exp $
"""
from unittest import TestCase, TestSuite, main, makeSuite
from zope.app.services.auth import User
class UserTest(TestCase):
def setUp(self):
self._user = User('srichter', 'Stephan Richter',
'Local Authentication Service Developer',
'srichter', 'hello')
def testGetLogin(self):
user = self._user
self.assertEqual('srichter', user.getLogin())
def testGetRoles(self):
user = self._user
self.assertEqual([], user.getRoles())
def testValidate(self):
user = self._user
self.assertEqual(1, user.validate('hello'))
def testGetId(self):
user = self._user
self.assertEqual('srichter', user.getId())
def testGetTitle(self):
user = self._user
self.assertEqual('Stephan Richter', user.getTitle())
def testGetDescription(self):
user = self._user
self.assertEqual('Local Authentication Service Developer',
user.getDescription())
def testSetTitle(self):
user = self._user
user.setTitle('Stephan 2')
self.assertEqual('Stephan 2', user.getTitle())
def testSetDescription(self):
user = self._user
user.setDescription('Programmer')
self.assertEqual('Programmer', user.getDescription())
def testSetLogin(self):
user = self._user
user.setLogin('srichter2')
self.assertEqual('srichter', user.getLogin())
def testSetRoles(self):
# XXX Needs a test
user = self._user
def testSetPassword(self):
user = self._user
user.setPassword('hello2')
self.assertEqual(1, user.validate('hello2'))
def test_suite():
return makeSuite(UserTest)
if __name__=='__main__':
main(defaultTest='test_suite')
=== Added File Zope3/src/zope/app/services/tests/test_view.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.
#
##############################################################################
"""Test the view module
$Id: test_view.py,v 1.1.2.1 2002/12/23 19:32:24 jim Exp $
"""
from unittest import TestCase, TestSuite, main, makeSuite
from zope.app.services.tests.testingiconfigurable import TestingIConfigurable
from zope.app.services.view import ViewService
from zope.interface import Interface
from zope.proxy.context.context import ContextWrapper
from zope.component.exceptions import ComponentLookupError
from zope.app.services.tests.placefulsetup \
import PlacefulSetup
from zope.app.services.service import ServiceManager
from zope.app.services.view import ViewConfiguration
from zope.app.content.folder import RootFolder
from zope.component import getServiceManager
from Zope.App.Traversing import traverse
from zope.component.interfaces import IServiceService
from zope.component.view import provideView
from zope.publisher.browser import TestRequest
from zope.publisher.interfaces.browser import IBrowserPresentation
from zope.app.interfaces.services.interfaces import IZPTTemplate
from zope.app.services.view import PageConfiguration, BoundTemplate
from zope.interface.verify import verifyObject
from zope.component.interfaces import IViewService
class I1(Interface):
pass
class I1E(I1):
pass
I2 = IBrowserPresentation
class I3(Interface):
pass
class I4(Interface):
pass
class Configuration:
forInterface = I1
presentationType = I2
viewName = 'test'
layer = 'default'
def getView(self, object, request):
return self.factory(object, request)
def activated(self): pass
def deactivated(self): pass
class C: pass
class A:
def __init__(self, object, request):
self.context = object
self.request = request
class TestViewService(PlacefulSetup, TestingIConfigurable, TestCase):
def setUp(self):
PlacefulSetup.setUp(self)
self.buildFolders()
self.rootFolder.setServiceManager(ServiceManager())
self._service = ContextWrapper(ViewService(), self.rootFolder)
def test_implements_IViewService(self):
from zope.component.interfaces import IViewService
from zope.interface.verify import verifyObject
verifyObject(IViewService, self._service)
def createTestingConfigurable(self):
return ContextWrapper(ViewService(), C())
def createTestingConfiguration(self):
return Configuration()
def test_implements_IViewService(self):
verifyObject(IViewService, ViewService())
def test_queryView_no_view(self):
service = self._service
class O:
__implements__ = I1
o = O()
request = TestRequest()
self.assertEqual(service.queryView(o, 'test', request), None)
self.assertEqual(service.queryView(o, 'test', request, 42), 42)
def test_getView_no_view(self):
service = self._service
class O:
__implements__ = I1
o = O()
request = TestRequest()
self.assertRaises(ComponentLookupError,
service.getView, O(), 'test', request)
def test_queryView_and_getView(self):
service = self._service
sm = traverse(self.rootFolder, '++etc++Services')
configure = traverse(sm, 'Packages/default/configure')
configuration = Configuration()
configure.setObject('', configuration)
configuration = traverse(configure, '1')
class O:
__implements__ = I1
configuration.factory = A
registry = service.createConfigurationsFor(configuration)
registry.register(configuration)
registry.activate(configuration)
o = O()
request = TestRequest()
for m in 'queryView', 'getView':
for r in I1, I1E:
o = O()
o.__implements__ = r
view = getattr(service, m)(o, 'test', request)
self.assertEqual(view.__class__, A)
self.assertEqual(view.context, o)
self.assertEqual(view.request, request)
def test_queryView_delegation(self):
service = self._service
self.buildFolders()
self.rootFolder.setServiceManager(ServiceManager())
sm = traverse(self.rootFolder, '++etc++Services')
configure = traverse(sm, 'Packages/default/configure')
configuration = Configuration()
configure.setObject('', configuration)
configuration = traverse(configure, '1')
class O:
__implements__ = I1
o = O()
request = TestRequest()
class A2(A): pass
provideView(I1, 'test', IBrowserPresentation, A2)
view = service.queryView(o, 'test', request)
self.assertEqual(view.__class__, A2)
self.assertEqual(view.context, o)
self.assertEqual(view.request, request)
def test_getRegisteredMatching(self):
self.test_queryView_and_getView()
registry = self._service.queryConfigurationsFor(Configuration())
for args in ((), (I1E, ), (None, I2), (I1E, I2), ):
r = self._service.getRegisteredMatching(*args)
self.assertEqual(list(r), [(I1, I2, registry, 'default', 'test')])
class PhonyServiceManager(ServiceManager):
__implements__ = IServiceService
def resolve(self, name):
if name == 'Foo.Bar.A':
return A
class TestViewConfiguration(PlacefulSetup, TestCase):
def setUp(self):
PlacefulSetup.setUp(self)
rootFolder = RootFolder()
rootFolder.setServiceManager(PhonyServiceManager())
self.configuration = ContextWrapper(
ViewConfiguration(I1, 'test', IBrowserPresentation, "Foo.Bar.A"),
rootFolder,
)
def test_getView(self):
c = C()
request = TestRequest()
view = self.configuration.getView(c, request)
self.assertEqual(view.__class__, A)
self.assertEqual(view.context, c)
self.assertEqual(view.request, request)
self.assertEqual(self.configuration.forInterface, I1)
self.assertEqual(self.configuration.presentationType, I2)
class PhonyTemplate:
__implements__ = IZPTTemplate
class TestPageConfiguration(PlacefulSetup, TestCase):
def setUp(self):
PlacefulSetup.setUp(self)
rootFolder = RootFolder()
rootFolder.setServiceManager(PhonyServiceManager())
default = traverse(rootFolder, '++etc++Services/Packages/default')
self.__template = PhonyTemplate()
default.setObject('t', self.__template)
self.__configuration = ContextWrapper(
PageConfiguration(I1, 'test', IBrowserPresentation,
"Foo.Bar.A",
'/++etc++Services/Packages/default/t',
),
rootFolder,
)
def test_getView(self):
c = C()
request = TestRequest()
view = self.__configuration.getView(c, request)
self.assertEqual(view.__class__, BoundTemplate)
self.assertEqual(view.template, self.__template)
view = view.view
self.assertEqual(view.__class__, A)
self.assertEqual(view.context, c)
self.assertEqual(view.request, request)
self.assertEqual(self.__configuration.forInterface, I1)
self.assertEqual(self.__configuration.presentationType, I2)
def test_suite():
return TestSuite((
makeSuite(TestViewService),
makeSuite(TestViewConfiguration),
makeSuite(TestPageConfiguration),
))
if __name__=='__main__':
main(defaultTest='test_suite')
=== Added File Zope3/src/zope/app/services/tests/test_viewpackage.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.
#
##############################################################################
"""View package tests.
$Id: test_viewpackage.py,v 1.1.2.1 2002/12/23 19:32:24 jim Exp $
"""
from unittest import TestCase, TestSuite, main, makeSuite
from zope.app.services.tests.placefulsetup \
import PlacefulSetup
from zope.app.services.viewpackage import ViewPackage
from Zope.App.Traversing import traverse
from zope.app.services.zpt import ZPTTemplate
from zope.app.services.view import ViewService
from zope.app.interfaces.services.configuration import Active
from zope.app.services.service import ServiceManager
from zope.app.services.service \
import ServiceConfiguration
from zope.interface import Interface
class Test(PlacefulSetup, TestCase):
def test_setObject(self):
self.buildFolders()
self.rootFolder.setServiceManager(ServiceManager())
default = traverse(self.rootFolder, '++etc++Services/Packages/default')
default.setObject('Views', ViewPackage())
views = traverse(default, 'Views')
views.forInterface = Interface
views.factoryName = None
#set up view service
default.setObject('ViewService', ViewService())
configure = traverse(default, 'configure')
configuration = ServiceConfiguration(
'Views',
'/++etc++Services/Packages/default/ViewService')
configure.setObject('', configuration)
configuration = traverse(configure, '1')
configuration.status = Active
views.setObject('foo.html', ZPTTemplate())
configuration = traverse(views, 'configure/1')
self.assertEqual(configuration.status, Active)
self.assertRaises(TypeError,
views.setObject, 'bar.html', ViewPackage())
def test_suite():
return TestSuite((
makeSuite(Test),
))
if __name__=='__main__':
main(defaultTest='test_suite')
=== Added File Zope3/src/zope/app/services/tests/testingconfigurationregistry.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.
#
##############################################################################
__metaclass__ = type
class TestingConfiguration:
def __init__(self, id):
self.id = id
def __eq__(self, other):
return self.id == getattr(other, 'id', 0)
class TestingConfigurationRegistry:
class_ = TestingConfiguration
def __init__(self, *args):
self._data = args
def register(self, configuration):
cid = configuration.id
if self._data:
if cid in self._data:
return # already registered
else:
# Nothing registered. Need to stick None in front so that nothing
# is active.
self._data = (None, )
self._data += (cid, )
def unregister(self, configuration):
cid = configuration.id
data = self._data
if data:
if data[0] == cid:
# It's active, we need to switch in None
self._data = (None, ) + data[1:]
else:
self._data = tuple([item for item in data if item != cid])
def registered(self, configuration):
cid = configuration.id
return cid in self._data
def activate(self, configuration):
cid = configuration.id
if self._data[0] == cid:
return # already active
if self._data[0] is None:
# Remove leading None marker
self._data = self._data[1:]
self._data = (cid, ) + tuple(
[item for item in self._data if item != cid]
)
def deactivate(self, configuration):
cid = configuration.id
if self._data[0] != cid:
return # already inactive
# Just stick None on the front
self._data = (None, ) + self._data
def active(self):
if self._data:
return self.class_(self._data[0])
return None
def __nonzero__(self):
return bool(self._data)
def info(self):
result = [{'id': path,
'active': False,
'configuration': self.class_(path),
}
for path in self._data
]
if result:
if result[0]['configuration'] is None:
del result[0]
else:
result[0]['active'] = True
return result
=== Added File Zope3/src/zope/app/services/tests/testingiconfigurable.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.
#
##############################################################################
"""XXX short summary goes here.
XXX longer description goes here.
$Id: testingiconfigurable.py,v 1.1.2.1 2002/12/23 19:32:24 jim Exp $
"""
from zope.app.interfaces.services.configuration import IConfigurable
from zope.interface.verify import verifyObject
from zope.proxy.context.context import getWrapperContainer
class TestingIConfigurable:
"""Base class for testing implementors of IConfigurable
Subclasses must implement:
- createTestingConfigurable()
that returns a new configurable object with no configurations.
This configuration object must be in the context of something
that is not None.
- createTestingConfiguration()
that returns a configuration object.
"""
def _assertInContext(self, ob, parent):
"""Assert that we have the proper context
The container of ob must be the parent, and the parent must
have some context.
"""
self.assertEqual(getWrapperContainer(ob), parent)
self.failIf(getWrapperContainer(getWrapperContainer(ob)) is None)
def test_implements_IConfigurable(self):
verifyObject(IConfigurable, self.createTestingConfigurable())
def test_queryConfigurationsFor_no_config(self):
configurable = self.createTestingConfigurable()
configuration = self.createTestingConfiguration()
self.failIf(configurable.queryConfigurationsFor(configuration))
self.assertEqual(
configurable.queryConfigurationsFor(configuration, 42),
42)
def test_createConfigurationsFor(self):
configurable = self.createTestingConfigurable()
configuration = self.createTestingConfiguration()
registry = configurable.createConfigurationsFor(configuration)
self.assertEqual(getWrapperContainer(registry), configurable)
# If we call it again, we should get the same object
self.assertEqual(configurable.createConfigurationsFor(configuration),
registry)
self._assertInContext(registry, configurable)
return registry
def test_queryConfigurationsFor(self):
configurable = self.createTestingConfigurable()
configuration = self.createTestingConfiguration()
cregistry = configurable.createConfigurationsFor(configuration)
registry = configurable.queryConfigurationsFor(configuration)
self.assertEqual(registry, cregistry)
self._assertInContext(registry, configurable)
registry = configurable.queryConfigurationsFor(configuration, 42)
self.assertEqual(registry, cregistry)
self._assertInContext(registry, configurable)
=== Added File Zope3/src/zope/app/services/tests/testingservicemanager.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: testingservicemanager.py,v 1.1.2.1 2002/12/23 19:32:24 jim Exp $
"""
__metaclass__ = type
from zope.component.interfaces import IServiceService
from zope.app.component.nextservice \
import getNextService, getNextServiceManager
from zope.proxy.context.context import ContextWrapper
from zope.app.interfaces.services.service import IBindingAware
from zope.proxy.context import ContextMethod
class TestingServiceManager:
"""Simple placeful service manager used for writing tests
"""
__implements__ = IServiceService
def getServiceDefinitions(self):
"See Zope.ComponentArchitecture.IServiceService.IServiceService"
return getNextServiceManager(self).getServiceDefinitions()
def getInterfaceFor(self, name):
"See Zope.ComponentArchitecture.IServiceService.IServiceService"
return getNextServiceManager(self).getInterfaceFor(name)
def getService(self, name):
"See Zope.ComponentArchitecture.IServiceService.IServiceService"
if hasattr(self, name):
return ContextWrapper(getattr(self, name), self, name=name)
return getNextServiceManager(self).getService(name)
getService = ContextMethod(getService)
def queryService(self, name, default=None):
"See Zope.ComponentArchitecture.IServiceService.IServiceService"
if hasattr(self, name):
return ContextWrapper(getattr(self, name), self, name=name)
return getNextServiceManager(self).queryService(name, default)
queryService = ContextMethod(queryService)
def bindService(self, name, ob):
setattr(self, name, ob)
if IBindingAware.isImplementedBy(ob):
ob.bound(name)
bindService = ContextMethod(bindService)
def unbindService(self, name):
ob = getattr(self, name)
if IBindingAware.isImplementedBy(ob):
ob.unbound(name)
delattr(self, name, ob)
unbindService = ContextMethod(unbindService)
__doc__ = TestingServiceManager.__doc__ + __doc__