summaryrefslogtreecommitdiff
path: root/morphlib/repocache_tests.py
diff options
context:
space:
mode:
Diffstat (limited to 'morphlib/repocache_tests.py')
-rw-r--r--morphlib/repocache_tests.py281
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')
+