~veebers/juju-ci-tools/model_migration_check_all_units_of_charm

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
#!/usr/bin/env python
from __future__ import print_function

from argparse import ArgumentParser
from contextlib import contextmanager
from copy import deepcopy
from difflib import ndiff
import os
from pprint import pformat

import yaml

from jujupy import client_from_config
from utility import (
    add_arg_juju_bin,
    JujuAssertionError,
    )


def remove_display_attributes(cloud):
    """Remove the attributes added by display.

    The 'defined' attribute is asserted to be 'local'.
    The description attribute is asserted to be appropriate for the cloud type.
    """
    type_descriptions = {
        'openstack': 'Openstack Cloud',
        'vsphere': '',
        'manual': '',
        'maas': 'Metal As A Service',
        }
    # The lack of built-in descriptions for vsphere and manual is
    # bug #1646128.  The inability to specify descriptions interactively is
    # bug #1645783.
    defined = cloud.pop('defined')
    assert_equal(defined, 'local')
    description = cloud.pop('description')
    assert_equal(description, type_descriptions[cloud['type']])


def get_clouds(client):
    cloud_list = yaml.safe_load(client.get_juju_output(
        'clouds', '--format', 'yaml', include_e=False))
    for cloud_name, cloud in cloud_list.items():
        if cloud['defined'] == 'built-in':
            del cloud_list[cloud_name]
            continue
        remove_display_attributes(cloud)
    return cloud_list


def get_home_path(client, subpath):
    return os.path.join(client.env.juju_home, subpath)


def assert_equal(first, second):
    """If two values are not the same, raise JujuAssertionError.

    The text of the error is a diff of the pretty-printed values.
    """
    if first != second:
        diff = ndiff(pformat(first).splitlines(), pformat(second).splitlines())
        raise JujuAssertionError('\n' + '\n'.join(diff))


def assess_clouds(client, expected):
    """Assess how clouds behaves when only expected clouds are defined."""
    cloud_list = get_clouds(client)
    assert_equal(cloud_list, expected)


def assess_show_cloud(client, expected):
    """Assess how show-cloud behaves."""
    for cloud_name, expected_cloud in expected.items():
        actual_cloud = yaml.safe_load(client.get_juju_output(
            'show-cloud', cloud_name, '--format', 'yaml', include_e=False))
        remove_display_attributes(actual_cloud)
        assert_equal(actual_cloud, expected_cloud)


def strip_redundant_endpoints(clouds):
    no_region_endpoint = deepcopy(clouds)
    for cloud in no_region_endpoint.values():
        for region in cloud.get('regions', {}).values():
            if region['endpoint'] == cloud['endpoint']:
                region.pop('endpoint')
    return no_region_endpoint


@contextmanager
def testing(test_name):
    try:
        yield
    except Exception:
        print('{}: FAIL'.format(test_name))
        raise
    else:
        print('{}: PASS'.format(test_name))


def main():
    parser = ArgumentParser()
    parser.add_argument('clouds_file')
    add_arg_juju_bin(parser)
    args = parser.parse_args()
    client = client_from_config(None, args.juju_bin)
    with client.env.make_jes_home(client.env.juju_home, 'mytest',
                                  {}) as juju_home:
        client.env.juju_home = juju_home
        with open(get_home_path(client, 'public-clouds.yaml'), 'w') as f:
            f.write('')
        with testing('assess_clouds (no_clouds)'):
            assess_clouds(client, {})
        with open(args.clouds_file) as f:
            supplied_clouds = yaml.safe_load(f.read().decode('utf-8'))
        client.env.write_clouds(client.env.juju_home, supplied_clouds)
        no_region_endpoint = strip_redundant_endpoints(
            supplied_clouds['clouds'])
        with testing('assess_clouds'):
            assess_clouds(client, no_region_endpoint)
        with testing('assess_show_cloud'):
            assess_show_cloud(client, no_region_endpoint)


if __name__ == '__main__':
    main()