[Zodb-checkins] CVS: ZODB3/ZEO/tests - testClientCache.py:1.1
Guido van Rossum
guido@python.org
Tue, 27 Aug 2002 14:19:38 -0400
Update of /cvs-repository/ZODB3/ZEO/tests
In directory cvs.zope.org:/tmp/cvs-serv25829
Added Files:
testClientCache.py
Log Message:
The beginnings of a test suite for ClientCache.py.
This will cause three failures with the current version of
ClientCache.py, because some of the tests were specifically written to
verify certain bugs. I will check in fixes for thse bugs shortly.
=== Added File ZODB3/ZEO/tests/testClientCache.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 suite for the ZEO.ClientCache module.
At times, we do 'white box' testing, i.e. we know about the internals
of the ClientCache object."""
import os
import time
import tempfile
import unittest
from ZEO.ClientCache import ClientCache
class ClientCacheTests(unittest.TestCase):
def setUp(self):
unittest.TestCase.setUp(self)
self.cache = ClientCache()
self.cache.open()
def tearDown(self):
self.cache.close()
unittest.TestCase.tearDown(self)
def testOpenClose(self):
pass # All the work is done by setUp() / tearDown()
def testStoreLoad(self):
cache = self.cache
oid = 'abcdefgh'
data = '1234'*100
serial = 'ABCDEFGH'
cache.store(oid, data, serial, '', '', '')
loaded = cache.load(oid, '')
self.assertEqual(loaded, (data, serial))
def testMissingLoad(self):
cache = self.cache
oid = 'abcdefgh'
data = '1234'*100
serial = 'ABCDEFGH'
cache.store(oid, data, serial, '', '', '')
loaded = cache.load('garbage1', '')
self.assertEqual(loaded, None)
def testInvalidate(self):
cache = self.cache
oid = 'abcdefgh'
data = '1234'*100
serial = 'ABCDEFGH'
cache.store(oid, data, serial, '', '', '')
loaded = cache.load(oid, '')
self.assertEqual(loaded, (data, serial))
cache.invalidate(oid, '')
loaded = cache.load(oid, '')
self.assertEqual(loaded, None)
def testVersion(self):
cache = self.cache
oid = 'abcdefgh'
data = '1234'*100
serial = 'ABCDEFGH'
vname = 'myversion'
vdata = '5678'*200
vserial = 'IJKLMNOP'
cache.store(oid, data, serial, vname, vdata, vserial)
loaded = cache.load(oid, '')
self.assertEqual(loaded, (data, serial))
vloaded = cache.load(oid, vname)
self.assertEqual(vloaded, (vdata, vserial))
def testVersionOnly(self):
cache = self.cache
oid = 'abcdefgh'
data = ''
serial = ''
vname = 'myversion'
vdata = '5678'*200
vserial = 'IJKLMNOP'
cache.store(oid, data, serial, vname, vdata, vserial)
loaded = cache.load(oid, '')
self.assertEqual(loaded, None)
vloaded = cache.load(oid, vname)
self.assertEqual(vloaded, (vdata, vserial))
def testInvalidateNonVersion(self):
cache = self.cache
oid = 'abcdefgh'
data = '1234'*100
serial = 'ABCDEFGH'
vname = 'myversion'
vdata = '5678'*200
vserial = 'IJKLMNOP'
cache.store(oid, data, serial, vname, vdata, vserial)
loaded = cache.load(oid, '')
self.assertEqual(loaded, (data, serial))
vloaded = cache.load(oid, vname)
self.assertEqual(vloaded, (vdata, vserial))
cache.invalidate(oid, '')
loaded = cache.load(oid, '')
self.assertEqual(loaded, None)
# The version data is also invalidated at this point
vloaded = cache.load(oid, vname)
self.assertEqual(vloaded, None)
def testInvalidateVersion(self):
# Invalidating a version should not invalidate the non-version data.
# (This tests for the same bug as testInvalidatePersists below.)
cache = self.cache
oid = 'abcdefgh'
data = '1234'*100
serial = 'ABCDEFGH'
cache.store(oid, data, serial, '', '', '')
loaded = cache.load(oid, '')
self.assertEqual(loaded, (data, serial))
cache.invalidate(oid, 'bogus')
loaded = cache.load(oid, '')
self.assertEqual(loaded, (data, serial))
def testVerify(self):
cache = self.cache
def verifier(oid, serial, vserial):
results.append((oid, serial, vserial))
results = []
cache.verify(verifier)
self.assertEqual(results, [])
oid = 'abcdefgh'
data = '1234'*100
serial = 'ABCDEFGH'
cache.store(oid, data, serial, '', '', '')
results = []
cache.verify(verifier)
self.assertEqual(results, [(oid, serial, None)])
class PersistentClientCacheTests(unittest.TestCase):
def setUp(self):
unittest.TestCase.setUp(self)
self.vardir = os.getcwd() # Don't use /tmp, it's a security risk
self.cachesize = 10*1000*1000
self.storagename = 'foo'
self.clientname = 'test'
# Predict file names
fn0 = 'c%s-%s-0.zec' % (self.storagename, self.clientname)
fn1 = 'c%s-%s-1.zec' % (self.storagename, self.clientname)
for fn in fn0, fn1:
fn = os.path.join(self.vardir, fn)
try:
os.unlink(fn)
except os.error:
pass
self.openCache()
def openCache(self):
self.cache = ClientCache(storage=self.storagename,
size=self.cachesize,
client=self.clientname,
var=self.vardir)
self.cache.open()
def reopenCache(self):
self.cache.close()
self.openCache()
return self.cache
def tearDown(self):
self.cache.close()
for filename in self.cache._p:
if filename is not None:
try:
os.unlink(filename)
except os.error:
pass
unittest.TestCase.tearDown(self)
def testCacheFileSelection(self):
# A bug in __init__ read the wrong slice of the file to determine
# the serial number of the first record, reading the
# last byte of the data size plus the first seven bytes of the
# serial number. This caused random selection of the proper
# 'current' file when a persistent cache was opened.
cache = self.cache
self.assertEqual(cache._current, 0) # Check that file 0 is current
oid = 'abcdefgh'
data = '1234'
serial = 'ABCDEFGH'
cache.store(oid, data, serial, '', '', '')
cache.checkSize(10*self.cachesize) # Force a file flip
self.assertEqual(cache._current, 1) # Check that the flip worked
oid = 'abcdefgh'
data = '123'
serial = 'ABCDEFGZ'
cache.store(oid, data, serial, '', '', '')
cache = self.reopenCache()
loaded = cache.load(oid, '')
# Check that we got the most recent data:
self.assertEqual(loaded, (data, serial))
self.assertEqual(cache._current, 1) # Double check that 1 is current
def testInvalidationPersists(self):
# A bug in invalidate() caused invalidation to overwrite the
# 2nd byte of the data size on disk, rather rather than
# overwriting the status byte. For certain data sizes this
# can be observed by reopening a persistent cache: the
# invalidated data will appear valid (but with altered size).
cache = self.cache
magicsize = (ord('i') + 1) << 16
cache = self.cache
oid = 'abcdefgh'
data = '!'*magicsize
serial = 'ABCDEFGH'
cache.store(oid, data, serial, '', '', '')
loaded = cache.load(oid, '')
self.assertEqual(loaded, (data, serial))
cache.invalidate(oid, '')
cache = self.reopenCache()
loaded = cache.load(oid, '')
if loaded != None:
self.fail("invalidated data resurrected, size %d, was %d" %
(len(loaded[0]), len(data)))
def test_suite():
suite = unittest.TestSuite()
suite.addTest(unittest.makeSuite(ClientCacheTests))
suite.addTest(unittest.makeSuite(PersistentClientCacheTests))
return suite
if __name__ == '__main__':
unittest.main(defaultTest='test_suite')