[Zope3-checkins] CVS: Zope3/src/zope/app/catalog/tests - test_catalog.py:1.1.2.1

Anthony Baxter anthony@interlink.com.au
Sat, 12 Jul 2003 22:55:14 -0400


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

Added Files:
      Tag: melb-2003-content-catalog-branch
	test_catalog.py 
Log Message:
basic test suites.


=== Added File Zope3/src/zope/app/catalog/tests/test_catalog.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.
#
##############################################################################
"""Tests for catalog

Note that indexes &c already have test suites, we only have to check that
a catalog passes on events that it receives.
$Id: test_catalog.py,v 1.1.2.1 2003/07/13 02:55:09 anthony Exp $
"""

import unittest

from zope.interface import Interface, implements
from zope.app.interfaces.index.field import IUIFieldCatalogIndex
from zope.app.interfaces.catalog.index import ICatalogIndex
from zope.app.interfaces.event import ISubscriber
from zope.app.interfaces.services.hub import IObjectHub
from zodb.btrees.IIBTree import IISet

from zope.app.catalog.catalog import Catalog
from zope.app.tests.placelesssetup import PlacelessSetup
from zope.component import getServiceManager
from zope.app.services.servicenames import HubIds

from zope.app.index.tests.test_objectretrievingprocessor import FakeObjectHub

import zope.app.services.hub as Hub

regEvt = Hub.ObjectRegisteredHubEvent
unregEvt = Hub.ObjectUnregisteredHubEvent
modEvt = Hub.ObjectModifiedHubEvent

class CFakeObjectHub(FakeObjectHub):
    def iterObjectRegistrations(self):
        def gen(things):
            for hubid, obj in things:
                loc = "/%s"%hubid
                yield loc,hubid,obj
        return gen(self.data.items())

class StubIndex(object):
    implements(ISubscriber, ICatalogIndex, IUIFieldCatalogIndex)

    def __init__(self, field_name, interface=None):
	self._field_name = field_name
	self.interface = interface
	self._notifies = []

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

    def clear(self):
	self._notifies = []

    def _getterms(self):
        d = {}
        for e in self._notifies:
            ob = e.object
            term = getattr(e.object ,self._field_name, '')
            d.setdefault(term, []).append(e.hubid)
        return d

    def search(self, term):
	termdict = self._getterms()
        res = termdict.get(term, [])
        return IISet(res)

class stoopid(object):
    def __init__(self, **kw):
        self.__dict__ = kw

class Test(PlacelessSetup, unittest.TestCase):

    def setUp(self):
	pass

    def teardown(self):
	pass

    def test_catalog_add_del_indexes(self):
	catalog = Catalog()
	index = StubIndex('author', None)
	catalog.setObject('author', index)
	self.assertEqual(catalog.keys(), ['author'])
	index = StubIndex('title', None)
	catalog.setObject('title', index)
	indexes = catalog.keys()
	indexes.sort()
	self.assertEqual(indexes, ['author', 'title'])
	del catalog['author']
	self.assertEqual(catalog.keys(), ['title'])

    def test_catalog_notification_passing(self):
	catalog = Catalog()
	catalog.setObject('author', StubIndex('author', None))
	catalog.setObject('title', StubIndex('title', None))
        catalog.notify(regEvt(None, None, 'reg1', 1))
        catalog.notify(regEvt(None, None, 'reg2', 2))
        catalog.notify(regEvt(None, None, 'reg3', 3))
        catalog.notify(unregEvt(None, None, 'unreg4', 4))
        catalog.notify(unregEvt(None, None, 'unreg5', 5))
        catalog.notify(modEvt(None, None, 'mod6', 6))
        for index in catalog.values():
            checkNotifies = index._notifies
            self.assertEqual(len(checkNotifies), 6)
            notifLocs = [ x.location for x in checkNotifies ]
            self.assertEqual(notifLocs, ['reg1', 'reg2', 'reg3', 
                                         'unreg4', 'unreg5','mod6' ])
            self.assertEqual(notifLocs, ['reg1', 'reg2', 'reg3', 
                                         'unreg4', 'unreg5','mod6' ])
        catalog.clearIndexes()
        for index in catalog.values():
            checkNotifies = index._notifies
            self.assertEqual(len(checkNotifies), 0)

    def _frob_objecthub(self, ints=1, apes=1):
        hub = CFakeObjectHub()
        service_manager = getServiceManager(None)
        service_manager.defineService(HubIds, IObjectHub)
        service_manager.provideService(HubIds, hub)
        # whack some objects in our little objecthub
        if ints:
            for i in range(10):
                hub.register("<object %s>"%i)
        if apes:
            hub.register(stoopid(simiantype='monkey', name='bobo'))
            hub.register(stoopid(simiantype='monkey', name='bubbles'))
            hub.register(stoopid(simiantype='monkey', name='ginger'))
            hub.register(stoopid(simiantype='bonobo', name='ziczac'))
            hub.register(stoopid(simiantype='bonobo', name='bobo'))
            hub.register(stoopid(simiantype='punyhuman', name='anthony'))
            hub.register(stoopid(simiantype='punyhuman', name='andy'))
            hub.register(stoopid(simiantype='punyhuman', name='kev'))

    def test_updateindexes(self):
        "test a full refresh"
        self._frob_objecthub()
        catalog = Catalog()
	catalog.setObject('author', StubIndex('author', None))
	catalog.setObject('title', StubIndex('author', None))
        catalog.updateIndexes()
        for index in catalog.values():
            checkNotifies = index._notifies
            self.assertEqual(len(checkNotifies), 18)
            notifLocs = [ x.location for x in checkNotifies ]
            notifLocs.sort()
            expected = [ "/%s"%(i+1) for i in range(18) ]
            expected.sort()
            self.assertEqual(notifLocs, expected)

    def test_basicsearch(self):
        "test the simple searchresults interface"
        self._frob_objecthub(ints=0)
        catalog = Catalog()
	catalog.setObject('simiantype', StubIndex('simiantype', None))
	catalog.setObject('name', StubIndex('name', None))
        catalog.updateIndexes()
        res = catalog.searchResults(simiantype='monkey')
        names = [ x.name for x in res ]
        names.sort()
        self.assertEqual(len(names), 3)
        self.assertEqual(names, ['bobo', 'bubbles', 'ginger'])
        res = catalog.searchResults(name='bobo')
        names = [ x.simiantype for x in res ]
        names.sort()
        self.assertEqual(len(names), 2)
        self.assertEqual(names, ['bonobo', 'monkey'])
        res = catalog.searchResults(simiantype='punyhuman', name='anthony')
        res = list(res)
        self.assertEqual(len(res), 1)
        ob = res[0]
        self.assertEqual((ob.name,ob.simiantype), ('anthony', 'punyhuman'))
        res = catalog.searchResults(simiantype='ape', name='bobo')
        res = list(res)
        self.assertEqual(len(res), 0)
        res = catalog.searchResults(simiantype='ape', name='mwumi')
        res = list(res)
        self.assertEqual(len(res), 0)
        self.assertRaises(ValueError, catalog.searchResults, 
                            simiantype='monkey', hat='beret')
        res = list(res)

def test_suite():
    return unittest.makeSuite(Test)


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