/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_hashcache.py

  • Committer: Martin
  • Date: 2011-04-15 21:22:52 UTC
  • mto: This revision was merged to the branch mainline in revision 5797.
  • Revision ID: gzlist@googlemail.com-20110415212252-lhqulomwg2y538xj
Add user encoding name to argv decoding error message per poolie in review

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