# Copyright (C) 2005-2011, 2016 Canonical Ltd # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation; either version 2 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA import os import stat import time from .. import osutils from ..errors import BzrError from ..hashcache import HashCache from . import ( TestCaseInTempDir, ) from .features import ( OsFifoFeature, ) sha1 = osutils.sha_string def pause(): time.sleep(5.0) class TestHashCache(TestCaseInTempDir): """Test the hashcache against a real directory""" def make_hashcache(self): # make a dummy bzr directory just to hold the cache os.mkdir('.bzr') hc = HashCache(u'.', '.bzr/stat-cache') return hc def reopen_hashcache(self): hc = HashCache(u'.', '.bzr/stat-cache') hc.read() return hc def test_hashcache_initial_miss(self): """Get correct hash from an empty hashcache""" hc = self.make_hashcache() self.build_tree_contents([('foo', b'hello')]) self.assertEqual(hc.get_sha1('foo'), b'aaf4c61ddcc5e8a2dabede0f3b482cd9aea9434d') self.assertEqual(hc.miss_count, 1) self.assertEqual(hc.hit_count, 0) def test_hashcache_new_file(self): hc = self.make_hashcache() self.build_tree_contents([('foo', b'goodbye')]) # now read without pausing; it may not be possible to cache it as its # so new self.assertEqual(hc.get_sha1('foo'), sha1(b'goodbye')) def test_hashcache_nonexistent_file(self): hc = self.make_hashcache() self.assertEqual(hc.get_sha1('no-name-yet'), None) def test_hashcache_replaced_file(self): hc = self.make_hashcache() self.build_tree_contents([('foo', b'goodbye')]) self.assertEqual(hc.get_sha1('foo'), sha1(b'goodbye')) os.remove('foo') self.assertEqual(hc.get_sha1('foo'), None) self.build_tree_contents([('foo', b'new content')]) self.assertEqual(hc.get_sha1('foo'), sha1(b'new content')) def test_hashcache_not_file(self): hc = self.make_hashcache() self.build_tree(['subdir/']) self.assertEqual(hc.get_sha1('subdir'), None) def test_hashcache_load(self): hc = self.make_hashcache() self.build_tree_contents([('foo', b'contents')]) pause() self.assertEqual(hc.get_sha1('foo'), sha1(b'contents')) hc.write() hc = self.reopen_hashcache() self.assertEqual(hc.get_sha1('foo'), sha1(b'contents')) self.assertEqual(hc.hit_count, 1) def test_hammer_hashcache(self): hc = self.make_hashcache() for i in range(10000): with open('foo', 'wb') as f: last_content = b'%08x' % i f.write(last_content) last_sha1 = sha1(last_content) self.log("iteration %d: %r -> %r", i, last_content, last_sha1) got_sha1 = hc.get_sha1('foo') self.assertEqual(got_sha1, last_sha1) hc.write() hc = self.reopen_hashcache() def test_hashcache_raise(self): """check that hashcache can raise BzrError""" self.requireFeature(OsFifoFeature) hc = self.make_hashcache() os.mkfifo('a') # It's possible that the system supports fifos but the filesystem # can't. In that case we should skip at this point. But in fact # such combinations don't usually occur for the filesystem where # people test bzr. self.assertRaises(BzrError, hc.get_sha1, 'a') class FakeHashCache(HashCache): """Hashcache that consults a fake clock rather than the real one. This lets us examine how old or new files would be handled, without actually having to wait for time to pass. """ def __init__(self): # set root and cache file name to none to make sure we won't touch the # real filesystem HashCache.__init__(self, u'.', 'hashcache') self._files = {} # simulated clock running forward as operations happen self._clock = 0 def put_file(self, filename, file_contents): abspath = './' + filename self._files[abspath] = (file_contents, self._clock) def _fingerprint(self, abspath, fs=None): entry = self._files[abspath] return (len(entry[0]), entry[1], entry[1], 10, 20, stat.S_IFREG | 0o600) def _really_sha1_file(self, abspath, filters): if abspath in self._files: return sha1(self._files[abspath][0]) else: return None def _cutoff_time(self): return self._clock - 2 def pretend_to_sleep(self, secs): self._clock += secs class TestHashCacheFakeFilesystem(TestCaseInTempDir): """Tests the hashcache using a simulated OS. """ def make_hashcache(self): return FakeHashCache() def test_hashcache_miss_new_file(self): """A new file gives the right sha1 but misses""" hc = self.make_hashcache() hc.put_file('foo', b'hello') self.assertEqual(hc.get_sha1('foo'), sha1(b'hello')) self.assertEqual(hc.miss_count, 1) self.assertEqual(hc.hit_count, 0) # if we try again it's still too new; self.assertEqual(hc.get_sha1('foo'), sha1(b'hello')) self.assertEqual(hc.miss_count, 2) self.assertEqual(hc.hit_count, 0) def test_hashcache_old_file(self): """An old file gives the right sha1 and hits""" hc = self.make_hashcache() hc.put_file('foo', b'hello') hc.pretend_to_sleep(20) # file is new; should get the correct hash but miss self.assertEqual(hc.get_sha1('foo'), sha1(b'hello')) self.assertEqual(hc.miss_count, 1) self.assertEqual(hc.hit_count, 0) # and can now be hit self.assertEqual(hc.get_sha1('foo'), sha1(b'hello')) self.assertEqual(hc.miss_count, 1) self.assertEqual(hc.hit_count, 1) hc.pretend_to_sleep(3) # and again self.assertEqual(hc.get_sha1('foo'), sha1(b'hello')) self.assertEqual(hc.miss_count, 1) self.assertEqual(hc.hit_count, 2) def test_hashcache_invalidates(self): hc = self.make_hashcache() hc.put_file('foo', b'hello') hc.pretend_to_sleep(20) hc.get_sha1('foo') hc.put_file('foo', b'h1llo') self.assertEqual(hc.get_sha1('foo'), sha1(b'h1llo')) self.assertEqual(hc.miss_count, 2) self.assertEqual(hc.hit_count, 0)