[Zodb-checkins] CVS: ZODB3/ZEO/tests - test_cache.py:1.1.2.1

[Jeremy Hylton]

Update of /cvs-repository/ZODB3/ZEO/tests
In directory cvs.zope.org:/tmp/cvs-serv12826/tests

Added Files:
      Tag: ZODB3-mvcc-2-branch
	test_cache.py 
Log Message:
Strawman implementation of new cache API and simple unit tests.


=== Added File ZODB3/ZEO/tests/test_cache.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.
#
##############################################################################
"""Basic unit tests for a multi-version client cache."""

import unittest

import ZEO.cache
from ZODB.utils import p64

n1 = p64(1)
n2 = p64(2)
n3 = p64(3)
n4 = p64(4)
n5 = p64(5)

class CacheTests(unittest.TestCase):

    def setUp(self):
        self.cache = ZEO.cache.Cache()

    def testLastTid(self):
        self.assertEqual(self.cache.getLastTid(), None)
        self.cache.setLastTid(n2)
        self.assertEqual(self.cache.getLastTid(), n2)
        self.cache.invalidate(None, "", n1)
        self.assertEqual(self.cache.getLastTid(), n2)
        self.cache.invalidate(None, "", n3)
        self.assertEqual(self.cache.getLastTid(), n3)
        self.assertRaises(ValueError, self.cache.setLastTid, n2)

    def testLoad(self):
        data1 = "data for n1"
        self.assertEqual(self.cache.load(n1, ""), None)
        self.assertEqual(self.cache.load(n1, "version"), None)
        self.cache.store(n1, "", n2, n3, None, data1)
        self.assertEqual(self.cache.load(n1, ""), (data1, n2))
        self.assertEqual(self.cache.load(n1, "version"), (data1, n2))
        self.assertEqual(self.cache.modifiedInVersion(n1), None)

    def testInvalidate(self):
        data1 = "data for n1"
        self.cache.store(n1, "", n2, n3, None, data1)
        self.cache.invalidate(n1, "", n4)
        self.cache.invalidate(n2, "", n2)
        self.assertEqual(self.cache.load(n1, ""), None)
        self.assertEqual(self.cache.loadNonCurrent(n1, n4),
                         (data1, n2, n3, n4))

    def testVersion(self):
        data1 = "data for n1"
        data1v = "data for n1 in version"
        self.cache.store(n1, "version", n2, n3, None, data1v)
        self.assertEqual(self.cache.load(n1, ""), None)
        self.assertEqual(self.cache.load(n1, "version"), (data1v, n2))
        self.assertEqual(self.cache.modifiedInVersion(n1), "version")
        self.cache.invalidate(n1, "version", n4)
        self.assertEqual(self.cache.load(n1, "version"), None)

    def testNonCurrent(self):
        data1 = "data for n1"
        data2 = "data for n2"
        self.cache.store(n1, "", n4, n4, None, data1)
        self.cache.store(n1, "", n2, n2, n3, data2)
        # can't say anything about state before n2
        self.assertEqual(self.cache.loadNonCurrent(n1, n2), None)
        # n3 is the upper bound of non-current record n2
        self.assertEqual(self.cache.loadNonCurrent(n1, n3),
                         (data2, n2, n2, n3))
        # no data for between n2 and n3
        self.assertEqual(self.cache.loadNonCurrent(n1, n4), None)
        self.cache.invalidate(n1, "", n5)
        self.assertEqual(self.cache.loadNonCurrent(n1, n5),
                         (data1, n4, n4, n5))

    def testException(self):
        self.assertRaises(ValueError,
                          self.cache.store,
                          n1, "version", n2, n2, n3, "data")
        
        
def test_suite():
    return unittest.makeSuite(CacheTests)