1
# Copyright (C) 2005, 2006 Canonical Ltd
 
 
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.
 
 
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.
 
 
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
 
 
23
from bzrlib.errors import BzrError
 
 
24
from bzrlib.hashcache import HashCache
 
 
25
from bzrlib.tests import TestCaseInTempDir, TestSkipped, TestCase
 
 
29
    return sha.new(t).hexdigest()
 
 
36
class TestHashCache(TestCaseInTempDir):
 
 
37
    """Test the hashcache against a real directory"""
 
 
39
    def make_hashcache(self):
 
 
40
        # make a dummy bzr directory just to hold the cache
 
 
42
        hc = HashCache('.', '.bzr/stat-cache')
 
 
45
    def reopen_hashcache(self):
 
 
46
        hc = HashCache('.', '.bzr/stat-cache')
 
 
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)
 
 
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
 
 
64
        self.assertEquals(hc.get_sha1('foo'), sha1('goodbye'))
 
 
66
    def test_hashcache_nonexistent_file(self):
 
 
67
        hc = self.make_hashcache()
 
 
68
        self.assertEquals(hc.get_sha1('no-name-yet'), None)
 
 
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'))
 
 
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'))
 
 
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)
 
 
84
    def test_hashcache_load(self):
 
 
85
        hc = self.make_hashcache()
 
 
86
        self.build_tree_contents([('foo', 'contents')])
 
 
88
        self.assertEquals(hc.get_sha1('foo'), sha1('contents'))
 
 
90
        hc = self.reopen_hashcache()
 
 
91
        self.assertEquals(hc.get_sha1('foo'), sha1('contents'))
 
 
92
        self.assertEquals(hc.hit_count, 1)
 
 
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())
 
 
100
                last_content = '%08x' % i
 
 
101
                f.write(last_content)
 
 
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)
 
 
110
            hc = self.reopen_hashcache()
 
 
112
    def test_hashcache_raise(self):
 
 
113
        """check that hashcache can raise BzrError"""
 
 
114
        hc = self.make_hashcache()
 
 
115
        if not hasattr(os, 'mkfifo'):
 
 
116
            raise TestSkipped('filesystem fifos not supported on this system')
 
 
118
        # It's possible that the system supports fifos but the filesystem
 
 
119
        # can't.  In that case we should skip at this point.  But in fact
 
 
120
        # such combinations don't usually occur for the filesystem where
 
 
122
        self.assertRaises(BzrError, hc.get_sha1, 'a')
 
 
125
class FakeHashCache(HashCache):
 
 
126
    """Hashcache that consults a fake clock rather than the real one.
 
 
128
    This lets us examine how old or new files would be handled, without
 
 
129
    actually having to wait for time to pass.
 
 
132
        # set root and cache file name to none to make sure we won't touch the
 
 
134
        HashCache.__init__(self, '.', 'hashcache')
 
 
136
        # simulated clock running forward as operations happen
 
 
139
    def put_file(self, filename, file_contents):
 
 
140
        abspath = './' + filename
 
 
141
        self._files[abspath] = (file_contents, self._clock)
 
 
143
    def _fingerprint(self, abspath):
 
 
144
        entry = self._files[abspath]
 
 
145
        return (len(entry[0]),
 
 
150
    def _really_sha1_file(self, abspath):
 
 
151
        if abspath in self._files:
 
 
152
            return sha1(self._files[abspath][0])
 
 
156
    def _cutoff_time(self):
 
 
157
        return self._clock - 2
 
 
159
    def pretend_to_sleep(self, secs):
 
 
163
class TestHashCacheFakeFilesystem(TestCaseInTempDir):
 
 
164
    """Tests the hashcache using a simulated OS.
 
 
167
    def make_hashcache(self):
 
 
168
        return FakeHashCache()
 
 
170
    def test_hashcache_miss_new_file(self):
 
 
171
        """A new file gives the right sha1 but misses"""
 
 
172
        hc = self.make_hashcache()
 
 
173
        hc.put_file('foo', 'hello')
 
 
174
        self.assertEquals(hc.get_sha1('foo'), sha1('hello'))
 
 
175
        self.assertEquals(hc.miss_count, 1)
 
 
176
        self.assertEquals(hc.hit_count, 0)
 
 
177
        # if we try again it's still too new; 
 
 
178
        self.assertEquals(hc.get_sha1('foo'), sha1('hello'))
 
 
179
        self.assertEquals(hc.miss_count, 2)
 
 
180
        self.assertEquals(hc.hit_count, 0)
 
 
182
    def test_hashcache_old_file(self):
 
 
183
        """An old file gives the right sha1 and hits"""
 
 
184
        hc = self.make_hashcache()
 
 
185
        hc.put_file('foo', 'hello')
 
 
186
        hc.pretend_to_sleep(20)
 
 
187
        # file is new; should get the correct hash but miss
 
 
188
        self.assertEquals(hc.get_sha1('foo'), sha1('hello'))
 
 
189
        self.assertEquals(hc.miss_count, 1)
 
 
190
        self.assertEquals(hc.hit_count, 0)
 
 
192
        self.assertEquals(hc.get_sha1('foo'), sha1('hello'))
 
 
193
        self.assertEquals(hc.miss_count, 1)
 
 
194
        self.assertEquals(hc.hit_count, 1)
 
 
195
        hc.pretend_to_sleep(3)
 
 
197
        self.assertEquals(hc.get_sha1('foo'), sha1('hello'))
 
 
198
        self.assertEquals(hc.miss_count, 1)
 
 
199
        self.assertEquals(hc.hit_count, 2)
 
 
201
    def test_hashcache_invalidates(self):
 
 
202
        hc = self.make_hashcache()
 
 
203
        hc.put_file('foo', 'hello')
 
 
204
        hc.pretend_to_sleep(20)
 
 
206
        hc.put_file('foo', 'h1llo')
 
 
207
        self.assertEquals(hc.get_sha1('foo'), sha1('h1llo'))
 
 
208
        self.assertEquals(hc.miss_count, 2)
 
 
209
        self.assertEquals(hc.hit_count, 0)