~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_disk_backed_cache.py

  • Committer: John Arbash Meinel
  • Date: 2006-07-24 20:05:07 UTC
  • mto: This revision was merged to the branch mainline in revision 1885.
  • Revision ID: john@arbash-meinel.com-20060724200507-6a9b467053048ace
Removing disk-backed-cache

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2006 by Canonical Ltd
2
 
#
3
 
# This program is free software; you can redistribute it and/or modify
4
 
# it under the terms of the GNU General Public License as published by
5
 
# the Free Software Foundation; either version 2 of the License, or
6
 
# (at your option) any later version.
7
 
#
8
 
# This program is distributed in the hope that it will be useful,
9
 
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
 
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
11
 
# GNU General Public License for more details.
12
 
#
13
 
# You should have received a copy of the GNU General Public License
14
 
# along with this program; if not, write to the Free Software
15
 
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
 
 
17
 
"""Test that caching works correctly."""
18
 
 
19
 
from bzrlib.disk_backed_cache import DiskBackedCache
20
 
from bzrlib.tests import TestCase
21
 
 
22
 
 
23
 
class TestDiskBackedCache(TestCase):
24
 
    """Tests for the disk-backed cache that don't use the disk."""
25
 
 
26
 
    def get_cache(self, **kwargs):
27
 
        return DiskBackedCache(**kwargs)
28
 
 
29
 
    def test_like_dict(self):
30
 
        """Test that the cache acts like a dict object."""
31
 
        cache = self.get_cache()
32
 
 
33
 
        cache['foo'] = 'bar'
34
 
        self.assertEqual(1, len(cache))
35
 
        self.assertEqual('bar', cache['foo'])
36
 
        self.assertEqual('bar', cache.get('foo'))
37
 
 
38
 
        self.assertRaises(KeyError, cache.__getitem__, 'baz')
39
 
        self.assertEqual(None, cache.get('baz'))
40
 
        self.assertEqual('newval', cache.get('baz', 'newval'))
41
 
 
42
 
        self.failUnless('foo' in cache)
43
 
        self.failUnless(cache.has_key('foo'))
44
 
        self.failIf('baz' in cache)
45
 
        self.failIf(cache.has_key('baz'))
46
 
 
47
 
        cache['tempkey'] = 'xxyyzz'
48
 
        self.assertEqual(2, len(cache))
49
 
        self.failUnless('tempkey' in cache)
50
 
 
51
 
        self.assertEqual(['foo', 'tempkey'], sorted(cache.keys()))
52
 
        self.assertEqual(['foo', 'tempkey'], sorted(cache.iterkeys()))
53
 
 
54
 
        self.assertEqual([('foo', 'bar'), ('tempkey', 'xxyyzz')],
55
 
                         sorted(cache.iteritems()))
56
 
        self.assertEqual(['bar', 'xxyyzz'], sorted(cache.itervalues()))
57
 
 
58
 
        # Make sure values() is a list, not an iterator
59
 
        values = cache.values()
60
 
        self.assertEqual(2, len(values))
61
 
        self.assertEqual(['bar', 'xxyyzz'], sorted(values))
62
 
 
63
 
        del cache['tempkey']
64
 
        self.failIf('tempkey' in cache)
65
 
        self.assertEqual(['foo'], sorted(cache.keys()))
66
 
 
67
 
        key, val = cache.popitem()
68
 
        self.assertEqual('foo', key)
69
 
        self.assertEqual('bar', val)
70
 
        self.assertEqual([], cache.keys())
71
 
 
72
 
        self.assertRaises(KeyError, cache.popitem)
73
 
        self.assertEqual(0, len(cache))
74
 
 
75
 
    def test_only_store_string(self):
76
 
        """DiskBackedCache can only store string objects."""
77
 
        cache = self.get_cache()
78
 
 
79
 
        self.assertRaises(TypeError, cache.__setitem__, 'foo', u'bar')
80
 
        self.assertRaises(TypeError, cache.__setitem__, 'foo', 1)
81
 
        self.assertRaises(TypeError, cache.__setitem__, 'foo', 1.0)
82
 
        self.assertRaises(TypeError, cache.__setitem__, 'foo', object())
83
 
        self.assertRaises(TypeError, cache.__setitem__, 'foo', object)
84
 
 
85
 
    def test_funky_key(self):
86
 
        """We suppport any Dict supported key type"""
87
 
        cache = self.get_cache()
88
 
 
89
 
        cache['str'] = 'str'
90
 
        cache[u'm\xb5'] = 'mu'
91
 
        cache[1] = 'one'
92
 
        cache[('x', 1)] = 'x1'
93
 
 
94
 
        self.failUnless('str' in cache)
95
 
        self.failUnless(u'm\xb5' in cache)
96
 
        self.failUnless(1 in cache)
97
 
        self.failUnless(('x', 1) in cache)
98
 
 
99
 
    def test_tracks_size(self):
100
 
        cache = self.get_cache()
101
 
 
102
 
        self.assertEqual(0, cache.cache_size)
103
 
        cache['foo'] = 'bar'
104
 
        self.assertEqual(3, cache.cache_size)
105
 
        cache['baz'] = 'jiggly'
106
 
        self.assertEqual(9, cache.cache_size)
107
 
 
108
 
        del cache['foo']
109
 
        self.assertEqual(6, cache.cache_size)
110
 
 
111
 
        cache['baz'] = 'alt'
112
 
        self.assertEqual(3, cache.cache_size)
113
 
 
114
 
        cache.clear()
115
 
        self.assertEqual(0, cache.cache_size)
116
 
        self.assertEqual({}, cache._dict)
117
 
 
118
 
    def test_no_disk_stops_caching(self):
119
 
        cache = self.get_cache(max_size=10, use_disk=False)
120
 
 
121
 
        cache['foo'] = 'bar'
122
 
        cache['baz'] = 'six'
123
 
        cache['bar'] = 'toomuch'
124
 
        self.assertEqual(['baz', 'foo'], sorted(cache.keys()))
125
 
 
126
 
    def test_disallow_replace(self):
127
 
        cache = self.get_cache(allow_replace=False)
128
 
        cache['foo'] = 'bar'
129
 
        self.assertRaises(KeyError, cache.__setitem__, 'foo', 'baz')
130
 
 
131
 
    def test_overflow_to_disk(self):
132
 
        cache = self.get_cache(max_size=10)
133
 
 
134
 
        cache['bar'] = '1234567890'
135
 
        self.assertEqual(10, cache.cache_size)
136
 
        self.assertEqual(None, cache._disk_cache)
137
 
        
138
 
        # This should spill to disk
139
 
        cache['baz'] = 'foobar'
140
 
        self.assertNotEqual(None, cache._disk_cache)
141
 
 
142
 
        self.assertEqual('foobar', cache['baz'])
143
 
 
144
 
        cache._disk_cache.seek(0)
145
 
        self.assertEqual('foobar', cache._disk_cache.read())
146
 
 
147
 
    def test_flush_all(self):
148
 
        """All entries are written to disk on overflow if flush_all is set."""
149
 
        cache = self.get_cache(max_size=10, flush_all=True)
150
 
 
151
 
        cache['bar'] = '1234567890'
152
 
        self.assertEqual(10, cache.cache_size)
153
 
        self.assertEqual(None, cache._disk_cache)
154
 
        
155
 
        # This should spill to disk
156
 
        cache['baz'] = 'foobar'
157
 
        self.assertNotEqual(None, cache._disk_cache)
158
 
 
159
 
        # The entries should still be accessible, but
160
 
        # the should all be on disk.
161
 
        self.assertEqual('foobar', cache['baz'])
162
 
        self.assertEqual('1234567890', cache['bar'])
163
 
 
164
 
        # The order on disk doesn't matter, but existing items
165
 
        # should be written before the new item, and since
166
 
        # we only have 1 existing item, the order is fixed
167
 
        cache._disk_cache.seek(0)
168
 
        self.assertEqual('1234567890foobar', cache._disk_cache.read())