← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_hashcache.py

  • Committer: mbp at sourcefrog
  • Date: 2005-03-29 07:16:19 UTC
  • Revision ID: mbp@sourcefrog.net-20050329071619-a42d323e52dfe5dd
delete obsolete Tree.file_kind()

Show diffs side-by-side

added added

removed removed

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