[Zope3-checkins] CVS: Zope3/src/zope/app/index/text/tests - __init__.py:1.1.2.1 test_batchedrankedprocessor.py:1.1.2.1 test_index.py:1.1.2.1

[Jim Fulton]

Update of /cvs-repository/Zope3/src/zope/app/index/text/tests
In directory cvs.zope.org:/tmp/cvs-serv19908/zope/app/index/text/tests

Added Files:
      Tag: NameGeddon-branch
	__init__.py test_batchedrankedprocessor.py test_index.py 
Log Message:
Initial renaming before debugging

=== Added File Zope3/src/zope/app/index/text/tests/__init__.py ===
#
# This file is necessary to make this directory a package.


=== Added File Zope3/src/zope/app/index/text/tests/test_batchedrankedprocessor.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_batchedrankedprocessor.py,v 1.1.2.1 2002/12/23 19:31:40 jim Exp $
"""

from unittest import TestCase, TestSuite, main, makeSuite
from zope.app.tests.placelesssetup import PlacelessSetup

from zope.interface.verify import verifyObject

from zope.component import getAdapter

from zope.textindex.textindexinterfaces import IQuerying
from zope.app.interfaces.index.interfaces import IBatchedQuery, ITextIndexQuery, \
    IBatchedResult, IRankedHubIdList
    
from zope.app.index.text.processors import \
    BatchedRankedProcessor, IBatchedRankedProcessor
from zope.app.index.queries import BatchedRankedResult
from zope.app.index.text.queries import BatchedTextIndexQuery

class StupidTextIndex:

    __implements__ = IQuerying

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

    def query(self, querytext, start, count):
        """a stub query processor"""
        self.args = (querytext, start, count)

        return self.returnvalue

#############################################################################
# If your tests change any global registries, then uncomment the
# following import and include CleanUp as a base class of your
# test. It provides a setUp and tearDown that clear global data that
# has registered with the test cleanup framework.  Don't use this
# tests outside the Zope package.

# from zope.testing.cleanup import CleanUp # Base class w registry cleanup

#############################################################################

class Test(PlacelessSetup, TestCase):

    def _Test__new(self):
        return BatchedRankedProcessor()

    ############################################################
    # Interface-driven tests:

    def test_IVerify(self):
        processor = BatchedRankedProcessor(StupidTextIndex(None))
        verifyObject(IBatchedRankedProcessor, processor)

    def test_parameter(self):
        query = BatchedTextIndexQuery(u"test AND foo OR bar", 0, 20)
        index = StupidTextIndex(([], 0))

        processor = BatchedRankedProcessor(index)
        result = processor(query)

        # Do introspection for parameterpassing on the index
        self.failUnlessEqual(index.args, (u"test AND foo OR bar", 0, 20))
        
        # Do introspection on the result
        
        # BatchedResult
        batch = getAdapter(result, IBatchedResult)
        self.failUnlessEqual(0, batch.totalSize)
        self.failUnlessEqual(0, batch.startPosition)
        self.failUnlessEqual(20, batch.batchSize)

        # RankedHubIdList
        list = getAdapter(result, IRankedHubIdList)
        self.failUnlessRaises(IndexError, list.__getitem__, 0)

def test_suite():
    return makeSuite(Test)

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


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

$Id: test_index.py,v 1.1.2.1 2002/12/23 19:31:40 jim Exp $
"""

from unittest import TestCase, TestSuite, main, makeSuite

from zope.component.adapter import provideAdapter
from zope.app.event.objectevent import ObjectModifiedEvent

from zope.app.services.tests.placefulsetup import \
     PlacefulSetup

from zope.app.interfaces.traversing.traverser import ITraverser
from Zope.App.Traversing import locationAsUnicode

from zope.app.interfaces.services.hub import \
     IRegistrationHubEvent, IObjectModifiedHubEvent
from zope.app.services.hub import \
     ObjectRegisteredHubEvent, \
     ObjectUnregisteredHubEvent, \
     ObjectModifiedHubEvent
from zope.app.services.hub import ObjectHub

from zope.app.interfaces.index.text.interfaces import ISearchableText
from zope.app.index.text.index import TextIndex

class FakeSearchableObject:
    __implements__ = ISearchableText
    def __init__(self):
        self.texts = [u"Bruce"]
    def getSearchableText(self):
        return self.texts

class FakeTraverser:
    __implements__ = ITraverser
    def __init__(self, object, location):
        self.__object = object
        self.__location = location
    def traverse(self, path):
        canonical_path = locationAsUnicode(path)
        if canonical_path == self.__location:
            return self.__object
        raise KeyError, (path, canonical_path)

Bruce = u"Bruce"
Sheila = u"Sheila"

class Test(PlacefulSetup, TestCase):

    def setUp(self):
        PlacefulSetup.setUp(self)
        self.index = TextIndex()
        self.object = FakeSearchableObject()

    def assertPresent(self, word, docid):
        results, total = self.index.query(word)
        self.assertEqual(total, 1)
        self.assertEqual(results[0][0], docid)

    def assertAbsent(self, word):
        self.assertEqual(self.index.query(word), ([], 0))

    def testNotification(self):
        docid = 1000
        event = ObjectRegisteredHubEvent(None, docid, object=self.object)
        self.index.notify(event)
        self.assertPresent(Bruce, docid)

        self.object.texts = [Sheila]
        event = ObjectModifiedHubEvent(None, docid, object=self.object)
        self.index.notify(event)
        self.assertPresent(Sheila, docid)
        self.assertAbsent(Bruce)

        event = ObjectUnregisteredHubEvent(None, docid,
                                           location="fake",
                                           object=self.object)
        self.index.notify(event)
        self.assertAbsent(Bruce)
        self.assertAbsent(Sheila)

    def testNotIndexing(self):
        docid = 1000
        self.object.texts = None
        event = ObjectRegisteredHubEvent(None, docid, object=self.object)
        self.index.notify(event)
        self.assertEqual(self.index.documentCount(), 0)
        
    

    def testHubMachinery(self):
        # Technically this is a functional test
        hub = ObjectHub()
        hub.subscribe(self.index, IRegistrationHubEvent)
        hub.subscribe(self.index, IObjectModifiedHubEvent)
        location = "/bruce"
        traverser = FakeTraverser(self.object, location)
        provideAdapter(None, ITraverser, lambda dummy: traverser)

        hubid = hub.register(location)
        self.assertPresent(Bruce, hubid)

        self.object.texts = [Sheila]
        event = ObjectModifiedEvent(self.object, location)
        hub.notify(event)
        self.assertPresent(Sheila, hubid)
        self.assertAbsent(Bruce)

        hub.unregister(location)
        self.assertAbsent(Bruce)
        self.assertAbsent(Sheila)

    def testBootstrap(self):
        self.assertEqual(self.index.isSubscribed(), False)
        self.assertAbsent(Bruce)
        self.assertAbsent(Sheila)

        hub = ObjectHub()
        location = "/bruce"
        traverser = FakeTraverser(self.object, location)
        provideAdapter(None, ITraverser, lambda dummy: traverser)
        hubid = hub.register(location)
        self.index.subscribe(hub)
        self.assertEqual(self.index.isSubscribed(), True)
        self.assertPresent(Bruce, hubid)

        self.index.unsubscribe(hub)
        self.assertEqual(self.index.isSubscribed(), False)
        self.assertPresent(Bruce, hubid)

        self.object.texts = [Sheila]
        event = ObjectModifiedEvent(self.object, location)
        hub.notify(event)
        self.assertPresent(Bruce, hubid)
        self.assertAbsent(Sheila)




def test_suite():
    return makeSuite(Test)

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