1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
|
# 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 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')
|
