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
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
|
# ============LICENSE_START=======================================================
# org.onap.dcae
# ================================================================================
# Copyright (c) 2017-2018 AT&T Intellectual Property. All rights reserved.
# ================================================================================
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ============LICENSE_END=========================================================
#
# ECOMP is a trademark and service mark of AT&T Intellectual Property.
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Provides dcae cli profile variables
"""
import os
from collections import namedtuple
import six
import click
from dcae_cli import util
from dcae_cli.util import get_app_dir, get_pref, write_pref
from dcae_cli.util import config
from dcae_cli.util.config import get_config, update_config
from dcae_cli.util.exc import DcaeException
from dcae_cli.util.logger import get_logger
logger = get_logger('Profile')
# reserved profile names
ACTIVE = 'active'
_reserved_names = {ACTIVE}
# create enums for profile keys so that they can be imported for testing, instead of using literals
CONSUL_HOST = 'consul_host'
CONFIG_BINDING_SERVICE = 'config_binding_service'
CDAP_BROKER = 'cdap_broker'
DOCKER_HOST = 'docker_host'
# TODO: Should probably lift this strict list of allowed keys and repurpose to be
# keys that are required.
_allowed_keys = set([CONSUL_HOST, CONFIG_BINDING_SERVICE, CDAP_BROKER, DOCKER_HOST])
Profile = namedtuple('Profile', _allowed_keys)
def _create_stub_profile():
"""Create a new stub of a profile"""
return { k: "" for k in _allowed_keys }
def _fmt_seq(seq):
'''Returns a sorted string formatted list'''
return list(sorted(map(str, seq)))
def get_profiles_path():
'''Returns the absolute path to the profiles file'''
return os.path.join(get_app_dir(), 'profiles.json')
def get_active_name():
'''Returns the active profile name in the config'''
return config.get_active_profile()
def _set_active_name(name):
'''Sets the active profile name in the config'''
update_config(active_profile=name)
class ProfilesInitError(RuntimeError):
pass
def reinit_profiles():
"""Reinitialize profiles
Grab the remote profiles and merge with the local profiles if there is one.
Returns:
--------
Dict of complete new profiles
"""
# Grab the remote profiles and merge it in
try:
server_url = config.get_server_url()
new_profiles = util.fetch_file_from_web(server_url, "/dcae-cli/profiles.json")
except:
# Failing to pull seed profiles from remote server is not considered
# a problem. Just continue and give user the option to use an empty
# default.
if click.confirm("Could not download initial profiles from remote server. Set empty default?"):
new_profiles = {"default": { "consul_host": "",
"config_binding_service": "config_binding_service",
"cdap_broker": "cdap_broker", "docker_host": ""}}
else:
raise ProfilesInitError("Could not setup dcae-cli profiles")
profiles_path = get_profiles_path()
if util.pref_exists(profiles_path):
existing_profiles = get_profiles(include_active=False)
# Make sure to clobber existing values and not other way
existing_profiles.update(new_profiles)
new_profiles = existing_profiles
write_pref(new_profiles, profiles_path)
return new_profiles
def get_profiles(user_only=False, include_active=True):
'''Returns a dict containing all available profiles
Example of the returned dict:
{
"profile-foo": {
"some_variable_A": "some_value_A",
"some_variable_B": "some_value_B",
"some_variable_C": "some_value_C"
}
}
'''
try:
profiles = get_pref(get_profiles_path(), reinit_profiles)
except ProfilesInitError as e:
raise DcaeException("Failed to initialize profiles: {0}".format(e))
if user_only:
return profiles
if include_active:
active_name = get_active_name()
if active_name not in profiles:
raise DcaeException("Active profile '{}' does not exist. How did this happen?".format(active_name))
profiles[ACTIVE] = profiles[active_name]
return profiles
def get_profile(name=ACTIVE):
'''Returns a `Profile` object'''
profiles = get_profiles()
if name not in profiles:
raise DcaeException("Specified profile '{}' does not exist.".format(name))
try:
profile = Profile(**profiles[name])
except TypeError as e:
raise DcaeException("Specified profile '{}' is malformed.".format(name))
return profile
def create_profile(name, **kwargs):
'''Creates a new profile'''
_assert_not_reserved(name)
profiles = get_profiles(user_only=True)
if name in profiles:
raise DcaeException("Profile '{}' already exists.".format(name))
profile = _create_stub_profile()
profile.update(kwargs)
_assert_valid_profile(profile)
profiles[name] = profile
_write_profiles(profiles)
def delete_profile(name):
'''Deletes a profile'''
_assert_not_reserved(name)
profiles = get_profiles(user_only=True)
if name not in profiles:
raise DcaeException("Profile '{}' does not exist.".format(name))
if name == get_active_name():
logger.warning("Profile '{}' is currently active. Activate another profile first."
.format(name))
return False
del profiles[name]
_write_profiles(profiles)
return True
def update_profile(name, **kwargs):
'''Creates or updates a profile'''
_assert_not_reserved(name)
_assert_valid_profile(kwargs)
profiles = get_profiles(user_only=True)
if name not in profiles:
raise DcaeException("Profile '{}' does not exist.".format(name))
profiles[name].update(kwargs)
_write_profiles(profiles)
def _assert_valid_profile(params):
'''Raises DcaeException if the profile parameter dict is invalid'''
if not params:
raise DcaeException('No update key-value pairs were provided.')
keys = set(params.keys())
if not _allowed_keys.issuperset(keys):
invalid_keys = keys - _allowed_keys
raise DcaeException("Invalid keys {} detected. Only keys {} are supported.".format(_fmt_seq(invalid_keys), _fmt_seq(_allowed_keys)))
def _assert_not_reserved(name):
'''Raises DcaeException if the profile is reserved'''
if name in _reserved_names:
raise DcaeException("Profile '{}' is reserved and cannot be modified.".format(name))
def _write_profiles(profiles):
'''Writes the profiles dictionary to disk'''
return write_pref(profiles, path=get_profiles_path())
def activate_profile(name):
'''Modifies the config and sets a new active profile'''
avail_profiles = set(get_profiles().keys()) - {ACTIVE, }
if name not in avail_profiles:
raise DcaeException("Profile name '{}' does not exist. Please select from {} or create a new profile.".format(name, _fmt_seq(avail_profiles)))
_set_active_name(name)
|
