diff options
Diffstat (limited to 'morphlib/repocache_tests.py')
-rw-r--r-- | morphlib/repocache_tests.py | 281 |
1 files changed, 281 insertions, 0 deletions
diff --git a/morphlib/repocache_tests.py b/morphlib/repocache_tests.py new file mode 100644 index 00000000..6e07aedb --- /dev/null +++ b/morphlib/repocache_tests.py @@ -0,0 +1,281 @@ +# Copyright (C) 2012-2016 Codethink Limited +# +# 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; version 2 of the License. +# +# 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, see <http://www.gnu.org/licenses/>. + + +import unittest +import urllib2 +import json +import os + +import cliapp +import fs.memoryfs +import tempfile + +import morphlib +import morphlib.gitdir_tests + + +class TestableRepoCache(morphlib.repocache.RepoCache): + '''Adapts the RepoCache class for unit testing. + + All Git operations are stubbed out. You can track what Git operations have + taken place by looking at the 'remotes' dict -- any 'clone' operations will + set an entry in there. The 'tarballs_fetched' list tracks what tarballs + of Git repos would have been downloaded. + + There is a single repo alias, 'example' which expands to + git://example.com/. + + ''' + def __init__(self, update_gits=True): + aliases = ['example=git://example.com/#example.com:%s.git'] + repo_resolver = morphlib.repoaliasresolver.RepoAliasResolver(aliases) + tarball_base_url = 'http://lorry.example.com/tarballs' + cachedir = '/cache/gits/' + memoryfs = fs.memoryfs.MemoryFS() + + morphlib.repocache.RepoCache.__init__( + self, cachedir, repo_resolver, tarball_base_url=tarball_base_url, + custom_fs=memoryfs, update_gits=update_gits) + + self.remotes = {} + self.tarballs_fetched = [] + + self._mkdtemp_count = 0 + + def _mkdtemp(self, dirname): + thing = "foo"+str(self._mkdtemp_count) + self._mkdtemp_count += 1 + self.fs.makedir(dirname+"/"+thing) + return thing + + def _fetch(self, url, path): + self.tarballs_fetched.append(url) + + def _git(self, args, **kwargs): + if args[0] == 'clone': + assert len(args) == 5 + remote = args[3] + local = args[4] + self.remotes['origin'] = {'url': remote, 'updates': 0} + self.fs.makedir(local, recursive=True) + elif args[0:2] == ['remote', 'set-url']: + remote = args[2] + url = args[3] + self.remotes[remote] = {'url': url} + elif args[0:2] == ['config', 'remote.origin.url']: + remote = 'origin' + url = args[2] + self.remotes[remote] = {'url': url} + elif args[0:2] == ['config', 'remote.origin.mirror']: + remote = 'origin' + elif args[0:2] == ['config', 'remote.origin.fetch']: + remote = 'origin' + else: + raise NotImplementedError() + + def _update_repo(self, cached_repo): + pass + + +class RepoCacheTests(unittest.TestCase): + + def test_has_not_got_repo_initially(self): + repo_cache = TestableRepoCache() + self.assertFalse(repo_cache.has_repo('example:repo')) + self.assertFalse(repo_cache.has_repo('git://example.com/repo')) + + def test_happily_caches_same_repo_twice(self): + repo_cache = TestableRepoCache() + with morphlib.gitdir_tests.allow_nonexistant_git_repos(): + repo_cache.get_updated_repo('example:repo', ref='master') + repo_cache.get_updated_repo('example:repo', ref='master') + + def test_fails_to_cache_when_remote_does_not_exist(self): + repo_cache = TestableRepoCache() + + def clone_fails(args, **kwargs): + repo_cache.fs.makedir(args[4]) + raise cliapp.AppException('') + repo_cache._git = clone_fails + + with self.assertRaises(morphlib.repocache.NoRemote): + repo_cache.get_updated_repo('example:repo', 'master') + + def test_does_not_mind_a_missing_tarball(self): + repo_cache = TestableRepoCache() + + def no_tarball(*args, **kwargs): + raise cliapp.AppException('Not found') + repo_cache._fetch = no_tarball + + with morphlib.gitdir_tests.allow_nonexistant_git_repos(): + repo_cache.get_updated_repo('example:repo', ref='master') + self.assertEqual(repo_cache.tarballs_fetched, []) + + def test_fetches_tarball_when_it_exists(self): + repo_url = 'git://example.com/reponame' + repo_cache = TestableRepoCache() + + with morphlib.gitdir_tests.allow_nonexistant_git_repos(): + repo_cache.get_updated_repo(repo_url, ref='master') + + tarball_url = '%s%s.tar' % (repo_cache._tarball_base_url, + repo_cache._escape(repo_url)) + self.assertEqual(repo_cache.tarballs_fetched, [tarball_url]) + + # Check that the cache updated the repo after fetching the tarball. + self.assertEqual(repo_cache.remotes['origin']['url'], repo_url) + + def test_escapes_repourl_as_filename(self): + repo_cache = TestableRepoCache() + escaped = repo_cache._escape('git://example.com/reponame') + self.assertFalse('/' in escaped) + + def test_noremote_error_message_contains_repo_name(self): + repo_url = 'git://example.com/reponame' + e = morphlib.repocache.NoRemote(repo_url, []) + self.assertTrue(repo_url in str(e)) + + def test_avoids_caching_local_repo(self): + repo_cache = TestableRepoCache() + + repo_cache.fs.makedir('/local/repo', recursive=True) + with morphlib.gitdir_tests.allow_nonexistant_git_repos(): + cached = repo_cache.get_updated_repo( + 'file:///local/repo', refs='master') + assert cached.dirname == '/local/repo' + + def test_no_git_update_setting(self): + repo_cache = TestableRepoCache(update_gits=False) + + with self.assertRaises(morphlib.repocache.NotCached): + repo_cache.get_updated_repo('example:repo', ref='master') + + +class RemoteRepoCacheTests(unittest.TestCase): + def _resolve_ref_for_repo_url(self, repo_url, ref): + return self.sha1s[repo_url][ref] + + def _cat_file_for_repo_url(self, repo_url, sha1, filename): + try: + return self.files[repo_url][sha1][filename] + except KeyError: + raise urllib2.HTTPError(url='', code=404, msg='Not found', + hdrs={}, fp=None) + + def _ls_tree_for_repo_url(self, repo_url, sha1): + return json.dumps({ + 'repo': repo_url, + 'ref': sha1, + 'tree': self.files[repo_url][sha1] + }) + + def setUp(self): + self.sha1s = { + 'git://gitorious.org/baserock/morph': { + 'master': 'e28a23812eadf2fce6583b8819b9c5dbd36b9fb9' + } + } + self.files = { + 'git://gitorious.org/baserock-morphs/linux': { + 'e28a23812eadf2fce6583b8819b9c5dbd36b9fb9': { + 'linux.morph': 'linux morphology' + } + } + } + self.server_url = 'http://foo.bar' + aliases = [ + 'upstream=git://gitorious.org/baserock-morphs/#foo', + 'baserock=git://gitorious.org/baserock/#foo' + ] + resolver = morphlib.repoaliasresolver.RepoAliasResolver(aliases) + self.cache = morphlib.repocache.RemoteRepoCache( + self.server_url, resolver) + self.cache._resolve_ref_for_repo_url = self._resolve_ref_for_repo_url + self.cache._cat_file_for_repo_url = self._cat_file_for_repo_url + self.cache._ls_tree_for_repo_url = self._ls_tree_for_repo_url + + def test_sets_server_url(self): + self.assertEqual(self.cache.server_url, self.server_url) + + def test_resolve_existing_ref_for_existing_repo(self): + sha1 = self.cache.resolve_ref('baserock:morph', 'master') + self.assertEqual( + sha1, + self.sha1s['git://gitorious.org/baserock/morph']['master']) + + def test_fail_resolving_existing_ref_for_non_existent_repo(self): + self.assertRaises(morphlib.repocache.RemoteResolveRefError, + self.cache.resolve_ref, 'non-existent-repo', + 'master') + + def test_fail_resolving_non_existent_ref_for_existing_repo(self): + self.assertRaises(morphlib.repocache.RemoteResolveRefError, + self.cache.resolve_ref, 'baserock:morph', + 'non-existent-ref') + + def test_fail_resolving_non_existent_ref_for_non_existent_repo(self): + self.assertRaises(morphlib.repocache.RemoteResolveRefError, + self.cache.resolve_ref, 'non-existent-repo', + 'non-existent-ref') + + def test_cat_existing_file_in_existing_repo_and_ref(self): + content = self.cache.cat_file( + 'upstream:linux', 'e28a23812eadf2fce6583b8819b9c5dbd36b9fb9', + 'linux.morph') + self.assertEqual(content, 'linux morphology') + + def test_fail_cat_file_using_invalid_sha1(self): + self.assertRaises(morphlib.repocache.RemoteCatFileError, + self.cache.cat_file, 'upstream:linux', 'blablabla', + 'linux.morph') + + def test_fail_cat_non_existent_file_in_existing_repo_and_ref(self): + self.assertRaises(morphlib.repocache.RemoteCatFileError, + self.cache.cat_file, 'upstream:linux', + 'e28a23812eadf2fce6583b8819b9c5dbd36b9fb9', + 'non-existent-file') + + def test_fail_cat_file_in_non_existent_ref_in_existing_repo(self): + self.assertRaises(morphlib.repocache.RemoteCatFileError, + self.cache.cat_file, 'upstream:linux', + 'ecd7a325095a0d19b8c3d76f578d85b979461d41', + 'linux.morph') + + def test_fail_cat_file_in_non_existent_repo(self): + self.assertRaises(morphlib.repocache.RemoteCatFileError, + self.cache.cat_file, 'non-existent-repo', + 'e28a23812eadf2fce6583b8819b9c5dbd36b9fb9', + 'some-file') + + def test_ls_tree_in_existing_repo_and_ref(self): + content = self.cache.ls_tree( + 'upstream:linux', 'e28a23812eadf2fce6583b8819b9c5dbd36b9fb9') + self.assertEqual(content, ['linux.morph']) + + def test_fail_ls_tree_using_invalid_sha1(self): + self.assertRaises(morphlib.repocache.RemoteLsTreeError, + self.cache.ls_tree, 'upstream:linux', 'blablabla') + + def test_fail_ls_file_in_non_existent_ref_in_existing_repo(self): + self.assertRaises(morphlib.repocache.RemoteLsTreeError, + self.cache.ls_tree, 'upstream:linux', + 'ecd7a325095a0d19b8c3d76f578d85b979461d41') + + def test_fail_ls_tree_in_non_existent_repo(self): + self.assertRaises(morphlib.repocache.RemoteLsTreeError, + self.cache.ls_tree, 'non-existent-repo', + 'e28a23812eadf2fce6583b8819b9c5dbd36b9fb9') + |