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
163
164
165
166
167
168
169
170
|
# Copyright (C) 2012 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 morphlib
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.cache = set()
self.remotes = {}
self.fetched = []
self.removed = []
self.lrc = morphlib.localrepocache.LocalRepoCache(
object(), self.cachedir, repo_resolver, tarball_base_url)
self.lrc._git = self.fake_git
self.lrc._exists = self.fake_exists
self.lrc._fetch = self.not_found
self.lrc._mkdir = self.fake_mkdir
self.lrc._remove = self.fake_remove
self.lrc._runcmd = self.fake_runcmd
def fake_runcmd(self, args, cwd=None):
if args[0:2] == ['tar', 'xf']:
self.unpacked_tar = args[2]
self.cache.add(cwd)
else:
raise NotImplementedError()
def fake_git(self, args, cwd=None):
if args[0] == 'clone':
self.assertEqual(len(args), 5)
remote = args[3]
local = args[4]
if local in self.cache:
raise Exception('cloning twice to %s' % local)
self.remotes['origin'] = {'url': remote, 'updates': 0}
self.cache.add(local)
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_exists(self, filename):
return filename in self.cache
def fake_mkdir(self, dirname):
self.cache.add(dirname)
def fake_remove(self, filename):
self.removed.append(filename)
def not_found(self, url, path):
raise urllib2.URLError('Not found')
def fake_fetch(self, url, path):
self.fetched.append(url)
self.cache.add(path)
return True
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.cachedir in self.cache)
def test_creates_cachedir_if_missing(self):
self.lrc.cache_repo(self.repourl)
self.assertTrue(self.cachedir in self.cache)
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):
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 = self.fake_fetch
self.unpacked_tar = ""
self.mkdir_path = ""
self.lrc.cache_repo(self.repourl)
self.assertEqual(self.fetched, [self.tarball_url])
self.assertEqual(self.removed, [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.cache_repo('file:///local/repo')
cached = self.lrc.get_repo('file:///local/repo')
assert cached.path == '/local/repo'
|
