[Zope3-checkins] CVS: Zope3/src/zope/app/site/tests - __init__.py:1.1 placefulsetup.py:1.1 test_folder.py:1.1 test_servicemanager.py:1.1 test_serviceregistration.py:1.1

[Stephan Richter]

Update of /cvs-repository/Zope3/src/zope/app/site/tests
In directory cvs.zope.org:/tmp/cvs-serv17238/src/zope/app/site/tests

Added Files:
	__init__.py placefulsetup.py test_folder.py 
	test_servicemanager.py test_serviceregistration.py 
Log Message:


Move Local Service Manager and site code to zope.app.site.


=== Added File Zope3/src/zope/app/site/tests/__init__.py ===
# Import this.


=== Added File Zope3/src/zope/app/site/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.
#
##############################################################################
"""Base Mix-in class for Placeful Setups 

$Id: placefulsetup.py,v 1.1 2004/03/13 15:21:57 srichter Exp $
"""
from zope.app import zapi
from zope.app.tests import setup
from zope.app.tests.placelesssetup import PlacelessSetup
from zope.app.folder import rootFolder

class Place(object):

    def __init__(self, path):
        self.path = path

    def __get__(self, inst, cls=None):
        if inst is None:
            return self

        try:
            # Use __dict__ directly to avoid infinite recursion
            root = inst.__dict__['rootFolder']
        except KeyError:
            root = inst.rootFolder = setup.buildSampleFolderTree()

        return zapi.traverse(root, self.path)

class PlacefulSetup(PlacelessSetup):

    # Places :)
    rootFolder  = Place('')

    folder1     = Place('folder1')
    folder1_1   = Place('folder1/folder1_1')
    folder1_1_1 = Place('folder1/folder1_1/folder1_1_1')
    folder1_1_2 = Place('folder1/folder1_2/folder1_1_2')
    folder1_2   = Place('folder1/folder1_2')
    folder1_2_1 = Place('folder1/folder1_2/folder1_2_1')

    folder2     = Place('folder2')
    folder2_1   = Place('folder2/folder2_1')
    folder2_1_1 = Place('folder2/folder2_1/folder2_1_1')


    def setUp(self, folders=False, site=False):
        setup.placefulSetUp()
        if folders or site:
            return self.buildFolders(site)

    def tearDown(self):
        setup.placefulTearDown()
        # clean up folders and placeful service managers and services too?

    def buildFolders(self, site=False):
        self.rootFolder = setup.buildSampleFolderTree()
        if site:
            return self.makeSite()

    def makeSite(self, path='/'):
        folder = zapi.traverse(self.rootFolder, path)
        return setup.createServiceManager(folder)

    def createRootFolder(self):
        self.rootFolder = rootFolder()

    # The following is a hook that some base classes might want to override.
    def getObjectHub(self):
        from zope.app.hub import ObjectHub
        return ObjectHub()

    def createStandardServices(self):
        '''Create a bunch of standard placeful services'''

        setup.createStandardServices(self.rootFolder,
                                     hubids=self.getObjectHub())



=== Added File Zope3/src/zope/app/site/tests/test_folder.py ===
##############################################################################
#
# Copyright (c) 2003 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.
#
##############################################################################
"""Site-management folder tests

$Id: test_folder.py,v 1.1 2004/03/13 15:21:57 srichter Exp $
"""
import unittest
from zope.app.services.tests.test_registrationmanager \
     import RegistrationManagerContainerTests


class TestSomething(RegistrationManagerContainerTests, unittest.TestCase):
    "Test registration manager access"
        

def test_suite():
    suite = unittest.TestSuite()
    suite.addTest(unittest.makeSuite(TestSomething))
    return suite


if __name__ == '__main__':
    unittest.main()


=== Added File Zope3/src/zope/app/site/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.
#
##############################################################################
"""Service Manager Tests

$Id: test_servicemanager.py,v 1.1 2004/03/13 15:21:57 srichter Exp $
"""
from unittest import TestCase, TestLoader, TextTestRunner

from zope.app import zapi
from zope.app.tests import setup
from zope.interface import Interface, implements
from zope.app.site.service import ServiceManager
from zope.app.site.service import ServiceRegistration
from zope.component import getService, getServiceManager
from zope.app.site.tests.placefulsetup import PlacefulSetup
from zope.app.traversing import traverse
from zope.app.interfaces.services.registration import UnregisteredStatus
from zope.app.interfaces.services.registration import ActiveStatus
from zope.app.interfaces.services.registration import RegisteredStatus
from zope.component.service import serviceManager
from zope.app.interfaces.annotation import IAttributeAnnotatable

class ITestService(Interface):
    pass

class TestService:
    implements(ITestService, IAttributeAnnotatable)

class ServiceManagerTests(PlacefulSetup, TestCase):

    def setUp(self):
        PlacefulSetup.setUp(self, site=True)
        serviceManager.defineService('test_service', ITestService)

    def testGetService(self):
        sm = traverse(self.rootFolder, '++etc++site')
        default = traverse(sm, 'default')

        ts = TestService()
        default['test_service1'] = ts
        registration = ServiceRegistration(
            'test_service',
            '/++etc++site/default/test_service1')

        rm = default.getRegistrationManager()
        name = rm.addRegistration(registration)
        traverse(rm, name).status = ActiveStatus

        testOb = getService(self.rootFolder, 'test_service')
        self.assertEqual(testOb.__parent__.__parent__.__parent__,
                         self.rootFolder)
        self.assertEqual(testOb, ts)
        # used by one of the callers
        return name

    def test_queryLocalService(self):
        sm = traverse(self.rootFolder, '++etc++site')

        # Test no service case
        self.assertEqual(sm.queryLocalService('test_service'), None)
        self.assertEqual(sm.queryLocalService('test_service', 42), 42)

        # Test Services special case
        self.assertEqual(sm.queryLocalService('Services', 42), sm)

        # Test found local
        default = traverse(sm, 'default')
        ts = TestService()
        default['test_service1'] = ts
        registration = ServiceRegistration(
            'test_service',
            '/++etc++site/default/test_service1')
        rm = default.getRegistrationManager()
        name = rm.addRegistration(registration)
        traverse(rm, name).status = ActiveStatus

        testOb = sm.queryLocalService('test_service')
        self.assertEqual(testOb.__parent__.__parent__.__parent__,
                         self.rootFolder)
        self.assertEqual(testOb, ts)


    def test_get(self):
        sm = traverse(self.rootFolder, '++etc++site')
        default = sm.get('default')
        self.assertEqual(default, sm['default'])
        self.assertEqual(sm.get('spam'), None)

    def testAddService(self):
        sm = traverse(self.rootFolder, '++etc++site')
        default = traverse(sm, 'default')

        ts1 = TestService()
        default['test_service1'] = ts1
        registration = ServiceRegistration(
            'test_service',
            '/++etc++site/default/test_service1')
        rm = default.getRegistrationManager()
        name = rm.addRegistration(registration)
        traverse(rm, name).status = ActiveStatus

        ts2 = TestService()
        default['test_service2'] = ts2
        registration = ServiceRegistration(
            'test_service',
            '/++etc++site/default/test_service2')
        name = rm.addRegistration(registration)
        traverse(rm, name).status = RegisteredStatus

        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)

        name = self.testGetService() # set up localservice

        sm = traverse(self.rootFolder, '++etc++site')
        cm = traverse(sm, 'default').getRegistrationManager()
        traverse(cm, name).status = UnregisteredStatus

        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['default']['test_service1'])

    def testContextServiceLookupWithMultipleServiceManagers(self):
        self.testGetService() # set up root localservice
        sm=getServiceManager(self.rootFolder)

        sm2 = self.makeSite('folder1')

        self.assertEqual(getService(self.folder1, 'test_service'),
                         sm['default']['test_service1'])

    def testComponentArchitectureServiceLookup(self):
        self.makeSite()
        self.makeSite('folder1')

        ts = TestService()

        globsm=getServiceManager(None)
        globsm.provideService('test_service', ts)

        service = getService(self.folder1, 'test_service')
        self.assertEqual(service, ts)

    def donttest_resolve(self):
        # XXX This test expects that the local module implementation
        # (the Manager class) to register itself.  This is no longer
        # intentional behavior; the right tests need to be determined.
        from zope.app.services.module import Manager
        import zope.app.services.tests.sample1
        import zope.app.services.tests.sample2

        sm = self.makeSite()

        default = contained(sm['default'], self.rootFolder, name='default')
        default['m1'] = Manager('zope.app.services.tests.sample1',
                                'x = "root m1"\n')
        manager = contained(default['m1'], default, name='m1')
        manager.execute()
        default['m2'] = Manager('XXX.ZZZ', 'x = "root m2"\nZZZ = 42\n')
        manager = contained(default['m2'], default, name='m2')
        manager.execute()

        self.folder1.setSiteManager(ServiceManager(self.folder1))
        sm2=getServiceManager(self.folder1)
        default = contained(sm2['default'], self.folder1, name='default')
        default['m1'] = Manager('zope.app.services.tests.sample1',
                                'x = "folder1 m1 1"')
        manager = contained(default['m1'], default, name='m1')
        manager.execute()

        self.assertEqual(
          sm2.resolve("zope.app.services.tests.sample1.x"),
          "folder1 m1 1")
        self.assertEqual(
          sm.resolve("zope.app.services.tests.sample1.x"),
          "root m1")

        self.assertEqual(
          sm2.resolve("zope.app.services.tests.sample2.y"),
          "sample 2")
        self.assertEqual(
          sm.resolve("zope.app.services.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_site_manager_connections(self):
        root = self.rootFolder
        mr = root.getSiteManager()
        m1 = setup.createServiceManager(zapi.traverse(root, 'folder1')) 
        m2 = setup.createServiceManager(zapi.traverse(root, 'folder2'))
        m111 = setup.createServiceManager(
            zapi.traverse(root, 'folder1/folder1_1/folder1_1_1'))
        self.assertEqual(m1.next, mr)
        self.assertEqual(m2.next, mr)
        self.assertEqual(m111.next, m1)
        self.assertEqual(mr.subSites, (m1, m2))
        self.assertEqual(m1.subSites, (m111, ))

        # Now insert a site and make sure everything is still right:
        m11 = setup.createServiceManager(
            zapi.traverse(root, 'folder1/folder1_1'))
        self.assertEqual(m11.next, m1)
        self.assertEqual(m111.next, m11)
        self.assertEqual(m1.subSites, (m11, ))
        self.assertEqual(m11.subSites, (m111, ))

        


def test_suite():
    loader=TestLoader()
    return loader.loadTestsFromTestCase(ServiceManagerTests)

if __name__=='__main__':
    TextTestRunner().run(test_suite())


=== Added File Zope3/src/zope/app/site/tests/test_serviceregistration.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.
#
##############################################################################
"""Service Registration tests.

$Id: test_serviceregistration.py,v 1.1 2004/03/13 15:21:57 srichter Exp $
"""

from unittest import TestCase, main, makeSuite

from zope.interface import Interface, implements

from zope.component import getServiceManager
from zope.app.traversing import traverse, getPath
from zope.app.site.service import ServiceRegistration
from zope.app.site.tests.placefulsetup import PlacefulSetup
from zope.component.service import defineService
from zope.app.site.interfaces import IBindingAware
from zope.app.interfaces.services.registration import ActiveStatus
from zope.app.interfaces.services.registration import RegisteredStatus
from zope.app.interfaces.services.registration import IRegistered
from zope.app.site.interfaces import ISimpleService

from zope.app.interfaces.dependable import IDependable, DependencyError

class ITestService(Interface):
    pass

class TestServiceBase:
    __name__ = __parent__ = None
    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 TestService(TestServiceBase):
    implements(ISimpleService)

class Test(PlacefulSetup, TestCase):

    def setUp(self):
        PlacefulSetup.setUp(self, site=True)

        defineService('test_service', ITestService)

        default = traverse(self.rootFolder,
                           '++etc++site/default')
        self.__default = default

        default['c'] = TestService()

        registration = ServiceRegistration(
            'test_service', '/++etc++site/default/c')

        self.__c = traverse(default, 'c')
        self.__cm = default.getRegistrationManager()

        self.__registration_name = self.__cm.addRegistration(registration)

        self.__config = traverse(self.__cm, self.__registration_name)
        self.__configpath = getPath(self.__config)

    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.services.tests.testregistrations, but in a different way

    def test_getComponent(self):
        self.assertEqual(self.__config.getComponent(), self.__c)

    def test_addNotify(self):
        self.assertEqual(self.__c._dependents,
                         (self.__configpath, ))
        u = IRegistered(self.__c)
        self.assertEqual(list(u.usages()),
                         [self.__configpath])

    def test_removeNotify_and_unregistered(self):
        self.__config.status = RegisteredStatus

        sm = getServiceManager(self.__default)
        registry = sm.queryRegistrationsFor(self.__config)
        self.failUnless(registry, "The components should be registered")

        del self.__cm[self.__registration_name]
        self.assertEqual(self.__c._dependents, ())
        u = IRegistered(self.__c)
        self.assertEqual(len(u.usages()), 0)

        self.failIf(registry, "The components should not be registered")

    def test_disallow_delete_when_active(self):
        self.__config.status = ActiveStatus
        try:
            del self.__cm[self.__registration_name]
        except DependencyError:
            pass # OK
        else:
            self.fail("Should have gotten a dependency error")

    def test_not_a_local_service(self):
        defineService('test_service_2', ITestService)
        self.__default['c2'] = TestServiceBase()

        self.assertRaises(
            TypeError,
            ServiceRegistration,
            'test_service',
            '/++etc++site/default/c2',
            self.__default
            )


def test_suite():
    return makeSuite(Test)

if __name__=='__main__':
    main(defaultTest='test_suite')