1
from argparse import Namespace
7
from unittest import TestCase
9
from mock import patch, call
11
from download_juju import (
14
download_candidate_juju,
15
download_released_juju,
22
from tests import parse_error
23
from utility import temp_dir
26
class SelectBuild(TestCase):
28
def test_select_build(self):
30
'j/p/r/build-win-client/build-829/c/juju-setup-1.25-alpha1.exe']
31
build = select_build(builds)
33
build, 's3://juju-qa-data/j/p/r/build-win-client/build-829')
35
def test_select_builds(self):
37
'j/p/r/build-win-client/build-829/c/juju-setup-1.25-alpha1.exe',
38
'j/p/r/build-win-client/build-1000/c/juju-setup-1.25-alpha1.exe',
39
'j/p/v/build-win-client/build-2000/c/juju-setup-1.25-alpha1.exe',
40
'j/p/r/build-win-client/build-999/c/juju-setup-1.25-alpha1.exe',
42
build = select_build(builds)
44
build, 's3://juju-qa-data/j/p/v/build-win-client/build-2000')
47
class TestGetMD5(TestCase):
49
def test_get_md5_empty(self):
51
filename = os.path.join(d, "afile")
52
open(filename, "w").close()
53
expected_md5 = hashlib.md5("").hexdigest()
54
self.assertEqual(expected_md5, get_md5(filename))
56
def test_get_md5_size(self):
58
filename = os.path.join(d, "afile")
59
contents = "some string longer\nthan the size passed\n"
60
with open(filename, "w") as f:
62
expected_md5 = hashlib.md5(contents).hexdigest()
63
self.assertEqual(expected_md5, get_md5(filename, size=4))
66
class TestMkdir(TestCase):
68
def test_mkdir_p(self):
70
path = os.path.join(d, 'a/b/c')
72
self.assertTrue(os.path.isdir(path))
75
class TestParseArgs(TestCase):
77
def test_parse_args(self):
78
with parse_error(self) as stderr:
80
self.assertRegexpMatches(stderr.getvalue(), 'error: too few arguments')
82
parse_args(['/cred/path']),
83
Namespace(credential_path='/cred/path', platform=sys.platform,
84
released=False, revision=None, verbose=0))
86
parse_args(['/cred/path', '-r']),
87
Namespace(credential_path='/cred/path', platform=sys.platform,
88
released=True, revision=None, verbose=0))
90
parse_args(['/cred/path', '-r', '-c', '2999', '3001']),
91
Namespace(credential_path='/cred/path', platform=sys.platform,
92
released=True, revision=['2999', '3001'], verbose=0))
95
class TestDownloadFiles(TestCase):
97
def test_download(self):
99
key = KeyStub("test", "hello")
100
with temp_dir() as d:
101
dst = os.path.join(d, "test.txt")
103
self.assertEqual(content, get_file_content(dst))
105
def test_download_retries(self):
107
raise socket.error(errno.ECONNRESET, "reset")
108
key = KeyStub("test", "foo")
109
with temp_dir() as d:
110
dst = os.path.join(d, "test.txt")
111
with patch.object(key, 'get_contents_to_filename', autospec=True,
112
side_effect=se) as gcf:
113
with self.assertRaises(socket.error):
115
self.assertEqual(gcf.call_count, 3)
117
def test_download_files(self):
118
keys = [KeyStub("test.txt", "foo"), KeyStub("test2.txt", "foo2")]
119
with temp_dir() as dst:
120
downloaded_files = download_files(keys, dst)
121
self.assertItemsEqual(os.listdir(dst), ['test.txt', 'test2.txt'])
122
self.assertEqual("foo", get_file_content('test.txt', dst))
123
self.assertEqual("foo2", get_file_content('test2.txt', dst))
124
self.assertItemsEqual(downloaded_files, ['test.txt', 'test2.txt'])
126
def test_download_files__matching_file_already_exists(self):
127
keys = [KeyStub("test.txt", "foo"), KeyStub("test2.txt", "foo2")]
128
with temp_dir() as dst:
129
set_file_content("test.txt", "foo", dst)
130
downloaded_files = download_files(keys, dst)
131
self.assertItemsEqual(os.listdir(dst), ['test.txt', 'test2.txt'])
132
self.assertEqual("foo", get_file_content('test.txt', dst))
133
self.assertEqual("foo2", get_file_content('test2.txt', dst))
134
self.assertItemsEqual(downloaded_files, ['test2.txt'])
136
def test_download_files__md5_mismatch(self):
137
keys = [KeyStub("test.txt", "foo"), KeyStub("test2.txt", "foo2")]
138
with temp_dir() as dst:
139
set_file_content("test.txt", "not foo", dst)
140
downloaded_files = download_files(keys, dst)
141
self.assertItemsEqual(os.listdir(dst), ['test.txt', 'test2.txt'])
142
self.assertEqual("foo", get_file_content('test.txt', dst))
143
self.assertEqual("foo2", get_file_content('test2.txt', dst))
144
self.assertItemsEqual(downloaded_files, ['test.txt', 'test2.txt'])
146
def test_download_files__overwrite(self):
147
keys = [KeyStub("test.txt", "foo"), KeyStub("test2.txt", "foo2")]
148
with temp_dir() as dst:
149
set_file_content("test.txt", "foo", dst)
150
downloaded_files = download_files(keys, dst, overwrite=True)
151
self.assertItemsEqual(os.listdir(dst), ['test.txt', 'test2.txt'])
152
self.assertEqual("foo", get_file_content('test.txt', dst))
153
self.assertEqual("foo2", get_file_content('test2.txt', dst))
154
self.assertItemsEqual(downloaded_files, ['test.txt', 'test2.txt'])
156
def test_download_files__suffix(self):
157
keys = [KeyStub("test.txt", "foo"), KeyStub("test2.tar", "foo2")]
158
with temp_dir() as dst:
159
downloaded_files = download_files(keys, dst, suffix=".tar")
160
self.assertItemsEqual(os.listdir(dst), ['test2.tar'])
161
self.assertEqual("foo2", get_file_content('test2.tar', dst))
162
self.assertItemsEqual(downloaded_files, ['test2.tar'])
164
def test_download_files__dst_dir_none(self):
165
keys = [KeyStub("test.txt", "foo"), KeyStub("test2.txt", "foo2")]
166
with patch('download_juju._download', autospec=True) as dj:
168
self.assertFalse(dj.called)
170
def test_s3_download_files(self):
171
with temp_dir() as dst:
172
with patch('download_juju.s3_auth_with_rc', autospec=True) as ds:
174
'download_juju.download_files', autospec=True) as dd:
175
s3_download_files('s3://foo/path', '/cred/path', dst)
176
ds.assert_called_once_with('/cred/path')
177
dd.assert_called_once_with(
178
ds.return_value.get_bucket.return_value.list.return_value,
181
def test_download_released_juju(self):
182
with patch('download_juju.s3_download_files', autospec=True) as dd:
183
download_released_juju(
184
Namespace(platform='win32', credential_path='/tmp'))
185
s3path = 's3://juju-qa-data/client-archive/win/'
186
dst = os.path.join(os.environ['HOME'], 'old-juju', 'win')
187
dd.assert_called_once_with(s3path, credential_path='/tmp', dst_dir=dst)
189
def test_download_candidate_juju(self):
190
with patch('download_juju.s3_download_files', autospec=True) as ds:
191
with patch('download_juju.select_build', autospec=True) as db:
192
download_candidate_juju(
193
Namespace(platform='darwin', credential_path='/tmp',
194
revision=['11', '33']))
195
dst = os.path.join(os.environ['HOME'], 'candidate', 'osx')
196
self.assertEqual(ds.call_count, 4)
197
exp = 's3://juju-qa-data/juju-ci/products/version-{}/build-{}-client/'
199
call(exp.format('11', 'osx'), credential_path='/tmp',
201
call(db.return_value, credential_path='/tmp', suffix='.tar.gz',
203
call(exp.format('33', 'osx'), credential_path='/tmp',
205
call(db.return_value, credential_path='/tmp', suffix='.tar.gz',
208
self.assertEqual(ds.call_args_list, calls)
209
self.assertEqual(db.call_count, 2)
214
def __init__(self, name, content=""):
216
self.etag = '"{}"'.format(hashlib.md5(content).hexdigest())
217
self.content = content
219
def get_contents_to_filename(self, filename):
220
with open(filename, "wb") as f:
221
f.write(self.content)
224
def set_file_content(filename, content, dst_dir=None):
225
dst_path = os.path.join(dst_dir, filename) if dst_dir else filename
226
with open(dst_path, "wb") as f:
230
def get_file_content(filename, dst_dir=None):
231
dst_path = os.path.join(dst_dir, filename) if dst_dir else filename
232
with open(dst_path) as f:
added
removed
tests/test_download_juju.py
