[Zope3-checkins] CVS: Zope3/src/zope/app/hub/tests - __init__.py:1.1 objecthubsetup.py:1.1 test_hookedhubevent.py:1.1 test_hubevent.py:1.1 test_objecthub.py:1.1 test_registration.py:1.1

Stephan Richter srichter at cosmos.phy.tufts.edu
Thu Mar 11 04:19:27 EST 2004


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

Added Files:
	__init__.py objecthubsetup.py test_hookedhubevent.py 
	test_hubevent.py test_objecthub.py test_registration.py 
Log Message:


Moved object hub to zope.app.hub. I did provide module aliases.


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


=== Added File Zope3/src/zope/app/hub/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.
#
##############################################################################
"""
$Id: objecthubsetup.py,v 1.1 2004/03/11 09:19:25 srichter Exp $
"""
from zope.app import zapi
from zope.app.event.tests.eventsetup import EventSetup
from zope.app.services.servicenames import HubIds
from zope.app.traversing import traverse, canonicalPath

from zope.app.container.interfaces import IObjectAddedEvent, IObjectMovedEvent
from zope.app.event.interfaces import ISubscriber

from zope.interface import implements

class LoggingSubscriber:
    # XXX Jim mentioned there is a new generic
    # version of this in zope.app somewhere...

    implements(ISubscriber)

    def __init__(self):
        self.events_received = []

    def notify(self, event):
        self.events_received.append(event)

    def verifyEventsReceived(self, testcase, event_spec_list):
        # iterate through self.events_received and check
        # that each one implements the interface that is
        # in the same place, with the same location and hub id

        testcase.assertEqual(len(event_spec_list), len(self.events_received))

        for spec,event in zip(event_spec_list, self.events_received):
            if len(spec)==4:
                interface,hubid,location,obj = spec
            elif len(spec)==3:
                interface,hubid,location = spec
                obj = None
            elif len(spec)==2:
                interface, location = spec
                obj = None
                hubid = None
            location = canonicalPath(location)
            testcase.assert_(interface.providedBy(event),
                             'Interface %s' % interface.getName())
            testcase.assertEqual(canonicalPath(event.object), location)

            if obj is not None:
                testcase.assertEqual(event.object, obj)

            # Sometimes, the test won't care about the hubid. In this case,
            # it is passed into the spec as None.
            if hubid is not None:
                testcase.assertEqual(event.hubid, hubid)

        self.events_received = []

class RegistrationSubscriber(LoggingSubscriber):
    def __init__(self, objectHub):
        LoggingSubscriber.__init__(self)
        self.hub = objectHub

    def notify(self, event):
        LoggingSubscriber.notify(self, event)
        # The policy is to register on object adds and object copies.
        if IObjectAddedEvent.providedBy(event):
            self.hub.register(event.object)

class ObjectHubSetup(EventSetup):

    def setUpRegistrationSubscriber(self):
        subscriber = RegistrationSubscriber(self.object_hub)
        self.rootFolder['registration_subscriber'] = subscriber
        self.subscriber = traverse(self.rootFolder, 'registration_subscriber')
        self.object_hub.subscribe(self.subscriber)

    def setUpLoggingSubscriber(self):
        subscriber = LoggingSubscriber()
        self.rootFolder['logging_subscriber'] = subscriber
        self.subscriber = traverse(self.rootFolder, 'logging_subscriber')
        self.object_hub.subscribe(self.subscriber)

    def setUp(self):
        EventSetup.setUp(self)
        self.object_hub = zapi.getService(self.rootFolder, HubIds)



=== Added File Zope3/src/zope/app/hub/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.
#
##############################################################################
"""
$Id: test_hookedhubevent.py,v 1.1 2004/03/11 09:19:25 srichter 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
from zope.app.hub.tests.objecthubsetup import ObjectHubSetup
from zope.app.hub import \
     ObjectRegisteredHubEvent, ObjectUnregisteredHubEvent, \
     ObjectModifiedHubEvent, ObjectMovedHubEvent, ObjectRemovedHubEvent
from zope.app.traversing import getPath

class AbstractTestHubEvent(ObjectHubSetup, unittest.TestCase):

    klass = None

    def setUp(self):
        ObjectHubSetup.setUp(self)
        self.obj = self.folder1_2_1
        self.hubid = self.object_hub.register(self.obj)
        self.location = getPath(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.obj = self.folder1_2_1
        self.hubid = self.object_hub.register(self.obj)
        self.location = getPath(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.obj = self.folder1_2_1
        self.hubid = self.object_hub.register(self.obj)
        self.location = getPath(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.obj = self.folder1_2_1
        self.hubid = self.object_hub.register(self.obj)
        self.location = getPath(self.obj)
        self.event = self.klass(self.object_hub, self.hubid)

class TestObjectMovedHubEvent(AbstractTestHubEvent):

    fromLocation = '/old/location'

    def setUp(self):
        ObjectHubSetup.setUp(self)
        self.obj = self.folder1_2_1
        self.hubid = self.object_hub.register(self.obj)
        self.location = getPath(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.obj = self.folder1_2_1
        self.hubid = self.object_hub.register(self.obj)
        self.location = getPath(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/hub/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.
#
##############################################################################
"""
$Id: test_hubevent.py,v 1.1 2004/03/11 09:19:25 srichter Exp $
"""
import unittest

from zope.app.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 getPath(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/hub/tests/test_objecthub.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.
#
##############################################################################
"""testObjectHub

$Id: test_objecthub.py,v 1.1 2004/03/11 09:19:25 srichter Exp $
"""
import unittest
from zope.app.tests import ztapi
from zope.app import zapi
from zope.app.hub.tests.objecthubsetup import ObjectHubSetup

from zope.app.event.interfaces import IObjectModifiedEvent
from zope.app.container.interfaces import IObjectAddedEvent
from zope.app.container.interfaces import IObjectRemovedEvent
from zope.app.container.interfaces import IObjectMovedEvent

from zope.app.container.contained import ObjectAddedEvent, ObjectRemovedEvent
from zope.app.container.contained import ObjectMovedEvent
from zope.app.event.objectevent import ObjectModifiedEvent, ObjectCreatedEvent

from zope.app.hub.interfaces import ObjectHubError
from zope.app.hub.interfaces import IObjectRemovedHubEvent
from zope.app.hub.interfaces import IObjectModifiedHubEvent
from zope.app.hub.interfaces import IObjectMovedHubEvent
from zope.app.hub.interfaces import IObjectRegisteredHubEvent
from zope.app.hub.interfaces import IObjectUnregisteredHubEvent

from zope.app.hub import ObjectModifiedHubEvent, ObjectRemovedHubEvent
from zope.app.hub import ObjectMovedHubEvent, ObjectRegisteredHubEvent
from zope.app.hub import ObjectUnregisteredHubEvent
from zope.app.hub import canonicalSlash, userPath

from zope.app.interfaces.traversing import IContainmentRoot
from zope.app.location import Location

from zope.exceptions import NotFoundError

from zope.app.traversing import canonicalPath, traverse

from zope.interface import implements, directlyProvides

from zope.app.container.contained import Contained

# 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

class TransmitHubEventTest(ObjectHubSetup, unittest.TestCase):
    hubid = 23
    location = '/folder1/folder1_1'
    # Don't test the HubEvent base class.
    # See below for testing subclasses / subinterfaces
    # klass = HubEvent
    # interface = IHubEvent

    def setUp(self):
        ObjectHubSetup.setUp(self)
        self.setUpLoggingSubscriber()
        self.obj = zapi.traverse(self.rootFolder, self.location)
        
        self.hub_event = self.klass(self.object_hub,
                                    self.hubid,
                                    self.location,
                                    self.obj)

    def testTransmittedEvent(self):
        # Test that the HubEvents are transmitted by the notify method
        self.object_hub.notify(self.hub_event)

        self.subscriber.verifyEventsReceived(self, [
                (self.interface, self.hubid, self.location)
            ])

class TransmitObjectRemovedHubEventTest(TransmitHubEventTest):
    interface = IObjectRemovedHubEvent
    klass = ObjectRemovedHubEvent

class TransmitObjectModifiedHubEventTest(TransmitHubEventTest):
    interface = IObjectModifiedHubEvent
    klass = ObjectModifiedHubEvent

class TransmitObjectMovedHubEventTest(TransmitHubEventTest):
    interface = IObjectMovedHubEvent
    klass = ObjectMovedHubEvent

    def setUp(self):
        ObjectHubSetup.setUp(self)
        self.setUpLoggingSubscriber()
        self.obj = zapi.traverse(self.rootFolder, self.location)
        self.hub_event = self.klass(
                self.object_hub, self.hubid,
                '/old/location', self.location, self.obj)

class TransmitObjectRegisteredHubEventTest(TransmitHubEventTest):
    interface = IObjectRegisteredHubEvent
    klass = ObjectRegisteredHubEvent

class TransmitObjectUnregisteredHubEventTest(TransmitHubEventTest):
    interface = IObjectUnregisteredHubEvent
    klass = ObjectUnregisteredHubEvent

class Folder(Location):
    pass

class BasicHubTest(ObjectHubSetup, unittest.TestCase):
    location = '/folder1/folder1_1'
    new_location = '/folder2/folder2_1'

    def setUp(self):
        ObjectHubSetup.setUp(self)
        self.setUpLoggingSubscriber()
        self.obj = zapi.traverse(self.rootFolder, self.location)
        self.setEvents()

    def setEvents(self):
        obj = self.obj
        self.created_event = ObjectCreatedEvent(obj)
        self.added_event = ObjectAddedEvent(obj)
        newobj = zapi.traverse(self.rootFolder, self.new_location)
        self.added_new_location_event = ObjectAddedEvent(newobj)
        self.removed_event = ObjectRemovedEvent(obj)
        self.modified_event = ObjectModifiedEvent(obj)
        self.moved_event = ObjectMovedEvent(
            obj,
            obj.__parent__, obj.__name__,
            newobj.__parent__, newobj.__name__,
            )

class TestRegistrationEvents(BasicHubTest):
    def testRegistration(self):
        self.assertRaises(NotFoundError,
                          self.object_hub.unregister,
                          self.location)
        self.assertRaises(NotFoundError, self.object_hub.unregister, 42)

        hubid = self.object_hub.register(self.location)
        hubid2 = self.object_hub.register(self.new_location)

        self.subscriber.verifyEventsReceived(self, [
                (IObjectRegisteredHubEvent, hubid, self.location),
                (IObjectRegisteredHubEvent, hubid2, self.new_location)
            ])

        # register again and check for error
        self.assertRaises(ObjectHubError,
                          self.object_hub.register,
                          self.location)

        # unregister first object by location
        self.object_hub.unregister(self.location)
        self.subscriber.verifyEventsReceived(self, [
                (IObjectUnregisteredHubEvent, hubid, self.location)
            ])
        # unregister second object by hub id
        self.object_hub.unregister(hubid2)
        self.subscriber.verifyEventsReceived(self, [
                (IObjectUnregisteredHubEvent, hubid2, self.new_location)
            ])

    def testRegistrationRelativeLocation(self):
        self.assertRaises(ValueError, self.object_hub.register, 'foo/bar')

class TestSearchRegistrations(BasicHubTest):
    locations = (
        '/',
        '/foo',
        '/foo/baq',
        '/foo/baq/baz',
        '/foo/bar',
        '/foo/bar/baz',
        '/foo/bar2/baz',
        '/foo/bar/baz2',
        '/foo/bar/baz3',
        '/foo/bas',
        '/foo/bas/baz',
        )

    def setUp(self):
        ObjectHubSetup.setUp(self)

    def testSearchAll(self):
        object_hub = self.object_hub
        location_hubid = [(location,
                           object_hub.register(location))
                          for location in self.locations]
        location_hubid.sort()

        r = list(object_hub.iterRegistrations())
        self.assertEqual(r, location_hubid)

    def testSearchSome(self):
        object_hub = self.object_hub
        location_hubid= [(location,
                           object_hub.register(location))
                          for location in self.locations
                          if location.startswith('/foo/bar/')]
        location_hubid.sort()

        r = list(object_hub.iterRegistrations('/foo/bar'))
        self.assertEqual(r, location_hubid)

    def testIterObjectRegistrations(self):
        class FakeObject:
            def __init__(self, location):
                self.location = location
            def __str__(self):
                return 'FakeObject at %s' % self.location
            def __eq__(self, other):
                return self.location == other.location

        def fake_object_for_location(location):
            return FakeObject(canonicalPath(location))

        from zope.app.interfaces.traversing import ITraverser
        from zope.app.traversing.adapters import Traverser
        class DummyTraverser(Traverser):
            implements(ITraverser)
            def traverse(self, location, *args, **kw):
                if location in TestSearchRegistrations.locations:
                    return fake_object_for_location(location)
                else:
                    return Traverser.traverse(self, location, *args, **kw)

        ztapi.provideAdapter(None, ITraverser, DummyTraverser)

        object_hub = self.object_hub
        location_hubid_object = [(location,
                                  object_hub.register(location),
                                  fake_object_for_location(location)
                                 )
                                 for location in self.locations]
        location_hubid_object.sort()

        r = [loc_id for loc_id in object_hub.iterObjectRegistrations()]
        r.sort()
        self.assertEqual(r, location_hubid_object)

class TestNoRegistration(BasicHubTest):

    def testAddWithoutRegistration(self):
        # Test that no HubIdEvents are generated if there is no registration
        hub = self.object_hub
        event = self.added_event
        location = self.location

        hub.notify(event)

        self.subscriber.verifyEventsReceived(self, [
                (IObjectAddedEvent, location),
            ])

class TestObjectCreatedEvent(BasicHubTest):
    def setUp(self):
        ObjectHubSetup.setUp(self)
        self.setUpRegistrationSubscriber()
        self.setEvents()

    def testLookingUpLocation(self):
        hub = self.object_hub
        event = self.created_event
        location = None

        hub.notify(event)

        self.assertEqual(location_from_hub, location)

        self.subscriber.verifyEventsReceived(self, [
                (IObjectCreatedEvent, location),
            ])

    def testLookupUpAbsentLocation(self):
        # Test that we don't find an hub id for location that we haven't added.
        hub = self.object_hub
        event = self.added_event
        location = self.location

        # Do not add the location to the hub
        # hub.notify(event)

        self.assertRaises(NotFoundError, hub.getHubId, location)

        self.subscriber.verifyEventsReceived(self, [])

class TestObjectAddedEvent(BasicHubTest):
    def setUp(self):
        ObjectHubSetup.setUp(self)
        self.setUpRegistrationSubscriber()
        self.obj = zapi.traverse(self.rootFolder, self.location)
        self.setEvents()

    def testLookingUpLocation(self):
        # Test that the location is in the lookup
        # Compare getHubIdForLocation and getLocationForHubId
        # Checks the sequence of events
        hub = self.object_hub
        event = self.added_event
        location = self.location

        hub.notify(event)

        hubid = hub.getHubId(location)
        # check that hub id is an int
        self.failUnless(isinstance(hubid, int)) # int(hubid)

        location_from_hub = hub.getPath(hubid)

        self.assertEqual(location_from_hub, location)

        self.subscriber.verifyEventsReceived(self, [
                (IObjectAddedEvent, location),
                (IObjectRegisteredHubEvent, hubid, location),
            ])

    def testLookupUpAbsentLocation(self):
        # Test that we don't find an hub id for location that we haven't added.
        hub = self.object_hub
        event = self.added_event
        location = self.location

        # Do not add the location to the hub
        # hub.notify(event)

        self.assertRaises(NotFoundError, hub.getHubId, location)

        self.subscriber.verifyEventsReceived(self, [])

    def testLookupUpAbsentHubId(self):
        # Test that we don't find a location for an hub id that isn't there.
        hub = self.object_hub
        event = self.added_event

        # Do not add the location to the hub
        # hub.notify(event)

        absent_hubid = 12

        self.assertRaises(NotFoundError, hub.getPath, absent_hubid)

        self.subscriber.verifyEventsReceived(self, [])


class TestObjectRemovedEvent(BasicHubTest):
    def setUp(self):
        ObjectHubSetup.setUp(self)
        self.obj = zapi.traverse(self.rootFolder, self.location)
        self.setUpRegistrationSubscriber()
        self.setEvents()

    def testRemovedLocation(self):
        # Test that a location that is added then removed is actually gone.
        hub = self.object_hub
        added_event = self.added_event
        removed_event = self.removed_event
        location = self.location
        obj = self.obj

        hub.notify(added_event)

        hubid = hub.getHubId(location)

        # check that hubid is an int
        self.failUnless(isinstance(hubid, int)) # int(hubid)

        hub.notify(removed_event)

        self.assertRaises(NotFoundError, hub.getHubId, location)
        self.assertRaises(NotFoundError, hub.getPath, hubid)

        self.subscriber.verifyEventsReceived(self, [
                (IObjectAddedEvent, location),
                (IObjectRegisteredHubEvent, hubid, location),
                (IObjectRemovedEvent, location),
                (IObjectRemovedHubEvent, hubid, location, obj),
            ])


    def testRemovedAbsentLocation(self):
        # Test that removing an absent location is silently ignored.
        hub = self.object_hub
        added_event = self.added_event
        removed_event = self.removed_event
        location = self.location

        # Do not add location
        # hub.notify(added_event)

        hub.notify(removed_event)

        self.subscriber.verifyEventsReceived(self, [
                (IObjectRemovedEvent, location),
            ])


    def testRemoveDescendants(self):
        # Test that removing an object also removes its descendants
        locations = (
            '/',
            '/1',
            '/1/1_1',
            '/1/1_1/1_1_1',
            '/2',
            '/2/2_1',
            '/2/2_2' )
        hub = self.object_hub
        for path in locations:
            hub.register(path)
        removed_event = self.removed_event
        removed_event.oldParent = '/'
        removed_event.oldName = '1'
        hub.notify(removed_event)
        newLocations = [path for (path, oid) in hub.iterRegistrations()]
        newLocations.sort()
        self.assertEqual([
            '/', 
            '/2', 
            '/2/2_1', 
            '/2/2_2'], 
            newLocations)


class TestObjectModifiedEvent(BasicHubTest):
    def setUp(self):
        ObjectHubSetup.setUp(self)
        self.setUpRegistrationSubscriber()
        self.obj = zapi.traverse(self.rootFolder, self.location)
        self.setEvents()

    def testModifiedLocation(self):
        # Test that lookup state does not change after an object modify event.
        hub = self.object_hub
        added_event = self.added_event
        modified_event = self.modified_event
        location = self.location

        hub.notify(added_event)

        hubid = hub.getHubId(location)
        # check that hubid is an int
        self.failUnless(isinstance(hubid, int)) # int(hubid)

        location_from_hub = hub.getPath(hubid)
        self.assertEqual(location_from_hub, location)

        hub.notify(modified_event)

        hubid2 = hub.getHubId(location)
        location_from_hub2 = hub.getPath(hubid2)

        self.assertEqual(location_from_hub, location_from_hub2)
        self.assertEqual(hubid, hubid2)

        self.subscriber.verifyEventsReceived(self, [
                (IObjectAddedEvent, location),
                (IObjectRegisteredHubEvent, hubid, location),
                (IObjectModifiedEvent, location),
                (IObjectModifiedHubEvent, hubid, location)
            ])


    def testModifiedAbsentLocation(self):
        # Test that lookup state does not change after an object modify event.
        # In this case, modify of an absent location is a noop.
        hub = self.object_hub
        added_event = self.added_event
        modified_event = self.modified_event
        location = self.location

        # Do not add location
        # hub.notify(added_event)

        hub.notify(modified_event)
        self.assertRaises(NotFoundError, hub.getHubId, location)

        self.subscriber.verifyEventsReceived(self, [
                (IObjectModifiedEvent, location),
            ])


class TestObjectMovedEvent(BasicHubTest):
    def setUp(self):
        ObjectHubSetup.setUp(self)
        self.setUpRegistrationSubscriber()
        self.obj = zapi.traverse(self.rootFolder, self.location)
        self.setEvents()

    def testMovedLocation(self):
        # Test that the location does indeed change after a move.
        hub = self.object_hub
        added_event = self.added_event
        moved_event = self.moved_event
        location = self.location
        new_location = self.new_location

        hub.notify(added_event)
        hubid = hub.getHubId(location)

        self.subscriber.verifyEventsReceived(self, [
                (IObjectAddedEvent, location),
                (IObjectRegisteredHubEvent, hubid, location),
            ])

        # simulate moving the object
        self.obj.__parent__ = zapi.traverse(self.rootFolder, "folder2")
        self.obj.__name__ = "folder2_1"

        hub.notify(moved_event)

        location_from_hub = hub.getPath(hubid)

        self.assertEqual(location_from_hub, new_location)
        self.assertRaises(NotFoundError, hub.getHubId, location)

        hubid2 = hub.getHubId(new_location)
        self.assertEqual(hubid2, hubid)

        self.subscriber.verifyEventsReceived(self, [
                (IObjectMovedEvent, new_location),
                (IObjectMovedHubEvent, hubid, new_location)
            ])


    def testMovedAbsentLocation(self):
        # Test that moving an absent location is a no-op.
        hub = self.object_hub
        added_event = self.added_event
        moved_event = self.moved_event
        location = self.location
        new_location = self.new_location

        # Do not add location
        # hub.notify(added_event)

        hub.notify(moved_event)
        self.assertRaises(NotFoundError, hub.getHubId, location)
        self.assertRaises(NotFoundError, hub.getHubId, new_location)

        # simulate moving the object
        self.obj.__parent__ = zapi.traverse(self.rootFolder, "folder2")
        self.obj.__name__ = "folder2_1"

        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.subscriber.verifyEventsReceived(self, [
                (IObjectAddedEvent, location),
                (IObjectRegisteredHubEvent, None, location),
                (IObjectAddedEvent, new_location),
                (IObjectRegisteredHubEvent, None, new_location),
            ])

        # simulate moving the object
        self.obj.__parent__ = zapi.traverse(self.rootFolder, "folder2")
        self.obj.__name__ = "folder2_1"

        self.assertRaises(ObjectHubError, hub.notify, moved_event)

        self.subscriber.verifyEventsReceived(self, [
                (IObjectMovedEvent, new_location),
            ])

    def testRepathDescendantsOnRename(self):
        # Test that the paths of descendants of renamed objects are updated
        locations = (
            '/',
            '/1',
            '/1/1_1',
            '/1/1_1/1_1_1',
            '/2',
            '/2/2_1',
            '/2/2_2' )
        hub = self.object_hub
        for path in locations:
            hub.register(path)
        moved_event = self.moved_event
        moved_event.oldParent = '/'
        moved_event.oldName = '1'
        moved_event.newParent = '/'
        moved_event.newName = '3'
        hub.notify(moved_event)
        newLocations = [path for (path, oid) in hub.iterRegistrations()]
        newLocations.sort()
        self.assertEqual([
            '/',
            '/2',
            '/2/2_1',
            '/2/2_2',
            '/3',
            '/3/1_1',
            '/3/1_1/1_1_1'],
            newLocations)

    def testRepathDescendantsOnMove(self):
        # Test that the paths of descendants of moved objects are updated
        locations = (
            '/',
            '/1',
            '/1/1_1',
            '/1/1_1/1_1_1',
            '/2',
            '/2/2_1',
            '/2/2_2' )
        hub = self.object_hub
        for path in locations:
            hub.register(path)
        moved_event = self.moved_event
        moved_event.oldParent = '/'
        moved_event.oldName = '1'
        moved_event.newParent = '/2/2_1'
        moved_event.newName = '1'
        hub.notify(moved_event)
        newLocations = [path for (path, oid) in hub.iterRegistrations()]
        newLocations.sort()
        self.assertEqual([
            '/',
            '/2',
            '/2/2_1',
            '/2/2_1/1',
            '/2/2_1/1/1_1',
            '/2/2_1/1/1_1/1_1_1',
            '/2/2_2'],
            newLocations)

class TestPathFunctions(BasicHubTest):

    def testCanonicalSlash(self):
        self.assertEqual(canonicalSlash('/'), '/')
        self.assertEqual(canonicalSlash('/', 'bar'), '/bar/')
        self.assertEqual(canonicalSlash('/foo'), '/foo/')
        self.assertEqual(canonicalSlash('/foo', 'bar'), '/foo/bar/')
        self.assertRaises(ValueError, canonicalSlash, '\\')
        self.assertRaises(ValueError, canonicalSlash, '')

    def testUserPath(self):
        self.assertEqual(userPath('/'), '/')
        self.assertEqual(userPath('/foo'), '/foo')
        self.assertEqual(userPath('/foo/'), '/foo')
        self.assertEqual(userPath('/foo/bar'), '/foo/bar')
        self.assertEqual(userPath('/foo/bar/'), '/foo/bar')

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),
        unittest.makeSuite(TestPathFunctions),
        ))

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


=== Added File Zope3/src/zope/app/hub/tests/test_registration.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_registration.py,v 1.1 2004/03/11 09:19:25 srichter Exp $
"""
from unittest import makeSuite, main, TestCase

from zope.app import zapi
from zope.app.hub import Registration
from zope.app.services.tests.placefulsetup import PlacefulSetup
from zope.app.traversing import traverse, canonicalPath
from zope.app.container.contained import ObjectAddedEvent
from zope.component import getService
from zope.app.services.servicenames import EventPublication, HubIds

class TestRegistration(PlacefulSetup, TestCase):
    def setUp(self):
        PlacefulSetup.setUp(self, site=True)
        self.createStandardServices()
        r = Registration()
        default = traverse(self.rootFolder, '++etc++site/default')
        default['registrar'] = r
        self.registrar = traverse(default, 'registrar')
        self.hub = getService(self.rootFolder, HubIds)
        self.events = getService(self.rootFolder, EventPublication)

    def testSubscribeUnsubscribe(self):
        r = self.registrar
        self.assertEqual(r.isSubscribed(), False)
        r.subscribe()
        self.assertEqual(r.isSubscribed(), True)
        self.assertRaises(RuntimeError, r.subscribe)
        r.unsubscribe()
        self.assertEqual(r.isSubscribed(), False)
        self.assertRaises(RuntimeError, r.unsubscribe)

    def testRegister(self):
        self.registrar.subscribe()
        self.assertEqual(self.hub.numRegistrations(), 0)
        content = zapi.traverse(self.rootFolder, "folder1/folder1_1")

        event = ObjectAddedEvent(content)
        self.events.publish(event)
        self.assertEqual(self.hub.numRegistrations(), 1)

    def testRegisterExisting(self):
        self.registrar.subscribe()
        self.registrar.registerExisting()
        # there are ten folders set up by PlacefulSetup that
        # should get registered
        self.assertEqual(self.hub.numRegistrations(), 10)

def test_suite():
    return makeSuite(TestRegistration)

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




More information about the Zope3-Checkins mailing list