diff options
Diffstat (limited to 'morphlib/localrepocache_tests.py')
| -rw-r--r-- | morphlib/localrepocache_tests.py | 160 |
1 files changed, 160 insertions, 0 deletions
diff --git a/morphlib/localrepocache_tests.py b/morphlib/localrepocache_tests.py new file mode 100644 index 00000000..22b5ea54 --- /dev/null +++ b/morphlib/localrepocache_tests.py @@ -0,0 +1,160 @@ +# Copyright (C) 2012-2013 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, write to the Free Software Foundation, Inc., +# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. + + +import unittest +import urllib2 +import os + +import cliapp +import fs.memoryfs + +import morphlib + + +class FakeApplication(object): + + def status(self, msg): + pass + + +class LocalRepoCacheTests(unittest.TestCase): + + def setUp(self): + aliases = ['upstream=git://example.com/#example.com:%s.git'] + repo_resolver = morphlib.repoaliasresolver.RepoAliasResolver(aliases) + tarball_base_url = 'http://lorry.example.com/tarballs/' + self.reponame = 'upstream:reponame' + self.repourl = 'git://example.com/reponame' + escaped_url = 'git___example_com_reponame' + self.tarball_url = '%s%s.tar' % (tarball_base_url, escaped_url) + self.cachedir = '/cache/dir' + self.cache_path = '%s/%s' % (self.cachedir, escaped_url) + self.remotes = {} + self.fetched = [] + self.removed = [] + self.lrc = morphlib.localrepocache.LocalRepoCache( + FakeApplication(), self.cachedir, repo_resolver, tarball_base_url) + self.lrc.fs = fs.memoryfs.MemoryFS() + self.lrc._git = self.fake_git + self.lrc._fetch = self.not_found + self.lrc._mkdtemp = self.fake_mkdtemp + self._mkdtemp_count = 0 + + def fake_git(self, args, cwd=None): + if args[0] == 'clone': + self.assertEqual(len(args), 5) + remote = args[3] + local = args[4] + self.remotes['origin'] = {'url': remote, 'updates': 0} + self.lrc.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 fake_mkdtemp(self, dirname): + thing = "foo"+str(self._mkdtemp_count) + self._mkdtemp_count += 1 + self.lrc.fs.makedir(dirname+"/"+thing) + return thing + + def not_found(self, url, path): + raise cliapp.AppException('Not found') + + def test_has_not_got_shortened_repo_initially(self): + self.assertFalse(self.lrc.has_repo(self.reponame)) + + def test_has_not_got_absolute_repo_initially(self): + self.assertFalse(self.lrc.has_repo(self.repourl)) + + def test_caches_shortened_repository_on_request(self): + self.lrc.cache_repo(self.reponame) + self.assertTrue(self.lrc.has_repo(self.reponame)) + self.assertTrue(self.lrc.has_repo(self.repourl)) + + def test_caches_absolute_repository_on_request(self): + self.lrc.cache_repo(self.repourl) + self.assertTrue(self.lrc.has_repo(self.reponame)) + self.assertTrue(self.lrc.has_repo(self.repourl)) + + def test_cachedir_does_not_exist_initially(self): + self.assertFalse(self.lrc.fs.exists(self.cachedir)) + + def test_creates_cachedir_if_missing(self): + self.lrc.cache_repo(self.repourl) + self.assertTrue(self.lrc.fs.exists(self.cachedir)) + + def test_happily_caches_same_repo_twice(self): + self.lrc.cache_repo(self.repourl) + self.lrc.cache_repo(self.repourl) + + def test_fails_to_cache_when_remote_does_not_exist(self): + def fail(args): + self.lrc.fs.makedir(args[4]) + raise cliapp.AppException('') + self.lrc._git = fail + self.assertRaises(morphlib.localrepocache.NoRemote, + self.lrc.cache_repo, self.repourl) + + def test_does_not_mind_a_missing_tarball(self): + self.lrc.cache_repo(self.repourl) + self.assertEqual(self.fetched, []) + + def test_fetches_tarball_when_it_exists(self): + self.lrc._fetch = lambda url, path: self.fetched.append(url) + self.unpacked_tar = "" + self.mkdir_path = "" + self.lrc.cache_repo(self.repourl) + self.assertEqual(self.fetched, [self.tarball_url]) + self.assertFalse(self.lrc.fs.exists(self.cache_path + '.tar')) + self.assertEqual(self.remotes['origin']['url'], self.repourl) + + def test_gets_cached_shortened_repo(self): + self.lrc.cache_repo(self.reponame) + cached = self.lrc.get_repo(self.reponame) + self.assertTrue(cached is not None) + + def test_gets_cached_absolute_repo(self): + self.lrc.cache_repo(self.repourl) + cached = self.lrc.get_repo(self.repourl) + self.assertTrue(cached is not None) + + def test_get_repo_raises_exception_if_repo_is_not_cached(self): + self.assertRaises(Exception, self.lrc.get_repo, self.repourl) + + def test_escapes_repourl_as_filename(self): + escaped = self.lrc._escape(self.repourl) + self.assertFalse('/' in escaped) + + def test_noremote_error_message_contains_repo_name(self): + e = morphlib.localrepocache.NoRemote(self.repourl, []) + self.assertTrue(self.repourl in str(e)) + + def test_avoids_caching_local_repo(self): + self.lrc.fs.makedir('/local/repo', recursive=True) + self.lrc.cache_repo('file:///local/repo') + cached = self.lrc.get_repo('file:///local/repo') + assert cached.path == '/local/repo' |