[Zope3-checkins] CVS: Zope3/src/zope/app/cache/tests - __init__.py:1.1.2.1 test_annotationcacheable.py:1.1.2.1 test_cachename.py:1.1.2.1 test_caching.py:1.1.2.1 test_icache.py:1.1.2.1 test_ramcache.py:1.1.2.1

Jim Fulton jim@zope.com
Mon, 23 Dec 2002 14:31:26 -0500


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

Added Files:
      Tag: NameGeddon-branch
	__init__.py test_annotationcacheable.py test_cachename.py 
	test_caching.py test_icache.py test_ramcache.py 
Log Message:
Initial renaming before debugging

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


=== Added File Zope3/src/zope/app/cache/tests/test_annotationcacheable.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 AnnotationCacheable adapter.

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

from unittest import TestCase, TestSuite, main, makeSuite
from zope.app.tests.placelesssetup import PlacelessSetup
from zope.component import getService
from zope.app.interfaces.annotation import IAnnotations
from zope.app.interfaces.annotation import IAttributeAnnotatable
from zope.app.attributeannotations import AttributeAnnotations
from zope.app.cache.annotationcacheable import AnnotationCacheable
from zope.app.interfaces.cache.cache import ICachingService
from zope.component.service import \
     serviceManager as sm

class ObjectStub:
    __implements__ = IAttributeAnnotatable


class CacheStub:
    def __init__(self):
        self.invalidated = []

    def invalidate(self, obj):
        self.invalidated.append(obj)


class CachingServiceStub:
    __implements__ = ICachingService

    def __init__(self):
        self.caches = {}

    def getCache(self, name):
        return self.caches[name]


class TestAnnotationCacheable(PlacelessSetup, TestCase):
    def setUp(self):
        PlacelessSetup.setUp(self)
        getService(None, "Adapters").provideAdapter(
            IAttributeAnnotatable, IAnnotations,
            AttributeAnnotations)
        self.service = CachingServiceStub()
        sm.defineService('Caching', ICachingService)
        sm.provideService('Caching', self.service)

    def testNormal(self):
        ob = ObjectStub()
        adapter = AnnotationCacheable(ob)
        self.assertEquals(adapter.getCacheId(), None,
                          "initially cache ID should be None")

        adapter.setCacheId("my_id")
        self.assertEquals(adapter.getCacheId(), "my_id",
                          "failed to set cache ID")

    def testInvalidate(self):
        # Test that setting a different cache ID invalidates the old cached
        # value
        self.service.caches['cache1'] = cache1 = CacheStub()
        self.service.caches['cache2'] = cache2 = CacheStub()
        ob = ObjectStub()
        adapter = AnnotationCacheable(ob)
        adapter.setCacheId('cache1')
        self.assertEquals(cache1.invalidated, [],
                          "called invalidate too early")
        adapter.setCacheId('cache2')
        self.assertEquals(cache1.invalidated, [ob], "did not call invalidate")
        adapter.setCacheId('cache2')
        self.assertEquals(
            cache2.invalidated, [],
            "called invalidate when reassigning to the same cache")


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


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


=== Added File Zope3/src/zope/app/cache/tests/test_cachename.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 CacheName field

In particular, test the proper getting of cache names in allowed_values.

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

from unittest import TestCase, TestSuite, main, makeSuite
from zope.proxy.context.context import ContextWrapper

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

from zope.app.interfaces.cache.cache import CacheName

class CachingServiceStub(object):

    def getAvailableCaches(self):
        return 'foo', 'bar', 'baz'

    
class Test(PlacefulSetup, TestCase):

    def setUp(self):
        PlacefulSetup.setUp(self)
        self.buildFolders()
        sm = TestingServiceManager()
        self.rootFolder.setServiceManager(sm)
        sm.Caching = CachingServiceStub()        

    def test(self):
        field = CacheName().bind(self.rootFolder)
        allowed = list(field.allowed_values)
        allowed.sort()
        self.assertEqual(allowed, ['', 'bar', 'baz', 'foo'])

    
def test_suite():
    return TestSuite((
        makeSuite(Test),
        ))

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


=== Added File Zope3/src/zope/app/cache/tests/test_caching.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 caching helpers.

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

from unittest import TestCase, TestSuite, main, makeSuite
from zope.app.interfaces.cache.cache import ICacheable
from zope.app.interfaces.cache.cache import ICachingService
from zope.app.cache.caching import getCacheForObj
from zope.app.cache.annotationcacheable import AnnotationCacheable
from zope.app.interfaces.annotation import IAnnotatable
from zope.app.interfaces.annotation import IAnnotations
from zope.app.interfaces.annotation import IAttributeAnnotatable
from zope.app.attributeannotations import AttributeAnnotations
from zope.component import getAdapter
from zope.component import getService
from zope.app.tests.placelesssetup import PlacelessSetup
from zope.component.service import \
     serviceManager as sm

class ObjectStub:
    __implements__ = IAttributeAnnotatable

class CacheStub:
    pass

class CachingServiceStub:

    __implements__ = ICachingService

    def __init__(self):
        self.caches = {}

    def getCache(self, name):
        return self.caches[name]

class Test(PlacelessSetup, TestCase):

    def setUp(self):
        PlacelessSetup.setUp(self)
        getService(None, "Adapters").provideAdapter(
            IAttributeAnnotatable, IAnnotations,
            AttributeAnnotations)
        getService(None, "Adapters").provideAdapter(
            IAnnotatable, ICacheable,
            AnnotationCacheable)
        self.service = CachingServiceStub()
        sm.defineService('Caching', ICachingService)
        sm.provideService('Caching', self.service)

    def testGetCacheForObj(self):
        self.service.caches['my_cache'] = my_cache = CacheStub()

        obj = ObjectStub()
        self.assertEquals(getCacheForObj(obj), None)

        getAdapter(obj, ICacheable).setCacheId("my_cache")

        self.assertEquals(getCacheForObj(obj), my_cache)


def test_suite():
    return TestSuite((
        makeSuite(Test),
        ))

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


=== Added File Zope3/src/zope/app/cache/tests/test_icache.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 ICache interface

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

from unittest import TestSuite, main
from zope.interface.verify import verifyObject
from zope.app.interfaces.cache.cache import ICache


class BaseICacheTest:
    """Base class for ICache unit tests.  Subclasses should provide a
    _Test__new() method that returns a new empty cache object.
    """

    def testVerifyICache(self):
        # Verify that the object implements ICache
        verifyObject(ICache, self._Test__new())

    def testCaching(self):
        # Verify basic caching
        cache = self._Test__new()
        ob = "obj"
        data = "data"
        marker = []
        self.failIf(cache.query(ob, None, default=marker) is not marker,
                    "empty cache should not contain anything")

        cache.set(data, ob, key={'id': 35})
        self.assertEquals(cache.query(ob, {'id': 35}), data,
                    "should return cached result")
        self.failIf(cache.query(ob, {'id': 33}, default=marker) is not marker,
                    "should not return cached result for a different key")

        cache.invalidate(ob, {"id": 33})
        self.assertEquals(cache.query(ob, {'id': 35}), data,
                          "should return cached result")
        self.failIf(cache.query(ob, {'id': 33}, default=marker) is not marker,
                    "should not return cached result after invalidate")

    def testInvalidateAll(self):
        cache = self._Test__new()
        ob1 = object()
        ob2 = object()
        cache.set("data1", ob1)
        cache.set("data2", ob2, key={'foo': 1})
        cache.set("data3", ob2, key={'foo': 2})
        cache.invalidateAll()
        marker = []
        self.failIf(cache.query(ob1, default=marker) is not marker,
                    "should not return cached result after invalidateAll")
        self.failIf(cache.query(ob2, {'foo': 1}, default=marker) is not marker,
                    "should not return cached result after invalidateAll")
        self.failIf(cache.query(ob2, {'foo': 2}, default=marker) is not marker,
                    "should not return cached result after invalidateAll")


def test_suite():
    return TestSuite((
        ))

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






=== Added File Zope3/src/zope/app/cache/tests/test_ramcache.py === (477/577 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.
#
##############################################################################
"""Unit tests for RAM Cache.

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

from unittest import TestCase, TestSuite, main, makeSuite
from zope.app.cache.tests.test_icache import BaseICacheTest
from zope.app.tests.placelesssetup import PlacelessSetup
from zope.app.interfaces.traversing.physicallylocatable import IPhysicallyLocatable
from zope.proxy.context import Wrapper
from zope.component.adapter import provideAdapter
from zope.interface.verify import verifyClass, verifyObject
from time import time


class Locatable:
    __implements__ = IPhysicallyLocatable

    def __init__(self, path=('a', 'b')):
        self.path = path

    def getPhysicalRoot(self):
        return self

    def getPhysicalPath(self):
        return self.path

class TestRAMCache(PlacelessSetup,
                   TestCase,
                   BaseICacheTest,
                   ):

    def _Test__new(self):
        from zope.app.cache.ram import RAMCache
        return RAMCache()

    def test_interface(self):

[-=- -=- -=- 477 lines omitted -=- -=- -=-]

                   object2: {key1: [value, 4, 2],
                             key2: [value, 5, 1],
                             key3: [value, 6, 1]}}
        s._misses = {object: 11, object2: 42}
        len1 = len(dumps(s._data[object]))
        len2 = len(dumps(s._data[object2]))

        expected = ({'path': object,
                     'hits': 17,
                     'misses': 11,
                     'size': len1,
                     'entries': 3
                     },
                    {'path': object2,
                     'hits': 4,
                     'misses': 42,
                     'size': len2,
                     'entries': 3
                     },
                    )

        result = s.getStatistics()

        self.assertEqual(result, expected, "got unexpected stats")


class TestModule(TestCase):

    def test_locking(self):
        from zope.app.cache.ram import writelock
        writelock.acquire()
        try:
            self.failUnless(writelock.locked(), "locks don't work")
        finally:
            writelock.release()

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

class Test(TestCase):
    pass

def test_suite():
    return TestSuite((
        makeSuite(TestRAMCache),
        makeSuite(TestStorage),
        makeSuite(TestModule),
        ))

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