← Back to branch summary

~andrewjbeach/juju-ci-tools/make-local-patcher

« back to all changes in this revision

Viewing changes to tests/test_assess_autoload_credentials.py

  • Committer: Curtis Hovey
  • Date: 2016-09-20 01:59:47 UTC
  • mto: This revision was merged to the branch mainline in revision 1602.
  • Revision ID: curtis@canonical.com-20160920015947-ko27xkj3a4i774h6
Convert juju instance=ids to true azuzre ids.

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/test_assess_autoload_credentials.py
 
1
"""Tests for assess_autoload_credentials module."""
 
2
 
 
3
from argparse import Namespace
 
4
import ConfigParser
 
5
import logging
 
6
from mock import patch
 
7
import os
 
8
import StringIO
 
9
from textwrap import dedent
 
10
 
 
11
import yaml
 
12
 
 
13
import assess_autoload_credentials as aac
 
14
from deploy_stack import BootstrapManager
 
15
from tests import (
 
16
    TestCase,
 
17
    parse_error,
 
18
    )
 
19
from tests.test_jujupy import fake_juju_client
 
20
from utility import temp_dir
 
21
 
 
22
 
 
23
class TestParseArgs(TestCase):
 
24
 
 
25
    def test_common_args(self):
 
26
        args = aac.parse_args(['env', '/bin/juju'])
 
27
        self.assertEqual('env', args.env)
 
28
        self.assertEqual('/bin/juju', args.juju_bin)
 
29
 
 
30
    def test_help(self):
 
31
        fake_stdout = StringIO.StringIO()
 
32
        with parse_error(self) as fake_stderr:
 
33
            with patch('sys.stdout', fake_stdout):
 
34
                aac.parse_args(['--help'])
 
35
        self.assertEqual('', fake_stderr.getvalue())
 
36
        self.assertIn(
 
37
            'Test autoload-credentials command.', fake_stdout.getvalue())
 
38
 
 
39
    def test_verbose_is_set_to_debug_when_passed_verbose(self):
 
40
        args = aac.parse_args(['/bin/juju', '--verbose'])
 
41
        self.assertEqual(logging.DEBUG, args.verbose)
 
42
 
 
43
    def test_verbose_default_values(self):
 
44
        env = 'env'
 
45
        juju_bin = '/bin/juju'
 
46
        temp_env_name = 'functional-autoload-credentials'
 
47
        with temp_dir() as log:
 
48
            args = aac.parse_args([env, juju_bin, log, temp_env_name])
 
49
        self.assertEqual(
 
50
            args,
 
51
            Namespace(agent_stream=None, agent_url=None, bootstrap_host=None,
 
52
                      debug=False, env='env', juju_bin='/bin/juju',
 
53
                      keep_env=False, logs=log, machine=[],
 
54
                      region=None, series=None,
 
55
                      temp_env_name='functional-autoload-credentials',
 
56
                      upload_tools=False, verbose=logging.INFO,
 
57
                      ))
 
58
 
 
59
 
 
60
class TestCredentialIdCounter(TestCase):
 
61
 
 
62
    def setUp(self):
 
63
        # Make sure CredentialIdCounter is reset to initial values
 
64
        aac.CredentialIdCounter._counter.clear()
 
65
 
 
66
    def test_returns_zero_for_new_id(self):
 
67
        self.assertEqual(aac.CredentialIdCounter.id('test'), 0)
 
68
 
 
69
    def test_returns_iterations_for_same_id(self):
 
70
        generated_ids = [
 
71
            aac.CredentialIdCounter.id('test') for x in xrange(3)
 
72
        ]
 
73
        self.assertEqual(generated_ids, [0, 1, 2])
 
74
 
 
75
    def test_returns_new_ids_for_multiple_names(self):
 
76
        self.assertEqual(aac.CredentialIdCounter.id('test'), 0)
 
77
        self.assertEqual(aac.CredentialIdCounter.id('another_test'), 0)
 
78
        self.assertEqual(aac.CredentialIdCounter.id('test'), 1)
 
79
        self.assertEqual(aac.CredentialIdCounter.id('another_test'), 1)
 
80
        self.assertEqual(aac.CredentialIdCounter.id('test'), 2)
 
81
 
 
82
 
 
83
class TestAWSHelpers(TestCase):
 
84
 
 
85
    def test_credential_dict_generator_returns_different_details(self):
 
86
        """Each call must return unique details each time."""
 
87
        first_details = aac.aws_credential_dict_generator()
 
88
        second_details = aac.aws_credential_dict_generator()
 
89
 
 
90
        self.assertNotEqual(first_details, second_details)
 
91
 
 
92
    def test_get_aws_environment_supplies_all_keys(self):
 
93
        access_key = 'access_key'
 
94
        secret_key = 'secret_key'
 
95
        username = 'username'
 
96
 
 
97
        env = aac.get_aws_environment(username, access_key, secret_key)
 
98
 
 
99
        self.assertDictEqual(
 
100
            env,
 
101
            dict(
 
102
                USER=username,
 
103
                AWS_ACCESS_KEY_ID=access_key,
 
104
                AWS_SECRET_ACCESS_KEY=secret_key))
 
105
 
 
106
    def test_aws_envvar_test_details_returns_correct_expected_details(self):
 
107
        access_key = 'test_access_key'
 
108
        secret_key = 'test_secret_key'
 
109
        username = 'user'
 
110
        cloud_details = aac.aws_envvar_test_details(
 
111
            username,
 
112
            'tmp_dir',
 
113
            client=None,
 
114
            credential_details={
 
115
                'access_key': access_key,
 
116
                'secret_key': secret_key})
 
117
 
 
118
        self.assertDictEqual(
 
119
            cloud_details.expected_details, {
 
120
                'credentials': {
 
121
                    'aws': {
 
122
                        username: {
 
123
                            'auth-type': 'access-key',
 
124
                            'access-key': access_key,
 
125
                            'secret-key': secret_key,
 
126
                            }
 
127
                        }
 
128
                    }
 
129
                })
 
130
 
 
131
    def test_aws_envvar_test_details_returns_correct_envvar_settings(self):
 
132
        access_key = 'test_access_key'
 
133
        secret_key = 'test_secret_key'
 
134
        username = 'user'
 
135
        cloud_details = aac.aws_envvar_test_details(
 
136
            username,
 
137
            'tmp_dir',
 
138
            client=None,
 
139
            credential_details={
 
140
                'access_key': access_key,
 
141
                'secret_key': secret_key})
 
142
 
 
143
        self.assertDictEqual(
 
144
            cloud_details.env_var_changes,
 
145
            dict(
 
146
                USER=username,
 
147
                AWS_ACCESS_KEY_ID=access_key,
 
148
                AWS_SECRET_ACCESS_KEY=secret_key))
 
149
 
 
150
    def test_aws_directory_test_details_returns_correct_expected_details(self):
 
151
        access_key = 'test_access_key'
 
152
        secret_key = 'test_secret_key'
 
153
        username = 'user'
 
154
        with patch.object(aac, 'write_aws_config_file'):
 
155
            cloud_details = aac.aws_directory_test_details(
 
156
                username,
 
157
                'tmp_dir',
 
158
                client=None,
 
159
                credential_details={
 
160
                    'access_key': access_key, 'secret_key': secret_key})
 
161
 
 
162
        self.assertDictEqual(
 
163
            cloud_details.expected_details, {
 
164
                'credentials': {
 
165
                    'aws': {
 
166
                        username: {
 
167
                            'auth-type': 'access-key',
 
168
                            'access-key': access_key,
 
169
                            'secret-key': secret_key,
 
170
                            }
 
171
                        }
 
172
                    }
 
173
                })
 
174
 
 
175
    def test_aws_directory_test_details_returns_envvar_settings(self):
 
176
        with patch.object(aac, 'write_aws_config_file'):
 
177
            cloud_details = aac.aws_directory_test_details(
 
178
                'username',
 
179
                'tmp_dir',
 
180
                client=None)
 
181
        self.assertDictEqual(
 
182
            cloud_details.env_var_changes,
 
183
            dict(HOME='tmp_dir'))
 
184
 
 
185
    def test_write_aws_config_file_writes_credentials_file(self):
 
186
        """Ensure the file created contains the correct details."""
 
187
        user = 'different-user'
 
188
        access_key = 'access_key'
 
189
        secret_key = 'secret_key'
 
190
 
 
191
        with temp_dir() as tmp_dir:
 
192
            credentials_file = aac.write_aws_config_file(
 
193
                user, tmp_dir, access_key, secret_key)
 
194
            credentials = ConfigParser.ConfigParser()
 
195
            with open(credentials_file, 'r') as f:
 
196
                credentials.readfp(f)
 
197
 
 
198
        expected_items = [
 
199
            ('aws_access_key_id', access_key),
 
200
            ('aws_secret_access_key', secret_key)]
 
201
 
 
202
        self.assertEqual(credentials.sections(), [user])
 
203
        self.assertEqual(
 
204
            credentials.items(user), expected_items)
 
205
 
 
206
 
 
207
class TestOpenStackHelpers(TestCase):
 
208
 
 
209
    def test_credential_dict_generator_returns_different_details(self):
 
210
        """Each call must return uniquie details each time."""
 
211
        first_details = aac.openstack_credential_dict_generator('region1')
 
212
        second_details = aac.openstack_credential_dict_generator('region1')
 
213
 
 
214
        self.assertNotEqual(first_details, second_details)
 
215
 
 
216
    def test_expected_details_dict_returns_correct_values(self):
 
217
        os_username = 'username'
 
218
        os_password = 'password'
 
219
        os_tenant_name = 'tenant name'
 
220
        os_auth_url = 'url',
 
221
        os_region_name = 'region'
 
222
        expected_details = aac.get_openstack_expected_details_dict(
 
223
            os_username, {
 
224
                'os_password': os_password,
 
225
                'os_tenant_name': os_tenant_name,
 
226
                'os_auth_url': os_auth_url,
 
227
                'os_region_name': os_region_name,
 
228
                'os_user_name': os_username,
 
229
                })
 
230
        self.assertEqual(
 
231
            expected_details, {
 
232
                'credentials': {
 
233
                    'testing-openstack': {
 
234
                        'default-region': 'region',
 
235
                        os_username: {
 
236
                            'auth-type': 'userpass',
 
237
                            'domain-name': '',
 
238
                            'password': os_password,
 
239
                            'tenant-name': os_tenant_name,
 
240
                            'username': os_username
 
241
                            }
 
242
                        }
 
243
                    }
 
244
                })
 
245
 
 
246
    def test_get_openstack_envvar_changes_returns_correct_values(self):
 
247
        user = 'username'
 
248
        os_password = 'password'
 
249
        os_tenant_name = 'tenant name'
 
250
        os_auth_url = 'url',
 
251
        os_region_name = 'region'
 
252
        env_var_changes = aac.get_openstack_envvar_changes(
 
253
            user, {
 
254
                'os_password': os_password,
 
255
                'os_tenant_name': os_tenant_name,
 
256
                'os_auth_url': os_auth_url,
 
257
                'os_region_name': os_region_name,
 
258
                })
 
259
 
 
260
        self.assertEqual(
 
261
            env_var_changes, {
 
262
                'USER': user,
 
263
                'OS_USERNAME': user,
 
264
                'OS_PASSWORD': os_password,
 
265
                'OS_TENANT_NAME': os_tenant_name,
 
266
                'OS_AUTH_URL': os_auth_url,
 
267
                'OS_REGION_NAME': os_region_name,
 
268
                })
 
269
 
 
270
    def test_write_openstack_config_file_writes_credentials_file(self):
 
271
        """Ensure the file created contains the correct details."""
 
272
        credential_details = dict(
 
273
            os_tenant_name='tenant_name',
 
274
            os_password='password',
 
275
            os_auth_url='url',
 
276
            os_region_name='region')
 
277
        user = 'username'
 
278
 
 
279
        with temp_dir() as tmp_dir:
 
280
            credentials_file = aac.write_openstack_config_file(
 
281
                tmp_dir, user, credential_details)
 
282
            with open(credentials_file, 'r') as f:
 
283
                credential_contents = f.read()
 
284
 
 
285
        expected = dedent("""\
 
286
        export OS_USERNAME={user}
 
287
        export OS_PASSWORD={password}
 
288
        export OS_TENANT_NAME={tenant_name}
 
289
        export OS_AUTH_URL={auth_url}
 
290
        export OS_REGION_NAME={region}
 
291
        """.format(
 
292
            user=user,
 
293
            password=credential_details['os_password'],
 
294
            tenant_name=credential_details['os_tenant_name'],
 
295
            auth_url=credential_details['os_auth_url'],
 
296
            region=credential_details['os_region_name'],
 
297
            ))
 
298
 
 
299
        self.assertEqual(credential_contents, expected)
 
300
 
 
301
 
 
302
class TestGCEHelpers(TestCase):
 
303
    def test_get_gce_expected_details_dict_returns_correct_details(self):
 
304
        user = 'username'
 
305
        cred_path = '/some/path'
 
306
        self.assertEqual(
 
307
            aac.get_gce_expected_details_dict(user, cred_path),
 
308
            {
 
309
                'credentials': {
 
310
                    'google': {
 
311
                        user: {
 
312
                            'auth-type': 'jsonfile',
 
313
                            'file': cred_path,
 
314
                            }
 
315
                        }
 
316
                    }
 
317
                })
 
318
 
 
319
    def test_gce_credential_dict_generator_returns_unique_details(self):
 
320
        self.assertNotEqual(
 
321
            aac.gce_credential_dict_generator(),
 
322
            aac.gce_credential_dict_generator())
 
323
 
 
324
    def test_write_gce_config_file_creates_unique_credential_file(self):
 
325
        credentials = dict(
 
326
            client_id='client_id',
 
327
            client_email='client_email',
 
328
            private_key='private_key',
 
329
            )
 
330
 
 
331
        with patch.object(aac.CredentialIdCounter, 'id') as id_gen:
 
332
            id_gen.return_value = 0
 
333
            with temp_dir() as tmp_dir:
 
334
                file_path = aac.write_gce_config_file(tmp_dir, credentials)
 
335
        self.assertEqual(
 
336
            file_path,
 
337
            os.path.join(tmp_dir, 'gce-file-config-{}.json'.format(0)))
 
338
 
 
339
    def test_write_gce_config_file_creates_named_credential_file(self):
 
340
        credentials = dict(
 
341
            client_id='client_id',
 
342
            client_email='client_email',
 
343
            private_key='private_key',
 
344
            )
 
345
 
 
346
        with temp_dir() as tmp_dir:
 
347
            file_path = aac.write_gce_config_file(
 
348
                tmp_dir, credentials, 'file_name')
 
349
        self.assertEqual(file_path, os.path.join(tmp_dir, 'file_name'))
 
350
 
 
351
    def test_credential_generator_returns_correct_formats(self):
 
352
        """Three items are needed, one of them must be an email address."""
 
353
        with patch.object(aac.CredentialIdCounter, 'id') as id_gen:
 
354
            id_gen.return_value = 0
 
355
            details = aac.gce_credential_dict_generator()
 
356
 
 
357
            self.assertEqual(details['private_key'], 'gce-credentials-0')
 
358
            self.assertEqual(details['client_id'], 'gce-credentials-0')
 
359
            self.assertEqual(
 
360
                details['client_email'], 'gce-credentials-0@example.com')
 
361
 
 
362
 
 
363
class TestAssertCredentialsContainsExpectedResults(TestCase):
 
364
 
 
365
    def test_does_not_raise_when_credentials_match(self):
 
366
        cred_actual = dict(key='value')
 
367
        cred_expected = dict(key='value')
 
368
 
 
369
        aac.assert_credentials_contains_expected_results(
 
370
            cred_actual, cred_expected)
 
371
 
 
372
    def test_raises_when_credentials_do_not_match(self):
 
373
        cred_actual = dict(key='value')
 
374
        cred_expected = dict(key='value', another_key='extra')
 
375
 
 
376
        self.assertRaises(
 
377
            ValueError,
 
378
            aac.assert_credentials_contains_expected_results,
 
379
            cred_actual,
 
380
            cred_expected)
 
381
 
 
382
 
 
383
def bogus_credentials():
 
384
    """return an client with unusable crednetials.
 
385
 
 
386
    It uses an openstack config to match the fake_juju.AutoloadCredentials.
 
387
    """
 
388
    client = fake_juju_client()
 
389
    client.env.config['type'] = 'openstack'
 
390
    client.env.config['auth-url'] = 'url'
 
391
    client.env.config['region'] = 'region'
 
392
    client.env.credentials = {
 
393
        'credentials': {'bogus': {}}}
 
394
    return client
 
395
 
 
396
 
 
397
class TestEnsureAutoloadCredentialsStoresDetails(TestCase):
 
398
 
 
399
    def test_existing_credentials_openstack(self):
 
400
 
 
401
            aac.ensure_autoload_credentials_stores_details(
 
402
                bogus_credentials(), aac.openstack_envvar_test_details)
 
403
 
 
404
 
 
405
class TestEnsureAutoloadCredentialsOverwriteExisting(TestCase):
 
406
 
 
407
    def test_overwrite_existing(self):
 
408
            aac.ensure_autoload_credentials_overwrite_existing(
 
409
                bogus_credentials(), aac.openstack_envvar_test_details)
 
410
 
 
411
 
 
412
class TestAutoloadAndBootstrap(TestCase):
 
413
 
 
414
    def test_autoload_and_bootstrap(self):
 
415
 
 
416
        def cloud_details_fn(user, tmp_dir, client, credential_details):
 
417
            return aac.CloudDetails(credential_details, None, None)
 
418
 
 
419
        client = fake_juju_client()
 
420
        upload_tools = False
 
421
        real_credential_details = {'cloud-username': 'user',
 
422
                                   'cloud-password': 'password'
 
423
                                   }
 
424
        credential_file_details = {'credentials': {'cloud': {
 
425
                                   'credentials': real_credential_details
 
426
                                   }}}
 
427
 
 
428
        def write_credentials(*args, **kwargs):
 
429
            file_name = os.path.join(client.env.juju_home, 'credentials.yaml')
 
430
            with open(file_name, 'w') as write_file:
 
431
                yaml.safe_dump(credential_file_details, write_file)
 
432
 
 
433
        def credential_check(*args, **kwargs):
 
434
            self.assertEqual(client.env.credentials, credential_file_details)
 
435
 
 
436
        with temp_dir() as log_dir:
 
437
            bs_manager = BootstrapManager(
 
438
                'env', client, client, None, [], None, None, None, None,
 
439
                log_dir, False, True, True)
 
440
            with patch('assess_autoload_credentials.run_autoload_credentials',
 
441
                       autospec=True,
 
442
                       side_effect=write_credentials) as run_autoload_mock:
 
443
                with patch.object(
 
444
                        bs_manager.client, 'bootstrap',
 
445
                        autospec=True,
 
446
                        side_effect=credential_check) as bootstrap_mock:
 
447
                    aac.autoload_and_bootstrap(bs_manager, upload_tools,
 
448
                                               real_credential_details,
 
449
                                               cloud_details_fn)
 
450
        run_autoload_mock.assert_called_once_with(
 
451
            client, real_credential_details, None)
 
452
        bootstrap_mock.assert_called_once_with(False, bootstrap_series=None,
 
453
                                               credential='testing-user')