70
63
args = parse_args(['get-juju-bin', '3275', '-vv'])
71
64
self.assertEqual(2, args.verbose)
73
def test_get_defaults(self):
74
default_config = os.path.join(get_juju_home(), 'juju-qa.s3cfg')
75
args = parse_args(['get', '3275', 'job-foo', 'files-bar'])
76
self.assertEqual(Namespace(
77
command='get', config=default_config, revision_build=3275,
78
job='job-foo', file_pattern='files-bar', workspace='.', verbose=0),
81
def test_get_workspace(self):
82
args = parse_args(['get', '3275', 'job-foo', 'files-bar',
84
self.assertEqual('myworkspace', args.workspace)
86
def test_get_too_few(self):
87
with parse_error(self) as stderr:
88
parse_args(['get', '3275', 'job-foo'])
89
self.assertRegexpMatches(stderr.getvalue(), 'too few arguments$')
91
def test_get_verbosity(self):
92
args = parse_args(['get', '3275', 'job-foo', 'files-bar', '-v'])
93
self.assertEqual(1, args.verbose)
94
args = parse_args(['get', '3275', 'job-foo', 'files-bar', '-vv'])
95
self.assertEqual(2, args.verbose)
98
67
class TestGetS3Credentials(TestCase):
139
108
self.assertEqual(exc.exception.errno, errno.ENOENT)
142
def mock_key(revision_build, job, build, file_path):
111
def mock_package_key(revision_build, build=27, distro_release=None):
143
112
key = create_autospec(S3Key, instance=True)
144
key.name = '{}/build-{}/{}'.format(
145
get_job_path(revision_build, job), build, file_path)
149
def mock_package_key(revision_build, build=27, distro_release=None):
150
113
namer = PackageNamer.factory()
151
114
if distro_release is not None:
152
115
namer.distro_release = distro_release
153
116
package = namer.get_release_package('109.6')
154
job = JobNamer.factory().get_build_binary_job()
155
return mock_key(revision_build, job, build, package)
117
key.name = '{}/build-{}/{}'.format(
118
get_job_path(revision_build), build, package)
158
122
def mock_bucket(keys):
159
123
bucket = create_autospec(Bucket, instance=True)
161
def list_bucket(prefix):
162
return [key for key in keys if key.name.startswith(prefix)]
164
bucket.list.side_effect = list_bucket
124
bucket.list.return_value = keys
180
140
def test_find_package_key(self):
181
141
key = mock_package_key(390)
182
142
bucket = mock_bucket([key])
183
namer = JobNamer.factory()
184
job = namer.get_build_binary_job()
185
143
found_key, filename = find_package_key(bucket, 390)
186
bucket.list.assert_called_once_with(get_job_path(390, job))
144
bucket.list.assert_called_once_with(get_job_path(390))
187
145
self.assertIs(key, found_key)
188
146
self.assertEqual(filename, get_key_filename(key))
190
148
def test_selects_latest(self):
191
149
new_key = mock_package_key(390, build=27)
192
150
old_key = mock_package_key(390, build=9)
193
bucket = FakeBucket([old_key, new_key, old_key])
151
bucket = mock_bucket([old_key, new_key, old_key])
194
152
found_key = find_package_key(bucket, 390)[0]
195
153
self.assertIs(new_key, found_key)
197
155
def test_wrong_version(self):
198
156
key = mock_package_key(390, distro_release='01.01')
199
bucket = FakeBucket([key])
157
bucket = mock_bucket([key])
200
158
with self.assertRaises(PackageNotFound):
201
159
find_package_key(bucket, 390)
203
161
def test_wrong_file(self):
204
162
key = mock_package_key(390)
205
163
key.name = key.name.replace('juju-core', 'juju-dore')
206
bucket = FakeBucket([key])
164
bucket = mock_bucket([key])
207
165
with self.assertRaises(PackageNotFound):
208
166
find_package_key(bucket, 390)
236
194
self.assertEqual(os.path.join(eb_dir, 'juju'), extracted)
239
class TestFetchFiles(StrictTestCase):
242
use_context(self, patch('utility.get_deb_arch', return_value='amd65',
245
def test_fetch_files(self):
246
key = mock_key(275, 'job-foo', 27, 'file-pattern')
247
bucket = FakeBucket([key])
248
with temp_dir() as workspace:
249
downloaded = fetch_files(bucket, 275, 'job-foo', 'file-pat+ern',
251
local_file = os.path.join(workspace, 'file-pattern')
252
key.get_contents_to_filename.assert_called_once_with(local_file)
253
key_copy = os.path.join(workspace, local_file)
254
self.assertEqual([key_copy], downloaded)
259
def __init__(self, revision_build, job, build, file_path):
260
job_path = get_job_path(revision_build, job)
261
self.name = '{}/build-{}/{}'.format(job_path, build, file_path)
266
def __init__(self, keys):
269
def list(self, prefix):
270
return [key for key in self.keys if key.name.startswith(prefix)]
273
class TestFindFileKeys(StrictTestCase):
275
def test_find_file_keys(self):
276
key = FakeKey(275, 'job-foo', 27, 'file-pattern')
277
bucket = FakeBucket([key])
278
filtered = find_file_keys(bucket, 275, 'job-foo', 'file-pat+ern')
279
self.assertEqual([key], filtered)
281
def test_matches_pattern(self):
282
match_key = FakeKey(275, 'job-foo', 27, 'file-pattern')
283
wrong_name = FakeKey(275, 'job-foo', 27, 'file-pat+ern')
284
wrong_job = FakeKey(275, 'job.foo', 27, 'file-pattern')
285
wrong_rb = FakeKey(276, 'job-foo', 27, 'file-pattern')
286
keys = [match_key, wrong_name, wrong_job, wrong_rb]
287
bucket = FakeBucket(keys)
288
filtered = find_file_keys(
289
bucket, 275, 'job-foo', 'file-pat+ern')
290
self.assertEqual([match_key], filtered)
292
def test_uses_latest_build(self):
293
bucket = FakeBucket([FakeKey(275, 'job-foo', 1, 'file-pattern')])
294
filtered = find_file_keys(bucket, 275, 'job-foo', 'file-pat+ern')
295
self.assertEqual([bucket.keys[0]], filtered)
297
bucket.keys.append(FakeKey(275, 'job-foo', 2, 'non-match'))
298
filtered = find_file_keys(bucket, 275, 'job-foo', 'file-pat+ern')
299
self.assertEqual([], filtered)
301
bucket.keys.append(FakeKey(275, 'job-foo', 2, 'file-pattern'))
302
filtered = find_file_keys(bucket, 275, 'job-foo', 'file-pat+ern')
303
self.assertEqual([bucket.keys[2]], filtered)
305
def test_pattern_is_path(self):
306
match_key = FakeKey(275, 'job-foo', 27, 'dir/file')
307
bucket = FakeBucket([match_key])
308
filtered = find_file_keys(bucket, 275, 'job-foo', 'dir/file')
309
self.assertEqual([match_key], filtered)
312
197
class TestMain(StrictTestCase):
315
def temp_credentials(self):
316
with NamedTemporaryFile() as temp_file:
317
temp_file.write(dedent("""\
319
access_key = fake_username
320
secret_key = fake_pass
326
200
use_context(self, stdout_guard())
328
def test_main_args_get_juju_bin(self):
202
def test_main_args(self):
329
203
stdout = StringIO()
330
with self.temp_credentials() as config_file:
204
with NamedTemporaryFile() as temp_file:
205
temp_file.write(dedent("""\
207
access_key = fake_username
208
secret_key = fake_pass
331
211
with patch('sys.argv', [
332
212
'foo', 'get-juju-bin', '28', 'bar-workspace',
333
'--config', config_file]):
213
'--config', temp_file.name]):
334
214
with patch('s3ci.S3Connection', autospec=True) as s3c_mock:
335
215
with patch('s3ci.fetch_juju_binary', autospec=True,
336
216
return_value='gjb') as gbj_mock:
339
219
s3c_mock.assert_called_once_with('fake_username', 'fake_pass')
340
220
gb_mock = s3c_mock.return_value.get_bucket
341
gb_mock.assert_called_once_with(JUJU_QA_DATA)
221
gb_mock.assert_called_once_with('juju-qa-data')
342
222
gbj_mock.assert_called_once_with(gb_mock.return_value, 28,
344
224
self.assertEqual('gjb\n', stdout.getvalue())
346
def test_main_args_get(self):
348
with self.temp_credentials() as config_file:
349
with patch('sys.argv', [
350
'foo', 'get', '28', 'foo-job', 'bar-file', 'bar-workspace',
351
'--config', config_file]):
352
with patch('s3ci.S3Connection', autospec=True) as s3c_mock:
353
with patch('s3ci.fetch_files', autospec=True,
354
return_value=['ff', 'gg']) as ff_mock:
355
with patch('sys.stdout', stdout):
357
s3c_mock.assert_called_once_with('fake_username', 'fake_pass')
358
gb_mock = s3c_mock.return_value.get_bucket
359
gb_mock.assert_called_once_with(JUJU_QA_DATA)
360
ff_mock.assert_called_once_with(gb_mock.return_value, 28,
361
'foo-job', 'bar-file', 'bar-workspace')
362
self.assertEqual('ff\ngg\n', stdout.getvalue())