~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_hashcache.py

  • Committer: John Arbash Meinel
  • Author(s): Mark Hammond
  • Date: 2008-09-09 17:02:21 UTC
  • mto: This revision was merged to the branch mainline in revision 3697.
  • Revision ID: john@arbash-meinel.com-20080909170221-svim3jw2mrz0amp3
An updated transparent icon for bzr.

Show diffs side-by-side

added added

removed removed

Lines of Context:
 
1
# Copyright (C) 2005, 2006 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
import os
 
18
import sha
 
19
import stat
 
20
import sys
 
21
import time
 
22
 
 
23
from bzrlib.errors import BzrError
 
24
from bzrlib.hashcache import HashCache
 
25
from bzrlib.tests import OsFifoFeature, TestCaseInTempDir, TestCase
 
26
 
 
27
 
 
28
def sha1(t):
 
29
    return sha.new(t).hexdigest()
 
30
 
 
31
 
 
32
def pause():
 
33
    time.sleep(5.0)
 
34
 
 
35
 
 
36
class TestHashCache(TestCaseInTempDir):
 
37
    """Test the hashcache against a real directory"""
 
38
 
 
39
    def make_hashcache(self):
 
40
        # make a dummy bzr directory just to hold the cache
 
41
        os.mkdir('.bzr')
 
42
        hc = HashCache('.', '.bzr/stat-cache')
 
43
        return hc
 
44
 
 
45
    def reopen_hashcache(self):
 
46
        hc = HashCache('.', '.bzr/stat-cache')
 
47
        hc.read()
 
48
        return hc
 
49
 
 
50
    def test_hashcache_initial_miss(self):
 
51
        """Get correct hash from an empty hashcache"""
 
52
        hc = self.make_hashcache()
 
53
        self.build_tree_contents([('foo', 'hello')])
 
54
        self.assertEquals(hc.get_sha1('foo'),
 
55
                          'aaf4c61ddcc5e8a2dabede0f3b482cd9aea9434d')
 
56
        self.assertEquals(hc.miss_count, 1)
 
57
        self.assertEquals(hc.hit_count, 0)
 
58
 
 
59
    def test_hashcache_new_file(self):
 
60
        hc = self.make_hashcache()
 
61
        self.build_tree_contents([('foo', 'goodbye')])
 
62
        # now read without pausing; it may not be possible to cache it as its
 
63
        # so new
 
64
        self.assertEquals(hc.get_sha1('foo'), sha1('goodbye'))
 
65
 
 
66
    def test_hashcache_nonexistent_file(self):
 
67
        hc = self.make_hashcache()
 
68
        self.assertEquals(hc.get_sha1('no-name-yet'), None)
 
69
 
 
70
    def test_hashcache_replaced_file(self):
 
71
        hc = self.make_hashcache()
 
72
        self.build_tree_contents([('foo', 'goodbye')])
 
73
        self.assertEquals(hc.get_sha1('foo'), sha1('goodbye'))
 
74
        os.remove('foo')
 
75
        self.assertEquals(hc.get_sha1('foo'), None)
 
76
        self.build_tree_contents([('foo', 'new content')])
 
77
        self.assertEquals(hc.get_sha1('foo'), sha1('new content'))
 
78
 
 
79
    def test_hashcache_not_file(self):
 
80
        hc = self.make_hashcache()
 
81
        self.build_tree(['subdir/'])
 
82
        self.assertEquals(hc.get_sha1('subdir'), None)
 
83
 
 
84
    def test_hashcache_load(self):
 
85
        hc = self.make_hashcache()
 
86
        self.build_tree_contents([('foo', 'contents')])
 
87
        pause()
 
88
        self.assertEquals(hc.get_sha1('foo'), sha1('contents'))
 
89
        hc.write()
 
90
        hc = self.reopen_hashcache()
 
91
        self.assertEquals(hc.get_sha1('foo'), sha1('contents'))
 
92
        self.assertEquals(hc.hit_count, 1)
 
93
 
 
94
    def test_hammer_hashcache(self):
 
95
        hc = self.make_hashcache()
 
96
        for i in xrange(10000):
 
97
            self.log('start writing at %s', time.time())
 
98
            f = file('foo', 'w')
 
99
            try:
 
100
                last_content = '%08x' % i
 
101
                f.write(last_content)
 
102
            finally:
 
103
                f.close()
 
104
            last_sha1 = sha1(last_content)
 
105
            self.log("iteration %d: %r -> %r",
 
106
                     i, last_content, last_sha1)
 
107
            got_sha1 = hc.get_sha1('foo')
 
108
            self.assertEquals(got_sha1, last_sha1)
 
109
            hc.write()
 
110
            hc = self.reopen_hashcache()
 
111
 
 
112
    def test_hashcache_raise(self):
 
113
        """check that hashcache can raise BzrError"""
 
114
        self.requireFeature(OsFifoFeature)
 
115
        hc = self.make_hashcache()
 
116
        os.mkfifo('a')
 
117
        # It's possible that the system supports fifos but the filesystem
 
118
        # can't.  In that case we should skip at this point.  But in fact
 
119
        # such combinations don't usually occur for the filesystem where
 
120
        # people test bzr.
 
121
        self.assertRaises(BzrError, hc.get_sha1, 'a')
 
122
 
 
123
 
 
124
class FakeHashCache(HashCache):
 
125
    """Hashcache that consults a fake clock rather than the real one.
 
126
 
 
127
    This lets us examine how old or new files would be handled, without
 
128
    actually having to wait for time to pass.
 
129
    """
 
130
    def __init__(self):
 
131
        # set root and cache file name to none to make sure we won't touch the
 
132
        # real filesystem
 
133
        HashCache.__init__(self, '.', 'hashcache')
 
134
        self._files = {}
 
135
        # simulated clock running forward as operations happen
 
136
        self._clock = 0
 
137
 
 
138
    def put_file(self, filename, file_contents):
 
139
        abspath = './' + filename
 
140
        self._files[abspath] = (file_contents, self._clock)
 
141
 
 
142
    def _fingerprint(self, abspath, fs=None):
 
143
        entry = self._files[abspath]
 
144
        return (len(entry[0]),
 
145
                entry[1], entry[1],
 
146
                10, 20,
 
147
                stat.S_IFREG | 0600)
 
148
 
 
149
    def _really_sha1_file(self, abspath):
 
150
        if abspath in self._files:
 
151
            return sha1(self._files[abspath][0])
 
152
        else:
 
153
            return None
 
154
 
 
155
    def _cutoff_time(self):
 
156
        return self._clock - 2
 
157
 
 
158
    def pretend_to_sleep(self, secs):
 
159
        self._clock += secs
 
160
 
 
161
    
 
162
class TestHashCacheFakeFilesystem(TestCaseInTempDir):
 
163
    """Tests the hashcache using a simulated OS.
 
164
    """
 
165
 
 
166
    def make_hashcache(self):
 
167
        return FakeHashCache()
 
168
 
 
169
    def test_hashcache_miss_new_file(self):
 
170
        """A new file gives the right sha1 but misses"""
 
171
        hc = self.make_hashcache()
 
172
        hc.put_file('foo', 'hello')
 
173
        self.assertEquals(hc.get_sha1('foo'), sha1('hello'))
 
174
        self.assertEquals(hc.miss_count, 1)
 
175
        self.assertEquals(hc.hit_count, 0)
 
176
        # if we try again it's still too new; 
 
177
        self.assertEquals(hc.get_sha1('foo'), sha1('hello'))
 
178
        self.assertEquals(hc.miss_count, 2)
 
179
        self.assertEquals(hc.hit_count, 0)
 
180
 
 
181
    def test_hashcache_old_file(self):
 
182
        """An old file gives the right sha1 and hits"""
 
183
        hc = self.make_hashcache()
 
184
        hc.put_file('foo', 'hello')
 
185
        hc.pretend_to_sleep(20)
 
186
        # file is new; should get the correct hash but miss
 
187
        self.assertEquals(hc.get_sha1('foo'), sha1('hello'))
 
188
        self.assertEquals(hc.miss_count, 1)
 
189
        self.assertEquals(hc.hit_count, 0)
 
190
        # and can now be hit
 
191
        self.assertEquals(hc.get_sha1('foo'), sha1('hello'))
 
192
        self.assertEquals(hc.miss_count, 1)
 
193
        self.assertEquals(hc.hit_count, 1)
 
194
        hc.pretend_to_sleep(3)
 
195
        # and again
 
196
        self.assertEquals(hc.get_sha1('foo'), sha1('hello'))
 
197
        self.assertEquals(hc.miss_count, 1)
 
198
        self.assertEquals(hc.hit_count, 2)
 
199
 
 
200
    def test_hashcache_invalidates(self):
 
201
        hc = self.make_hashcache()
 
202
        hc.put_file('foo', 'hello')
 
203
        hc.pretend_to_sleep(20)
 
204
        hc.get_sha1('foo')
 
205
        hc.put_file('foo', 'h1llo')
 
206
        self.assertEquals(hc.get_sha1('foo'), sha1('h1llo'))
 
207
        self.assertEquals(hc.miss_count, 2)
 
208
        self.assertEquals(hc.hit_count, 0)