-
Notifications
You must be signed in to change notification settings - Fork 61
/
tools_cv.py
209 lines (182 loc) · 8.2 KB
/
tools_cv.py
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
#!/usr/bin/env python
# coding: utf-8 -*-
#
# Copyright 2019 Arista Networks
#
# 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.
#
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
import logging
import traceback
from ansible.module_utils.connection import Connection
try:
from cvprac.cvp_client import CvpClient
from cvprac.cvp_client_errors import CvpLoginError
HAS_CVPRAC = True
except ImportError:
HAS_CVPRAC = False
CVPRAC_IMP_ERR = traceback.format_exc()
LOGGER = logging.getLogger('arista.cvp.cv_tools')
def cv_connect(module):
"""
cv_connect CV Connection method.
Generic Cloudvision connection method to connect to either on-prem or cvaas instances.
Parameters
----------
module : AnsibleModule
Ansible module information
Returns
-------
CvpClient
Instanciated CvpClient with connection information.
"""
client = CvpClient()
LOGGER.info('Connecting to CVP')
connection = Connection(module._socket_path)
# use 'cvaas' user for CVaaS connection and 'svc_account' for CV on-prem using service account token
svc_accounts = ['cvaas', 'svc_account']
host = connection.get_option("host")
port = connection.get_option("port")
cert_validation = connection.get_option("validate_certs")
is_cvaas = True if connection.get_option("remote_user") == 'cvaas' else False
api_token = connection.get_option("password") if connection.get_option("remote_user") in svc_accounts else None
user = connection.get_option("remote_user") if connection.get_option("remote_user") != 'cvaas' else ''
user_authentication = connection.get_option("password") if connection.get_option("remote_user") != 'cvaas' else ''
ansible_command_timeout = connection.get_option(
"persistent_command_timeout")
ansible_connect_timeout = connection.get_option(
"persistent_connect_timeout")
# The following is needed because of https://github.com/ansible/ansible/issues/75503
# Which was fixed in https://github.com/ansible/ansible/pull/78236
# This is a failsafe in case the ansible version is not high enough to have the fix
if isinstance(user_authentication, dict) and "__ansible_vault" in user_authentication:
LOGGER.error('Cannot connect to CVP, password is encrypted')
raise NotImplementedError("Vault encrypted variables are not supported for password with your version of ansible. "
"Because of https://github.com/ansible/ansible/issues/75503. "
"You may either upgrade ansible or use ansible vault file instead."
"https://docs.ansible.com/ansible/latest/user_guide/vault.html#encrypting-files-with-ansible-vault")
if cert_validation:
LOGGER.debug(" Module will check CV certificate")
if user == 'cvaas':
LOGGER.debug(' Connecting to a cvaas instance')
if user == 'svc_account':
LOGGER.debug(' Connecting to a on-prem instance using service account token')
LOGGER.debug(' Connecting to a CV instance: %s with timers %s %s',
str(host),
str(ansible_connect_timeout),
str(ansible_command_timeout))
try:
client.connect(nodes=[host],
username=user,
password=user_authentication,
api_token=api_token,
protocol="https",
is_cvaas=is_cvaas,
port=port,
cert=cert_validation,
request_timeout=ansible_command_timeout,
connect_timeout=ansible_connect_timeout
)
except CvpLoginError as e:
LOGGER.error('Cannot connect to CVP: %s', str(e))
module.fail_json(msg=str(e))
LOGGER.info('Connected to CVP')
return client
def isIterable(testing_object=None):
"""
Test if an object is iterable or not.
Test if an object is iterable or not. If yes return True, else return False.
Parameters
----------
testing_object : any, optional
Object to test if it is iterable or not, by default None
"""
try:
iter(testing_object)
return True
except TypeError as te: # noqa # pylint: disable=unused-variable
return False
def match_filter(input, filter, default_always='all'):
"""
Function to test if an object match userdefined filter.
Function support list of string and string as filter.
A default value is provided when calling function and if this default value for always matching is configured by user, then return True (Always matching)
If filter is a list, then we iterate over the input and check if it matches an entry in the filter.
Parameters
----------
input : string
Input to test of that match filter or not.
filter : list
List of string to compare against input.
default_always : str, optional
Keyword to consider as always matching, by default 'all'
default_none : str, optional
Keyword to consider as never matching, by default 'none'
Returns
-------
bool
True if input matchs filter, False in other situation
"""
# W102 Workaround to avoid list as default value.
if filter is None:
LOGGER.critical('Filter is not set, configure default value to [\'all\']')
filter = ["all"]
LOGGER.debug(" * is_in_filter - filter is %s", str(filter))
LOGGER.debug(" * is_in_filter - input string is %s", str(input))
if "all" in filter:
return True
elif any(element in input for element in filter):
return True
LOGGER.debug(" * is_in_filter - NOT matched")
return False
def cv_update_configlets_on_device(module, device_facts, add_configlets, del_configlets):
response = dict()
device_deletion = None
device_addition = None
# Initial Logging
LOGGER.debug(' * cv_update_configlets_on_device - add_configlets: %s', str(add_configlets))
LOGGER.debug(' * cv_update_configlets_on_device - del_configlets: %s', str(del_configlets))
# Work on delete configlet scenario
LOGGER.info(" * cv_update_configlets_on_device - start device deletion process")
if len(del_configlets) > 0:
try:
device_deletion = module.client.api.remove_configlets_from_device(
app_name="Ansible",
dev=device_facts,
del_configlets=del_configlets,
create_task=True
)
response = device_deletion
except Exception as error:
errorMessage = str(error)
LOGGER.error('OK, something wrong happens, raise an exception: %s', str(errorMessage))
LOGGER.info(" * cv_update_configlets_on_device - device_deletion result: %s", str(device_deletion))
# Work on Add configlet scenario
LOGGER.debug(" * cv_update_configlets_on_device - start device addition process")
if len(add_configlets) > 0:
LOGGER.debug(' * cv_update_configlets_on_device - ADD configlets: %s', str(add_configlets))
try:
device_addition = module.client.api.apply_configlets_to_device(
app_name="Ansible",
dev=device_facts,
new_configlets=add_configlets,
create_task=True
)
response.update(device_addition)
except Exception as error:
errorMessage = str(error)
LOGGER.error('OK, something wrong happens, raise an exception: %s', str(errorMessage))
LOGGER.info(" * cv_update_configlets_on_device - device_addition result: %s", str(device_addition))
LOGGER.info(" * cv_update_configlets_on_device - final result: %s", str(response))
return response