summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
-rw-r--r--Makefile4
-rw-r--r--VERSION2
-rw-r--r--cvprac/__init__.py2
-rw-r--r--cvprac/cvp_api.py830
-rw-r--r--cvprac/cvp_client.py129
-rw-r--r--dev-requirements.txt1
-rw-r--r--docs/labs/lab06-provisioning/vc_task_retrigger.py4
-rw-r--r--docs/release-notes-1.4.0.rst20
-rw-r--r--requirements.txt1
-rw-r--r--setup.py11
10 files changed, 509 insertions, 495 deletions
diff --git a/Makefile b/Makefile
index 2c437d8..f981df1 100644
--- a/Makefile
+++ b/Makefile
@@ -49,7 +49,9 @@ coverage_report:
pep8:
-pep8 -r --max-line-length=120 --ignore=$(PEP8_IGNORE) cvprac/
- -pep8 -r --max-line-length=120 --ignore=$(PEP8_IGNORE),E402 test/
+ -pep8 -r --max-line-length=120 --ignore=$(PEP8_IGNORE),E402 test/lib/
+ -pep8 -r --max-line-length=120 --ignore=$(PEP8_IGNORE),E402 test/system/
+ -pep8 -r --ignore=$(PEP8_IGNORE),E402,E501 test/unit/
pyflakes:
pyflakes cvprac/ test/
diff --git a/VERSION b/VERSION
index 1892b92..88c5fb8 100644
--- a/VERSION
+++ b/VERSION
@@ -1 +1 @@
-1.3.2
+1.4.0
diff --git a/cvprac/__init__.py b/cvprac/__init__.py
index 8d62fa0..67d57ff 100644
--- a/cvprac/__init__.py
+++ b/cvprac/__init__.py
@@ -32,5 +32,5 @@
''' RESTful API Client class for Cloudvision(R) Portal
'''
-__version__ = '1.3.2'
+__version__ = '1.4.0'
__author__ = 'Arista Networks, Inc.'
diff --git a/cvprac/cvp_api.py b/cvprac/cvp_api.py
index c390989..0d73658 100644
--- a/cvprac/cvp_api.py
+++ b/cvprac/cvp_api.py
@@ -1,3 +1,4 @@
+# pylint: disable=fixme,too-many-locals,redefined-builtin
#
# Copyright (c) 2017, Arista Networks, Inc.
# All rights reserved.
@@ -35,7 +36,6 @@ import operator
import os
import time
# This import is for proper file IO handling support for both Python 2 and 3
-# pylint: disable=redefined-builtin
from io import open
from datetime import datetime
from re import split
@@ -56,7 +56,53 @@ OPERATOR_DICT = {
}
-class CvpApi(object):
+def sanitize_warnings(data):
+ ''' Sanitize the warnings returned after validation.
+
+ In some cases where the configlets have both errors
+ and warnings, CVP may split warnings that have
+ `,` across multiple strings.
+ This method concatenates the strings back into one string
+ per warning, and corrects the warningCount.
+
+ Args:
+ data (dict): A dict that contians the result
+ of the validation operation
+ Returns:
+ response (dict): A dict that contains the result of the
+ validation operation
+ '''
+ if "warnings" not in data:
+ # nothing to do here, we can return as is
+ return data
+ # Since there may be warnings incorrectly split on
+ # ', ' within the warning text by CVP, we join all the
+ # warnings together using ', ' into one large string
+ temp_warnings = ", ".join(data['warnings']).strip()
+
+ # To split the large string again we match on the
+ # 'at line XXX' that should indicate the end of the warning.
+ # We capture as well the remaining \\n or whitespace and include
+ # the extra ', ' added in the previous step in the matching criteria.
+ # The extra ', ' is not included in the strings of the new list
+ temp_warnings = split(
+ r'(.*?at line \d+.*?),\s+',
+ temp_warnings
+ )
+
+ # The behaviour of re.split will add empty strings
+ # if the regex matches on the begging or ending of the line.
+ # Refer to https://docs.python.org/3/library/re.html#re.split
+
+ # Use filter to remove any empty strings
+ # that re.split inserted
+ data['warnings'] = list(filter(None, temp_warnings))
+ # Update the count of warnings to the correct value
+ data['warningCount'] = len(data['warnings'])
+ return data
+
+
+class CvpApi():
''' CvpApi class contains calls to CVP RESTful API. The RESTful API
parameters are passed in as parameters to the method. The results of
the RESTful API call are converted from json to a dict and returned.
@@ -110,8 +156,7 @@ class CvpApi(object):
running CVP version to.
'''
if opr not in OPERATOR_DICT:
- self.log.error('%s is an invalid operation for version comparison'
- % opr)
+ self.log.error("%s is an invalid operation for version comparison", opr)
return False
# Since CVaaS is automatically the latest version of the API, if
@@ -165,9 +210,9 @@ class CvpApi(object):
user_type (str): type of AAA (Local/TACACS/RADIUS)
'''
if status not in ['Enabled', 'Disabled']:
- self.log.error('Invalid status %s.'
- ' Status must be Enabled or Disabled.'
- ' Defaulting to Disabled' % status)
+ self.log.error(f"Invalid status {status}."
+ " Status must be Enabled or Disabled."
+ " Defaulting to Disabled")
status = 'Disabled'
data = {"roles": [role],
"user": {"contactNumber": "",
@@ -198,9 +243,9 @@ class CvpApi(object):
user_type (str): type of AAA (Local/TACACS/RADIUS)
'''
if status not in ['Enabled', 'Disabled']:
- self.log.error('Invalid status %s.'
- ' Status must be Enabled or Disabled.'
- ' Defaulting to Disabled' % status)
+ self.log.error(f"Invalid status {status}."
+ f" Status must be Enabled or Disabled."
+ f" Defaulting to Disabled")
status = 'Disabled'
data = {"roles": [role],
"user": {"contactNumber": "",
@@ -211,7 +256,7 @@ class CvpApi(object):
"userId": username,
"userStatus": status,
"userType": user_type}}
- return self.clnt.post('/user/updateUser.do?userId={}'.format(username),
+ return self.clnt.post(f"/user/updateUser.do?userId={username}",
data=data, timeout=self.request_timeout)
def get_user(self, username):
@@ -220,7 +265,7 @@ class CvpApi(object):
Args:
username (str): username on CVP
'''
- return self.clnt.get('/user/getUser.do?userId={}'.format(username),
+ return self.clnt.get(f"/user/getUser.do?userId={qplus(username)}",
timeout=self.request_timeout)
def get_users(self, query='', start=0, end=0):
@@ -248,10 +293,9 @@ class CvpApi(object):
'currentStatus': 'Online',
'addedByUser': 'cvp system'}]}
'''
- self.log.debug('get_users: query: %s' % query)
- return self.clnt.get('/user/getUsers.do?'
- 'queryparam=%s&startIndex=%d&endIndex=%d' %
- (qplus(query), start, end),
+ self.log.debug(f"get_users: query: {query}")
+ return self.clnt.get(f"/user/getUsers.do?"
+ f"queryparam={qplus(query)}&startIndex={start}&endIndex={end}",
timeout=self.request_timeout)
def delete_user(self, username):
@@ -275,12 +319,12 @@ class CvpApi(object):
task (dict): The CVP task for the associated Id. Returns None
if the task_id was invalid.
'''
- self.log.debug('get_task_by_id: task_id: %s' % task_id)
+ self.log.debug(f"get_task_by_id: task_id: {task_id}")
try:
- task = self.clnt.get('/task/getTaskById.do?taskId=%s' % task_id,
+ task = self.clnt.get(f"/task/getTaskById.do?taskId={task_id}",
timeout=self.request_timeout)
except CvpApiError as error:
- self.log.debug('Caught error: %s attempting to get task.' % error)
+ self.log.debug(f"Caught error: {error} attempting to get task.")
# Catch an invalid task_id error and return None
return None
return task
@@ -297,10 +341,10 @@ class CvpApi(object):
Returns:
tasks (list): The list of tasks
'''
- self.log.debug('get_tasks_by_status: status: %s' % status)
+ self.log.debug(f"get_tasks_by_status: status: {status}")
data = self.clnt.get(
- '/task/getTasks.do?queryparam=%s&startIndex=%d&endIndex=%d' %
- (status, start, end), timeout=self.request_timeout)
+ f"/task/getTasks.do?queryparam={status}&startIndex={start}&endIndex={end}",
+ timeout=self.request_timeout)
return data['data']
def get_tasks(self, start=0, end=0):
@@ -316,9 +360,8 @@ class CvpApi(object):
the 'data' key contains a list of the tasks.
'''
self.log.debug('get_tasks:')
- return self.clnt.get('/task/getTasks.do?queryparam=&startIndex=%d&'
- 'endIndex=%d' % (start, end),
- timeout=self.request_timeout)
+ return self.clnt.get(f"/task/getTasks.do?queryparam=&startIndex={start}&"
+ f"endIndex={end}", timeout=self.request_timeout)
def get_logs_by_id(self, task_id, start=0, end=0):
''' Returns the log entries for the task with the specified TaskId.
@@ -334,14 +377,13 @@ class CvpApi(object):
task (dict): The CVP log for the associated Id. Returns None
if the task_id was invalid.
'''
- self.log.debug('get_logs_by_id: task_id: %s' % task_id)
+ self.log.debug(f"get_logs_by_id: task_id: {task_id}")
if self.clnt.apiversion is None:
self.get_cvp_info()
if self.clnt.apiversion < 5.0:
self.log.debug('v1 - v4 /task/getLogsByID.do?')
- resp = self.clnt.get('/task/getLogsById.do?id=%s&queryparam='
- '&startIndex=%d&endIndex=%d' %
- (task_id, start, end),
+ resp = self.clnt.get(f"/task/getLogsById.do?id={task_id}&queryparam="
+ f"&startIndex={start}&endIndex={end}",
timeout=self.request_timeout)
else:
self.log.debug('v5 /audit/getLogs.do')
@@ -350,22 +392,21 @@ class CvpApi(object):
if 'stageId' in task_info:
stage_id = task_info['stageId']
else:
- self.log.debug('No stage ID found for task %s' % task_id)
+ self.log.debug(f"No stage ID found for task {task_id}")
if 'ccIdV2' in task_info:
cc_id = task_info['ccIdV2']
if cc_id == '':
- self.log.debug('No ccIdV2 for task %s.'
- ' It was likely cancelled.'
- ' Using old /task/getLogsByID.do?'
- % task_id)
+ self.log.debug(f"No ccIdV2 for task {task_id}."
+ f" It was likely cancelled."
+ f" Using old /task/getLogsByID.do?")
resp = self.clnt.get(
- '/task/getLogsById.do?id=%s&queryparam='
- '&startIndex=%d&endIndex=%d' % (task_id, start, end),
+ f"/task/getLogsById.do?id={task_id}&queryparam="
+ f"&startIndex={start}&endIndex={end}",
timeout=self.request_timeout)
else:
resp = self.get_audit_logs_by_id(cc_id, stage_id)
else:
- self.log.debug('No change ID found for task %s' % task_id)
+ self.log.debug(f"No change ID found for task {task_id}")
resp = None
return resp
@@ -398,8 +439,7 @@ class CvpApi(object):
task_id (str): Task ID
note (str): Note to add to the task
'''
- self.log.debug('add_note_to_task: task_id: %s note: %s' %
- (task_id, note))
+ self.log.debug(f"add_note_to_task: task_id: {task_id} note: {note}")
data = {'workOrderId': task_id, 'note': note}
self.clnt.post('/task/addNoteToTask.do', data=data,
timeout=self.request_timeout)
@@ -416,10 +456,9 @@ class CvpApi(object):
Args:
task_id (str): Task ID
'''
- self.log.debug('execute_task: task_id: %s' % task_id)
+ self.log.debug(f"execute_task: task_id: {task_id}")
data = {'data': [task_id]}
- self.clnt.post('/task/executeTask.do', data=data,
- timeout=self.request_timeout)
+ self.clnt.post('/task/executeTask.do', data=data, timeout=self.request_timeout)
def cancel_task(self, task_id):
''' Cancel the task
@@ -427,7 +466,7 @@ class CvpApi(object):
Args:
task_id (str): Task ID
'''
- self.log.debug('cancel_task: task_id: %s' % task_id)
+ self.log.debug(f"cancel_task: task_id: {task_id}")
data = {'data': [task_id]}
return self.clnt.post('/task/cancelTask.do', data=data,
timeout=self.request_timeout)
@@ -442,29 +481,28 @@ class CvpApi(object):
'''
if self.clnt.apiversion is None:
self.get_cvp_info()
- configlets = self.clnt.get('/configlet/getConfiglets.do?'
- 'startIndex=%d&endIndex=%d' % (start, end),
+ configlets = self.clnt.get(f"/configlet/getConfiglets.do?"
+ f"startIndex={start}&endIndex={end}",
timeout=self.request_timeout)
if self.clnt.apiversion == 1.0 or self.clnt.apiversion >= 4.0:
self.log.debug('v1/v4+ Inventory API Call')
return configlets
- else:
- self.log.debug('v2 Inventory API Call')
- # New API getConfiglets does not return the actual configlet config
- # Get the actual configlet config using getConfigletByName
- if 'data' in configlets:
- for configlet in configlets['data']:
- full_cfglt_data = self.get_configlet_by_name(
- configlet['name'])
- configlet['config'] = full_cfglt_data['config']
- return configlets
+ self.log.debug('v2 Inventory API Call')
+ # New API getConfiglets does not return the actual configlet config
+ # Get the actual configlet config using getConfigletByName
+ if 'data' in configlets:
+ for configlet in configlets['data']:
+ full_cfglt_data = self.get_configlet_by_name(configlet['name'])
+ configlet['config'] = full_cfglt_data['config']
+ return configlets
def get_configlets_and_mappers(self):
''' Returns a list of all defined configlets and associated mappers
'''
self.log.debug(
'get_configlets_and_mappers: getConfigletsAndAssociatedMappers')
- return self.clnt.get('/configlet/getConfigletsAndAssociatedMappers.do')
+ return self.clnt.get('/configlet/getConfigletsAndAssociatedMappers.do',
+ timeout=self.request_timeout)
def get_configlet_builder(self, c_id):
''' Returns the configlet builder data for the given configlet ID.
@@ -472,8 +510,8 @@ class CvpApi(object):
Args:
c_id (str): The ID (key) for the configlet to be queried.
'''
- return self.clnt.get('/configlet/getConfigletBuilder.do?id=%s'
- % c_id, timeout=self.request_timeout)
+ return self.clnt.get(f"/configlet/getConfigletBuilder.do?id={c_id}",
+ timeout=self.request_timeout)
def search_configlets(self, query, start=0, end=0):
''' Returns a list of configlets that match a search query.
@@ -485,10 +523,9 @@ class CvpApi(object):
end (int): End index for the pagination. If end index is 0
then all the records will be returned. Default is 0.
'''
- self.log.debug('search_configlets: query: %s' % query)
- return self.clnt.get('/configlet/searchConfiglets.do?'
- 'queryparam=%s&startIndex=%d&endIndex=%d' %
- (qplus(query), start, end),
+ self.log.debug(f"search_configlets: query: {query}")
+ return self.clnt.get(f"/configlet/searchConfiglets.do?"
+ f"queryparam={qplus(query)}&startIndex={start}&endIndex={end}",
timeout=self.request_timeout)
def get_configlet_by_name(self, name):
@@ -500,9 +537,9 @@ class CvpApi(object):
Returns:
configlet (dict): The configlet dict.
'''
- self.log.debug('get_configlets_by_name: name: %s' % name)
- return self.clnt.get('/configlet/getConfigletByName.do?name=%s'
- % qplus(name), timeout=self.request_timeout)
+ self.log.debug(f"get_configlets_by_name: name: {name}")
+ return self.clnt.get(f"/configlet/getConfigletByName.do?name={qplus(name)}",
+ timeout=self.request_timeout)
def get_configlets_by_container_id(self, c_id, start=0, end=0):
''' Returns a list of configlets applied to the given container.
@@ -513,9 +550,8 @@ class CvpApi(object):
end (int): End index for the pagination. If end index is 0
then all the records will be returned. Default is 0.
'''
- return self.clnt.get('/provisioning/getConfigletsByContainerId.do?'
- 'containerId=%s&startIndex=%d&endIndex=%d'
- % (c_id, start, end),
+ return self.clnt.get(f"/provisioning/getConfigletsByContainerId.do?"
+ f"containerId={c_id}&startIndex={start}&endIndex={end}",
timeout=self.request_timeout)
def get_configlets_by_netelement_id(self, d_id, start=0, end=0):
@@ -527,9 +563,8 @@ class CvpApi(object):
end (int): End index for the pagination. If end index is 0
then all the records will be returned. Default is 0.
'''
- return self.clnt.get('/provisioning/getConfigletsByNetElementId.do?'
- 'netElementId=%s&startIndex=%d&endIndex=%d'
- % (d_id, start, end),
+ return self.clnt.get(f"/provisioning/getConfigletsByNetElementId.do?"
+ f"netElementId={d_id}&startIndex={start}&endIndex={end}",
timeout=self.request_timeout)
def get_image_bundle_by_container_id(self, container_id, start=0, end=0,
@@ -542,15 +577,13 @@ class CvpApi(object):
then all the records will be returned. Default is 0.
scope (string) the session scope (true or false).
'''
- if scope != 'true' and scope != 'false':
- self.log.error('scope value must be true or false.'
- ' %s is an invalid value.'
- ' Defaulting back to false' % scope)
+ if scope not in ('true', 'false'):
+ self.log.error("scope value must be true or false. %s is an invalid value."
+ " Defaulting back to false", scope)
scope = 'false'
- return self.clnt.get('/provisioning/getImageBundleByContainerId.do?'
- 'containerId=%s&startIndex=%d&endIndex=%d'
- '&sessionScope=%s'
- % (container_id, start, end, scope),
+ return self.clnt.get(f"/provisioning/getImageBundleByContainerId.do?"
+ f"containerId={container_id}&startIndex={start}&endIndex={end}"
+ f"&sessionScope={scope}",
timeout=self.request_timeout)
def get_configlet_history(self, key, start=0, end=0):
@@ -567,10 +600,10 @@ class CvpApi(object):
history (dict): The configlet dict with the changes from
most recent to oldest.
'''
- self.log.debug('get_configlets_history: key: %s' % key)
- return self.clnt.get('/configlet/getConfigletHistory.do?configletId='
- '%s&queryparam=&startIndex=%d&endIndex=%d' %
- (key, start, end), timeout=self.request_timeout)
+ self.log.debug(f"get_configlets_history: key: {key}")
+ return self.clnt.get(f"/configlet/getConfigletHistory.do?configletId="
+ f"{key}&queryparam=&startIndex={start}&endIndex={end}",
+ timeout=self.request_timeout)
def get_inventory(self, start=0, end=0, query='', provisioned=True):
''' Returns the a dict of the net elements known to CVP.
@@ -589,13 +622,12 @@ class CvpApi(object):
self.get_cvp_info()
if self.clnt.apiversion == 1.0:
self.log.debug('v1 Inventory API Call')
- data = self.clnt.get('/inventory/getInventory.do?'
- 'queryparam=%s&startIndex=%d&endIndex=%d' %
- (qplus(query), start, end),
+ data = self.clnt.get(f"/inventory/getInventory.do?"
+ f"queryparam={qplus(query)}&startIndex={start}&endIndex={end}",
timeout=self.request_timeout)
return data['netElementList']
self.log.debug('v2 Inventory API Call')
- data = self.clnt.get('/inventory/devices?provisioned=%s' % provisioned,
+ data = self.clnt.get(f"/inventory/devices?provisioned={provisioned}",
timeout=self.request_timeout)
containers = self.get_containers()
for dev in data:
@@ -703,8 +735,7 @@ class CvpApi(object):
# If any devices did not appear, there is a problem
# Join the missing IPs into a string for output
missing_ips = ', '.join(device_ips)
- raise RuntimeError('Devices {} failed to appear '
- 'in inventory'.format(missing_ips))
+ raise RuntimeError(f"Devices {missing_ips} failed to appear in inventory")
# Move the devices to their specified containers
for device in device_list:
@@ -733,12 +764,12 @@ class CvpApi(object):
}
self.add_devices_to_inventory([device], wait=wait)
- def retry_add_to_inventory(self, device_mac, device_ip, username,
+ def retry_add_to_inventory(self, dev_mac, device_ip, username,
password):
'''Retry addition of device to Cvp inventory
Args:
- device_mac (str): MAC address of device
+ dev_mac (str): MAC address of device
device_ip (str): ip address assigned to device
username (str): username for device login
password (str): password for user
@@ -748,7 +779,7 @@ class CvpApi(object):
self.get_cvp_info()
if self.clnt.apiversion == 1.0:
self.log.debug('v1 Inventory API Call')
- data = {"key": device_mac,
+ data = {"key": dev_mac,
"ipAddress": device_ip,
"userName": username,
"password": password}
@@ -761,11 +792,11 @@ class CvpApi(object):
self.log.warning(
'retry_add_to_inventory: not implemented for v2 APIs')
- def delete_device(self, device_mac):
+ def delete_device(self, dev_mac):
'''Delete the device and its pending tasks from Cvp inventory
Args:
- device_mac (str): mac address of device we are deleting
+ dev_mac (str): mac address of device we are deleting
For CVP 2020 this param is now required to
be the device serial number instead of MAC
address. This method will handle getting
@@ -775,13 +806,13 @@ class CvpApi(object):
data (dict): Contains success or failure message
'''
self.log.debug('delete_device: called')
- return self.delete_devices([device_mac])
+ return self.delete_devices([dev_mac])
- def delete_devices(self, device_macs):
+ def delete_devices(self, dev_macs):
'''Delete the device and its pending tasks from Cvp inventory
Args:
- device_macs (list): list of mac address for
+ dev_macs (list): list of mac address for
devices we're deleting
For CVP 2020 this param is now required to
be a list of device serial numbers instead
@@ -796,7 +827,7 @@ class CvpApi(object):
if self.clnt.apiversion is None:
self.get_cvp_info()
if self.clnt.apiversion < 4.0:
- data = {'data': device_macs}
+ data = {'data': dev_macs}
resp = self.clnt.post('/inventory/deleteDevices.do?', data=data,
timeout=self.request_timeout)
else:
@@ -806,7 +837,7 @@ class CvpApi(object):
' when deleting a device. Looking up each devices'
'serial num based on provided MAC addresses')
devices = []
- for dev_mac in device_macs:
+ for dev_mac in dev_macs:
device_info = self.get_device_by_mac(dev_mac)
if device_info is not None and 'serialNumber' in device_info:
devices.append(device_info)
@@ -902,7 +933,7 @@ class CvpApi(object):
device (dict): The net element device dict for the device if
otherwise returns an empty hash.
'''
- self.log.debug('get_device_by_name: fqdn: %s' % fqdn)
+ self.log.debug(f"get_device_by_name: fqdn: {fqdn}")
# data = self.get_inventory(start=0, end=0, query=fqdn)
data = self.search_topology(fqdn)
device = {}
@@ -918,23 +949,23 @@ class CvpApi(object):
break
return device
- def get_device_by_mac(self, device_mac):
+ def get_device_by_mac(self, dev_mac):
''' Returns the net element device dict for the devices mac address.
Args:
- device_mac (str): MAC Address of the device.
+ dev_mac (str): MAC Address of the device.
Returns:
device (dict): The net element device dict for the device if
otherwise returns an empty hash.
'''
- self.log.debug('get_device_by_mac: MAC address: %s' % device_mac)
- # data = self.get_inventory(start=0, end=0, query=device_mac)
- data = self.search_topology(device_mac)
+ self.log.debug(f"get_device_by_mac: MAC address: {dev_mac}")
+ # data = self.get_inventory(start=0, end=0, query=dev_mac)
+ data = self.search_topology(dev_mac)
device = {}
if 'netElementList' in data:
for netelem in data['netElementList']:
- if netelem['systemMacAddress'] == device_mac:
+ if netelem['systemMacAddress'] == dev_mac:
device = netelem
break
return device
@@ -949,8 +980,7 @@ class CvpApi(object):
device (dict): The net element device dict for the device if
otherwise returns an empty hash.
'''
- self.log.debug('get_device_by_serial: Serial Number: %s'
- % device_serial)
+ self.log.debug(f"get_device_by_serial: Serial Number: {device_serial}")
data = self.search_topology(device_serial)
device = {}
if 'netElementList' in data:
@@ -960,53 +990,51 @@ class CvpApi(object):
break
return device
- def get_device_configuration(self, device_mac):
+ def get_device_configuration(self, dev_mac):
''' Returns the running configuration for the device provided.
Args:
- device_mac (str): Mac address of the device to get the running
+ dev_mac (str): Mac address of the device to get the running
configuration for.
Returns:
device (dict): The net element device dict for the device if
otherwise returns an empty hash.
'''
- self.log.debug('get_device_configuration: device_mac: %s' % device_mac)
+ self.log.debug(f"get_device_configuration: dev_mac: {dev_mac}")
if self.clnt.apiversion is None:
self.get_cvp_info()
if self.clnt.apiversion < 4.0:
- data = self.clnt.get('/inventory/getInventoryConfiguration.do?'
- 'netElementId=%s' % device_mac,
+ data = self.clnt.get(f"/inventory/getInventoryConfiguration.do?netElementId={dev_mac}",
timeout=self.request_timeout)
else:
- data = self.clnt.get('/inventory/device/config?'
- 'netElementId=%s' % device_mac,
+ data = self.clnt.get(f"/inventory/device/config?netElementId={dev_mac}",
timeout=self.request_timeout)
running_config = ''
if 'output' in data:
running_config = data['output']
return running_config
- def get_device_image_info(self, device_mac):
+ def get_device_image_info(self, dev_mac):
''' Return a dict of info about a device in CVP.
Args:
- device_mac (str): Mac address of the device to get the running
+ dev_mac (str): Mac address of the device to get the running
configuration for.
Returns:
device_image_info (dict): Dict of image info for the device
if found. Otherwise returns None.
'''
- self.log.debug('Attempt to get net element data for %s' % device_mac)
+ self.log.debug(f"Attempt to get net element data for {dev_mac}")
try:
device_image_info = self.clnt.get(
- '/provisioning/getNetElementInfoById.do?netElementId=%s'
- % qplus(device_mac), timeout=self.request_timeout)
+ f"/provisioning/getNetElementInfoById.do?netElementId={qplus(dev_mac)}",
+ timeout=self.request_timeout)
except CvpApiError as error:
# Catch error when device for provided MAC is not found
if 'Invalid Netelement id' in str(error):
- self.log.debug('Device with MAC %s not found' % device_mac)
+ self.log.debug(f"Device with MAC {dev_mac} not found")
return None
raise error
return device_image_info
@@ -1029,8 +1057,8 @@ class CvpApi(object):
self.get_cvp_info()
if self.clnt.apiversion == 1.0:
self.log.debug('v1 Inventory API Call')
- return self.clnt.get('/inventory/add/searchContainers.do?'
- 'startIndex=%d&endIndex=%d' % (start, end))
+ return self.clnt.get(f"/inventory/add/searchContainers.do?"
+ f"startIndex={start}&endIndex={end}")
self.log.debug('v2 Inventory API Call')
containers = self.clnt.get('/inventory/containers')
for container in containers:
@@ -1046,9 +1074,8 @@ class CvpApi(object):
container['parentId'] = cont['Key']
break
else:
- self.log.debug(
- 'No container parentId found for parentName %s',
- full_cont_info['parentName'])
+ self.log.debug(f"No container parentId found for"
+ f" parentName {full_cont_info['parentName']}")
container['parentId'] = None
else:
container['parentName'] = None
@@ -1069,9 +1096,9 @@ class CvpApi(object):
Returns:
container (dict): Container info in dictionary format or None
'''
- self.log.debug('Get info for container %s' % name)
- conts = self.clnt.get('/provisioning/searchTopology.do?queryParam=%s'
- '&startIndex=0&endIndex=0' % qplus(name))
+ self.log.debug(f"Get info for container {name}")
+ conts = self.clnt.get(f"/provisioning/searchTopology.do?queryParam={qplus(name)}"
+ f"&startIndex=0&endIndex=0")
if conts['total'] > 0 and conts['containerList']:
for cont in conts['containerList']:
if cont['name'] == name:
@@ -1087,9 +1114,9 @@ class CvpApi(object):
Returns:
container (dict): Container info in dictionary format or None
'''
- self.log.debug('Get info for container %s' % key)
- return self.clnt.get('/provisioning/getContainerInfoById.do?'
- 'containerId=%s' % qplus(key))
+ self.log.debug(f"Get info for container {key}")
+ return self.clnt.get(f"/provisioning/getContainerInfoById.do?"
+ f"containerId={qplus(key)}")
def get_configlets_by_device_id(self, mac, start=0, end=0):
''' Returns the list of configlets applied to a device.
@@ -1104,7 +1131,7 @@ class CvpApi(object):
Returns:
configlets (list): The list of configlets applied to the device
'''
- self.log.debug('get_configlets_by_device: mac: %s' % mac)
+ self.log.debug(f"get_configlets_by_device: mac: {mac}")
data = self.get_configlets_by_netelement_id(mac, start, end)
return data['configletList']
@@ -1142,19 +1169,17 @@ class CvpApi(object):
if not form:
form = []
- self.log.debug('add_configlet_builder: name: %s config: %s form: %s'
- % (name, config, form))
+ self.log.debug(f"add_configlet_builder: name: {name} config: {config} form: {form}")
data = {'name': name,
'data': {'formList': form,
'main_script': {'data': config}}}
# Create the configlet builder
- self.clnt.post('/configlet/addConfigletBuilder.do?isDraft=%s' % draft,
+ self.clnt.post(f"/configlet/addConfigletBuilder.do?isDraft={draft}",
data=data, timeout=self.request_timeout)
# Get the key for the configlet
- data = self.clnt.get(
- '/configlet/getConfigletByName.do?name=%s' % qplus(name),
- timeout=self.request_timeout)
+ data = self.clnt.get(f"/configlet/getConfigletByName.do?name={qplus(name)}",
+ timeout=self.request_timeout)
return data['key']
def add_configlet(self, name, config):
@@ -1167,18 +1192,17 @@ class CvpApi(object):
Returns:
key (str): The key for the configlet
'''
- self.log.debug('add_configlet: name: %s config: %s' % (name, config))
+ self.log.debug(f"add_configlet: name: {name} config: {config}")
body = {'name': name, 'config': config}
# Create the configlet
self.clnt.post('/configlet/addConfiglet.do', data=body,
timeout=self.request_timeout)
# Get the key for the configlet
- data = self.clnt.get('/configlet/getConfigletByName.do?name=%s'
- % qplus(name), timeout=self.request_timeout)
+ data = self.clnt.get(f"/configlet/getConfigletByName.do?name={qplus(name)}",
+ timeout=self.request_timeout)
return data['key']
-
def delete_configlet(self, name, key):
''' Delete the configlet.
@@ -1186,7 +1210,7 @@ class CvpApi(object):
name (str): Configlet name
key (str): Configlet key
'''
- self.log.debug('delete_configlet: name: %s key: %s' % (name, key))
+ self.log.debug(f"delete_configlet: name: {name} key: {key}")
body = [{'name': name, 'key': key}]
# Delete the configlet
self.clnt.post('/configlet/deleteConfiglet.do', data=body,
@@ -1204,8 +1228,7 @@ class CvpApi(object):
Returns:
data (dict): Contains success or failure message
'''
- self.log.debug('update_configlet: config: %s key: %s name: %s' %
- (config, key, name))
+ self.log.debug(f"update_configlet: config: {config} key: {key} name: {name}")
# Update the configlet
body = {'config': config, 'key': key, 'name': name,
@@ -1256,21 +1279,18 @@ class CvpApi(object):
}
}
}
- debug_str = 'update_configlet_builder:' \
- ' config: {} key: {} name: {} form: {}'
- self.log.debug(debug_str.format(config, key, name, form))
+ self.log.debug(f"update_configlet_builder: config: {config}"
+ f" key: {key} name: {name} form: {form}")
# Update the configlet builder
- url_string = '/configlet/updateConfigletBuilder.do?' \
- 'isDraft={}&id={}&action=save'
- return self.clnt.post(url_string.format(draft, key),
- data=data, timeout=self.request_timeout)
+ url_string = f"/configlet/updateConfigletBuilder.do?isDraft={draft}&id={key}&action=save"
+ return self.clnt.post(url_string, data=data, timeout=self.request_timeout)
- def update_reconcile_configlet(self, device_mac, config, key, name,
+ def update_reconcile_configlet(self, dev_mac, config, key, name,
reconciled=False):
''' Update the reconcile configlet.
Args:
- device_mac (str): Mac address/Key for device whose reconcile
+ dev_mac (str): Mac address/Key for device whose reconcile
configlet is being updated
config (str): Reconciled config statements
key (str): Reconcile Configlet key
@@ -1280,12 +1300,10 @@ class CvpApi(object):
Returns:
data (dict): Contains success or failure message
'''
- log_str = ('update_reconcile_configlet:'
- ' device_mac: {} config: {} key: {} name: {}')
- self.log.debug(log_str.format(device_mac, config, key, name))
+ self.log.debug(f"update_reconcile_configlet: dev_mac: {dev_mac}"
+ f" config: {config} key: {key} name: {name}")
- url_str = ('/provisioning/updateReconcileConfiglet.do?'
- 'netElementId={}')
+ url_str = f"/provisioning/updateReconcileConfiglet.do?netElementId={dev_mac}"
body = {
'config': config,
'key': key,
@@ -1293,8 +1311,7 @@ class CvpApi(object):
'reconciled': reconciled,
'unCheckedLines': '',
}
- return self.clnt.post(url_str.format(device_mac), data=body,
- timeout=self.request_timeout)
+ return self.clnt.post(url_str, data=body, timeout=self.request_timeout)
def add_note_to_configlet(self, key, note):
''' Add a note to a configlet.
@@ -1310,66 +1327,20 @@ class CvpApi(object):
return self.clnt.post('/configlet/addNoteToConfiglet.do',
data=data, timeout=self.request_timeout)
- def sanitize_warnings(self, data):
- ''' Sanitize the warnings returned after validation.
-
- In some cases where the configlets has both errors
- and warnings, CVP may split any warnings that have
- `,` across multiple strings.
- This method concats the strings back into one string
- per warning, and correct the warningCount.
-
- Args:
- data (dict): A dict that contians the result
- of the validation operation
- Returns:
- response (dict): A dict that contains the result of the
- validation operation
- '''
- if "warnings" not in data:
- # nothing to do here, we can return as is
- return data
- # Since there may be warnings incorrectly split on
- # ', ' within the warning text by CVP, we join all the
- # warnings together using ', ' into one large string
- temp_warnings = ", ".join(data['warnings']).strip()
-
- # To split the large string again we match on the
- # 'at line XXX' that should indicate the end of the warning.
- # We capture as well the remaining \\n or whitespace and include
- # the extra ', ' added in the previous step in the matching criteria.
- # The extra ', ' is not included in the strings of the new list
- temp_warnings = split(
- r'(.*?at line \d+.*?),\s+',
- temp_warnings
- )
-
- # The behaviour of re.split will add empty strings
- # if the regex matches on the begging or ending of the line.
- # Refer to https://docs.python.org/3/library/re.html#re.split
-
- # Use filter to remove any empty strings
- # that re.split inserted
- data['warnings'] = list(filter(None, temp_warnings))
- # Update the count of warnings to the correct value
- data['warningCount'] = len(data['warnings'])
- return data
-
- def validate_config_for_device(self, device_mac, config):
+ def validate_config_for_device(self, dev_mac, config):
''' Validate a config against a device
Args:
- device_mac (str): Device MAC address
+ dev_mac (str): Device MAC address
config (str): Switch config statements
Returns:
response (dict): A dict that contains the result of the
validation operation
'''
- self.log.debug('validate_config_for_device: device_mac: %s config: %s'
- % (device_mac, config))
- body = {'netElementId': device_mac, 'config': config}
- return self.sanitize_warnings(
+ self.log.debug(f"validate_config_for_device: dev_mac: {dev_mac} config: {config}")
+ body = {'netElementId': dev_mac, 'config': config}
+ return sanitize_warnings(
self.clnt.post(
'/configlet/validateConfig.do',
data=body,
@@ -1377,40 +1348,35 @@ class CvpApi(object):
)
)
- def validate_config(self, device_mac, config):
+ def validate_config(self, dev_mac, config):
''' Validate a config against a device and parse response to
produce log messages are return a flag for the config validity.
Args:
- device_mac (str): Device MAC address
+ dev_mac (str): Device MAC address
config (str): Switch config statements
Returns:
response (boolean): A flag signifying if the config is valid or
not.
'''
- self.log.debug('validate_config: device_mac: %s config: %s'
- % (device_mac, config))
- result = self.validate_config_for_device(device_mac, config)
+ self.log.debug(f"validate_config: dev_mac: {dev_mac} config: {config}")
+ result = self.validate_config_for_device(dev_mac, config)
validated = True
if 'warningCount' in result and result['warnings']:
for warning in result['warnings']:
- self.log.warning('Validation of config produced warning - %s'
- % warning)
+ self.log.warning(f"Validation of config produced warning - {warning}")
if 'errorCount' in result:
- self.log.error('Validation of config produced %s errors'
- % result['errorCount'])
+ self.log.error(f"Validation of config produced {result['errorCount']} errors")
if 'errors' in result:
for error in result['errors']:
- self.log.error('Validation of config produced error - %s'
- % error)
+ self.log.error(f"Validation of config produced error - {error}")
validated = False
if 'result' in result:
for item in result['result']:
if 'messages' in item:
for message in item['messages']:
- self.log.info('Validation of config returned'
- ' message - %s' % message)
+ self.log.info(f"Validation of config returned message - {message}")
return validated
def get_all_temp_actions(self, start=0, end=0):
@@ -1425,8 +1391,7 @@ class CvpApi(object):
response (dict): A dict that contains a list of the current
temp actions.
'''
- url = ('/provisioning/getAllTempActions.do?startIndex=%d&endIndex=%d'
- % (start, end))
+ url = (f"/provisioning/getAllTempActions.do?startIndex={start}&endIndex={end}")
data = self.clnt.get(url, timeout=self.request_timeout)
return data
@@ -1462,7 +1427,7 @@ class CvpApi(object):
url = '/provisioning/v2/saveTopology.do'
return self.clnt.post(url, data=data, timeout=self.request_timeout)
- def apply_configlets_to_device(self, app_name, dev, new_configlets,
+ def apply_configlets_to_device(self, app_name, dev, new_configlets, # pylint: disable=too-many-locals
create_task=True, reorder_configlets=False, validate=False):
''' Apply the configlets to the device.
@@ -1497,8 +1462,7 @@ class CvpApi(object):
Ex: {u'data': {u'status': u'success', u'taskIds': [u'32']}}
'''
- self.log.debug('apply_configlets_to_device: dev: %s names: %s' %
- (dev, new_configlets))
+ self.log.debug(f"apply_configlets_to_device: dev: {dev} names: {new_configlets}")
# Get a list of the names and keys of the configlets
cnames = []
ckeys = []
@@ -1516,7 +1480,7 @@ class CvpApi(object):
cnames.append(entry['name'])
ckeys.append(entry['key'])
- info = '%s: Configlet Assign: to Device %s' % (app_name, dev['fqdn'])
+ info = f"{app_name}: Configlet Assign: to Device {dev['fqdn']}"
info_preview = '<b>Configlet Assign:</b> to Device' + dev['fqdn']
data = {'data': [{'info': info,
'infoPreview': info_preview,
@@ -1544,16 +1508,17 @@ class CvpApi(object):
'parentTask': ''}]}
if validate:
validation_result = self.validate_configlets_for_device(dev['systemMacAddress'], ckeys)
- data['data'][0].update({
- "configCompareCount": {
- "mismatch": validation_result['mismatch'],
- "reconcile": validation_result['reconcile'],
- "new": validation_result['new']
+ data['data'][0].update(
+ {
+ "configCompareCount":
+ {
+ "mismatch": validation_result['mismatch'],
+ "reconcile": validation_result['reconcile'],
+ "new": validation_result['new']
}
}
)
- self.log.debug('apply_configlets_to_device: saveTopology data:\n%s' %
- data['data'])
+ self.log.debug(f"apply_configlets_to_device: saveTopology data:\n{data['data']}")
self._add_temp_action(data)
if create_task:
return self._save_topology_v2([])
@@ -1583,8 +1548,7 @@ class CvpApi(object):
Ex: {u'data': {u'status': u'success', u'taskIds': [u'35']}}
'''
- self.log.debug('remove_configlets_from_device: dev: %s names: %s' %
- (dev, del_configlets))
+ self.log.debug(f"remove_configlets_from_device: dev: {dev} names: {del_configlets}")
# Get all the configlets assigned to the device.
configlets = self.get_configlets_by_device_id(dev['systemMacAddress'])
@@ -1608,7 +1572,7 @@ class CvpApi(object):
del_names.append(entry['name'])
del_keys.append(entry['key'])
- info = '%s Configlet Remove: from Device %s' % (app_name, dev['fqdn'])
+ info = f"{app_name} Configlet Remove: from Device {dev['fqdn']}"
info_preview = '<b>Configlet Remove:</b> from Device' + dev['fqdn']
data = {'data': [{'info': info,
'infoPreview': info_preview,
@@ -1635,17 +1599,18 @@ class CvpApi(object):
'childTasks': [],
'parentTask': ''}]}
if validate:
- validation_result = self.validate_configlets_for_device(dev['systemMacAddress'], keep_keys)
- data['data'][0].update({
- "configCompareCount": {
- "mismatch": validation_result['mismatch'],
- "reconcile": validation_result['reconcile'],
- "new": validation_result['new']
+ validation_result = self.validate_configlets_for_device(dev['systemMacAddress'],
+ keep_keys)
+ data['data'][0].update(
+ {
+ "configCompareCount": {
+ "mismatch": validation_result['mismatch'],
+ "reconcile": validation_result['reconcile'],
+ "new": validation_result['new']
}
}
)
- self.log.debug('remove_configlets_from_device: saveTopology data:\n%s'
- % data['data'])
+ self.log.debug(f"remove_configlets_from_device: saveTopology data:\n{data['data']}")
self._add_temp_action(data)
if create_task:
return self._save_topology_v2([])
@@ -1668,9 +1633,8 @@ class CvpApi(object):
Ex: {u'data': {u'status': u'success', u'taskIds': [u'32']}}
'''
- self.log.debug(
- 'apply_configlets_to_container: container: %s names: %s' %
- (container, new_configlets))
+ self.log.debug(f"apply_configlets_to_container: container: {container}"
+ f" names: {new_configlets}")
# Get all the configlets assigned to the device.
configlets = self.get_configlets_by_container_id(container['key'])
@@ -1695,8 +1659,7 @@ class CvpApi(object):
cnames.append(entry['name'])
ckeys.append(entry['key'])
- info = '%s: Configlet Assign: to Container %s' % (app_name,
- container['name'])
+ info = f"{app_name}: Configlet Assign: to Container {container['name']}"
info_preview = '<b>Configlet Assign:</b> to Container' + container[
'name']
data = {'data': [{'info': info,
@@ -1723,9 +1686,7 @@ class CvpApi(object):
'nodeTargetIpAddress': '',
'childTasks': [],
'parentTask': ''}]}
- self.log.debug(
- 'apply_configlets_to_container: saveTopology data:\n%s' %
- data['data'])
+ self.log.debug(f"apply_configlets_to_container: saveTopology data:\n{data['data']}")
self._add_temp_action(data)
if create_task:
return self._save_topology_v2([])
@@ -1750,9 +1711,8 @@ class CvpApi(object):
Ex: {u'data': {u'status': u'success', u'taskIds': [u'35']}}
'''
- self.log.debug(
- 'remove_configlets_from_container: container: %s names: %s' %
- (container, del_configlets))
+ self.log.debug(f"remove_configlets_from_container: container: {container}"
+ f" names: {del_configlets}")
# Get all the configlets assigned to the device.
configlets = self.get_configlets_by_container_id(container['key'])
@@ -1776,8 +1736,7 @@ class CvpApi(object):
del_names.append(entry['name'])
del_keys.append(entry['key'])
- info = '%s Configlet Remove: from Container %s' % (app_name,
- container['name'])
+ info = f"{app_name} Configlet Remove: from Container {container['name']}"
info_preview = '<b>Configlet Remove:</b> from Container' + container[
'name']
data = {'data': [{'info': info,
@@ -1804,9 +1763,7 @@ class CvpApi(object):
'nodeTargetIpAddress': '',
'childTasks': [],
'parentTask': ''}]}
- self.log.debug(
- 'remove_configlets_from_container: saveTopology data:\n%s'
- % data['data'])
+ self.log.debug(f"remove_configlets_from_container: saveTopology data:\n{data['data']}")
self._add_temp_action(data)
if create_task:
return self._save_topology_v2([])
@@ -1839,9 +1796,8 @@ class CvpApi(object):
"configletId": "string"}, ...]
}
'''
- self.log.debug('validate_configlets_for_device: '
- 'MAC: %s - conf keys: %s - page_type: %s' %
- (mac, configlet_keys, page_type))
+ self.log.debug(f"validate_configlets_for_device: "
+ f"MAC: {mac} - conf keys: {configlet_keys} - page_type: {page_type}")
data = {'configIdList': configlet_keys,
'netElementId': mac,
'pageType': page_type}
@@ -1858,9 +1814,8 @@ class CvpApi(object):
end (int): End index for the pagination. If end index is 0
then all the records will be returned. Default is 0.
'''
- return self.clnt.get('/configlet/getAppliedDevices.do?'
- 'configletName=%s&startIndex=%d&endIndex=%d'
- % (configlet_name, start, end),
+ return self.clnt.get(f"/configlet/getAppliedDevices.do?"
+ f"configletName={configlet_name}&startIndex={start}&endIndex={end}",
timeout=self.request_timeout)
def get_applied_containers(self, configlet_name, start=0, end=0):
@@ -1873,9 +1828,8 @@ class CvpApi(object):
end (int): End index for the pagination. If end index is 0
then all the records will be returned. Default is 0.
'''
- return self.clnt.get('/configlet/getAppliedContainers.do?'
- 'configletName=%s&startIndex=%d&endIndex=%d'
- % (configlet_name, start, end),
+ return self.clnt.get(f"/configlet/getAppliedContainers.do?"
+ f"configletName={configlet_name}&startIndex={start}&endIndex={end}",
timeout=self.request_timeout)
# pylint: disable=too-many-arguments
@@ -1896,8 +1850,7 @@ class CvpApi(object):
Ex: {u'data': {u'status': u'success', u'taskIds': []}}
'''
- msg = ('%s container %s under container %s' %
- (operation, container_name, parent_name))
+ msg = (f"{operation} container {container_name} under container {parent_name}")
data = {'data': [{'info': msg,
'infoPreview': msg,
'action': operation,
@@ -1936,8 +1889,8 @@ class CvpApi(object):
Ex: {u'data': {u'status': u'success', u'taskIds': []}}
'''
- self.log.debug('add_container: container: %s parent: %s parent_key: %s'
- % (container_name, parent_name, parent_key))
+ self.log.debug(f"add_container: container: {container_name}"
+ f" parent: {parent_name} parent_key: {parent_key}")
return self._container_op(container_name, 'new_container', parent_name,
parent_key, 'add')
@@ -1957,10 +1910,9 @@ class CvpApi(object):
Ex: {u'data': {u'status': u'success', u'taskIds': []}}
'''
- self.log.debug('delete_container: container: %s container_key: %s '
- 'parent: %s parent_key: %s' %
- (container_name, container_key, parent_name,
- parent_key))
+ self.log.debug(f"delete_container: container: {container_name}"
+ f" container_key: {container_key} parent: {parent_name}"
+ f" parent_key: {parent_key}")
resp = self._container_op(container_name, container_key, parent_name,
parent_key, 'delete')
# As of CVP version 2020.1 the addTempAction.do API endpoint stopped
@@ -1973,26 +1925,24 @@ class CvpApi(object):
except CvpApiError as error:
if 'Invalid Container id' in error.msg:
return resp
- else:
- raise
+ raise
if still_exists is not None:
raise CvpApiError('Container was not deleted. Check for children')
return resp
- def get_parent_container_for_device(self, device_mac):
+ def get_parent_container_for_device(self, dev_mac):
''' Add the container to the specified parent.
Args:
- device_mac (str): Device mac address
+ dev_mac (str): Device mac address
Returns:
response (dict): A dict that contains the parent container info
'''
- self.log.debug('get_parent_container_for_device: called for %s'
- % device_mac)
- data = self.clnt.get('/provisioning/searchTopology.do?'
- 'queryParam=%s&startIndex=0&endIndex=0'
- % device_mac, timeout=self.request_timeout)
+ self.log.debug(f"get_parent_container_for_device: called for {dev_mac}")
+ data = self.clnt.get(f"/provisioning/searchTopology.do?"
+ f"queryParam={dev_mac}&startIndex=0&endIndex=0",
+ timeout=self.request_timeout)
if data['total'] > 0:
cont_name = data['netElementContainerList'][0]['containerName']
return self.get_container_by_name(cont_name)
@@ -2015,11 +1965,9 @@ class CvpApi(object):
Ex: {u'data': {u'status': u'success', u'taskIds': []}}
'''
- info = 'Device Add {} to container {} by {}'.format(device['fqdn'],
- container['name'],
- app_name)
- self.log.debug('Attempting to move device %s to container %s'
- % (device['fqdn'], container['name']))
+ info = f"Device Add {device['fqdn']} to container {container['name']} by {app_name}"
+ self.log.debug(f"Attempting to move device {device['fqdn']} to"
+ f" container {container['name']}")
if 'parentContainerId' in device:
from_id = device['parentContainerId']
else:
@@ -2042,8 +1990,7 @@ class CvpApi(object):
# pylint: disable=invalid-name
except CvpApiError as e:
if 'Data already exists' in str(e):
- self.log.debug('Device %s already in container %s'
- % (device['fqdn'], container))
+ self.log.debug(f"Device {device['fqdn']} already in container {container}")
if create_task:
return self._save_topology_v2([])
return None
@@ -2062,20 +2009,35 @@ class CvpApi(object):
response (dict): A dict that contains the container and
netelement lists.
'''
- self.log.debug('search_topology: query: %s start: %d end: %d' %
- (query, start, end))
- data = self.clnt.get('/provisioning/searchTopology.do?queryParam=%s&'
- 'startIndex=%d&endIndex=%d'
- % (qplus(query), start, end),
- timeout=self.request_timeout)
+ self.log.debug(f"search_topology: query: {query} start: {start} end: {end}")
+ if self.clnt.apiversion is None:
+ self.get_cvp_info()
+ if self.clnt.apiversion <= 6.0:
+ # Original search topology endpoint
+ req_url = (f"/provisioning/searchTopology.do?queryParam={qplus(query)}&"
+ f"startIndex={start}&endIndex={end}")
+ else:
+ # Newer CVP versions should use the V3 version of search topology endpoint
+ req_url = (f"/provisioning/v3/searchTopology.do?queryParam={qplus(query)}&"
+ f"startIndex={start}&endIndex={end}")
+
+ data = self.clnt.get(req_url, timeout=self.request_timeout)
if 'netElementList' in data:
for device in data['netElementList']:
device['status'] = device['deviceStatus']
- device['mlagEnabled'] = device['isMLAGEnabled']
- device['danzEnabled'] = device['isDANZEnabled']
device['parentContainerKey'] = device['parentContainerId']
- device['bootupTimestamp'] = device['bootupTimeStamp']
- device['internalBuild'] = device['internalBuildId']
+ if 'isMLAGEnabled' in device:
+ # original key was mlagEnabled but it changed to isMLAGEnabled
+ device['mlagEnabled'] = device['isMLAGEnabled']
+ elif 'mlagEnabled' in device:
+ # Key for V3 search topology changes back to mlagEnabled again
+ device['isMLAGEnabled'] = device['mlagEnabled']
+ # Key isDANZEnabled for V3 search topology is no longer in return data.
+ device['danzEnabled'] = device.get('isDANZEnabled', '')
+ # Key bootupTimeStamp for V3 search topology is no longer in return data.
+ device['bootupTimestamp'] = device.get('bootupTimeStamp', '')
+ # Key internalBuildId for V3 search topology is no longer in return data.
+ device['internalBuild'] = device.get('internalBuildId', '')
return data
def filter_topology(self, node_id='root', fmt='topology',
@@ -2091,9 +2053,8 @@ class CvpApi(object):
end (int): End index for the pagination. If end index is 0
then all the records will be returned. Default is 0.
'''
- url = ('/provisioning/filterTopology.do?nodeId=%s&'
- 'format=%s&startIndex=%d&endIndex=%d'
- % (node_id, fmt, start, end))
+ url = (f"/provisioning/filterTopology.do?nodeId={node_id}&"
+ f"format={fmt}&startIndex={start}&endIndex={end}")
return self.clnt.get(url, timeout=self.request_timeout)
def check_compliance(self, node_key, node_type):
@@ -2110,15 +2071,14 @@ class CvpApi(object):
response (dict): A dict that contains the results of the
compliance check.
'''
- self.log.debug('check_compliance: node_key: %s node_type: %s' %
- (node_key, node_type))
+ self.log.debug(f"check_compliance: node_key: {node_key} node_type: {node_type}")
data = {'nodeId': node_key, 'nodeType': node_type}
resp = self.clnt.post('/provisioning/checkCompliance.do', data=data,
timeout=self.request_timeout)
if self.clnt.apiversion is None:
self.get_cvp_info()
if self.clnt.apiversion >= 2.0:
- if resp['complianceIndication'] == u'':
+ if resp['complianceIndication'] == '':
resp['complianceIndication'] = 'NONE'
return resp
@@ -2128,7 +2088,7 @@ class CvpApi(object):
Args:
e_id (str): The event id to be queried.
'''
- return self.clnt.get('/event/getEventById.do?eventId=%s' % e_id,
+ return self.clnt.get(f"/event/getEventById.do?eventId={e_id}",
timeout=self.request_timeout)
def get_default_snapshot_template(self):
@@ -2183,9 +2143,9 @@ class CvpApi(object):
'''
# Get the absolute file path to be uploaded
image_path = os.path.abspath(filepath)
- image_data = open(image_path, 'rb')
- response = self.clnt.post('/image/addImage.do',
- files={'file': image_data})
+ with open(image_path, 'rb') as image_data:
+ response = self.clnt.post('/image/addImage.do',
+ files={'file': image_data})
return response
def cancel_image(self, image_name):
@@ -2214,9 +2174,8 @@ class CvpApi(object):
the 'data' key contains a list of images and their info.
'''
self.log.debug('Get info about images')
- return self.clnt.get('/image/getImages.do?queryparam=&startIndex=%d&'
- 'endIndex=%d' % (start, end),
- timeout=self.request_timeout)
+ return self.clnt.get(f"/image/getImages.do?queryparam=&startIndex={start}&"
+ f"endIndex={end}", timeout=self.request_timeout)
def get_image_bundles(self, start=0, end=0):
''' Return a list of all image bundles.
@@ -2233,8 +2192,8 @@ class CvpApi(object):
'''
self.log.debug('Get image bundles that can be applied to devices or'
' containers')
- return self.clnt.get('/image/getImageBundles.do?queryparam=&'
- 'startIndex=%d&endIndex=%d' % (start, end),
+ return self.clnt.get(f"/image/getImageBundles.do?queryparam=&"
+ f"startIndex={start}&endIndex={end}",
timeout=self.request_timeout)
def get_image_bundle_by_name(self, name):
@@ -2247,14 +2206,14 @@ class CvpApi(object):
image bundle (dict): Dict of info specific to the image bundle
requested or None if the name requested doesn't exist.
'''
- self.log.debug('Attempt to get image bundle %s' % name)
+ self.log.debug(f"Attempt to get image bundle {name}")
try:
- image = self.clnt.get('/image/getImageBundleByName.do?name=%s'
- % qplus(name), timeout=self.request_timeout)
+ image = self.clnt.get(f"/image/getImageBundleByName.do?name={qplus(name)}",
+ timeout=self.request_timeout)
except CvpApiError as error:
# Catch an invalid task_id error and return None
if 'Entity does not exist' in str(error):
- self.log.debug('Bundle with name %s does not exist' % name)
+ self.log.debug(f"Bundle with name {name} does not exist")
return None
raise error
return image
@@ -2362,24 +2321,22 @@ class CvpApi(object):
Ex: {u'data': {u'status': u'success', u'taskIds': [u'32']}}
'''
- self.log.debug('Attempt to apply %s to %s %s' % (image['name'],
- id_type, name))
- info = 'Apply image: %s to %s %s' % (image['name'], id_type, name)
+ self.log.debug(f"Attempt to apply {image['name']} to {id_type} {name}")
+ info = f"Apply image: {image['name']} to {id_type} {name}"
node_id = ''
if 'imageBundleKeys' in image:
if image['imageBundleKeys']:
node_id = image['imageBundleKeys'][0]
- self.log.info('Provided image is an image object.'
- ' Using first value from imageBundleKeys - %s'
- % node_id)
+ self.log.info(f"Provided image is an image object."
+ f" Using first value from imageBundleKeys - {node_id}")
if 'id' in image:
node_id = image['id']
- self.log.info('Provided image is an image bundle object.'
- ' Found v1 API id field - %s' % node_id)
+ self.log.info(f"Provided image is an image bundle object."
+ f" Found v1 API id field - {node_id}")
elif 'key' in image:
node_id = image['key']
- self.log.info('Provided image is an image bundle object.'
- ' Found v2 API key field - %s' % node_id)
+ self.log.info(f"Provided image is an image bundle object."
+ f" Found v2 API key field - {node_id}")
data = {'data': [{'info': info,
'infoPreview': info,
'note': '',
@@ -2446,23 +2403,22 @@ class CvpApi(object):
Ex: {u'data': {u'status': u'success', u'taskIds': [u'32']}}
'''
- self.log.debug('Attempt to remove %s from %s' % (image['name'], name))
- info = 'Remove image: %s from %s' % (image['name'], name)
+ self.log.debug(f"Attempt to remove {image['name']} from {name}")
+ info = f"Remove image: {image['name']} from {name}"
node_id = ''
if 'imageBundleKeys' in image:
if image['imageBundleKeys']:
node_id = image['imageBundleKeys'][0]
- self.log.info('Provided image is an image object.'
- ' Using first value from imageBundleKeys - %s'
- % node_id)
+ self.log.info(f"Provided image is an image object."
+ f" Using first value from imageBundleKeys - {node_id}")
if 'id' in image:
node_id = image['id']
- self.log.info('Provided image is an image bundle object.'
- ' Found v1 API id field - %s' % node_id)
+ self.log.info(f"Provided image is an image bundle object."
+ f" Found v1 API id field - {node_id}")
elif 'key' in image:
node_id = image['key']
- self.log.info('Provided image is an image bundle object.'
- ' Found v2 API key field - %s' % node_id)
+ self.log.info(f"Provided image is an image bundle object."
+ f" Found v2 API key field - {node_id}")
data = {'data': [{'info': info,
'infoPreview': info,
'note': '',
@@ -2494,7 +2450,7 @@ class CvpApi(object):
Returns:
change controls (list): The list of change controls
'''
- self.log.debug('get_change_controls: query: %s' % query)
+ self.log.debug(f"get_change_controls: query: {query}")
if self.clnt.apiversion is None:
self.get_cvp_info()
if self.clnt.apiversion >= 3.0:
@@ -2505,8 +2461,8 @@ class CvpApi(object):
self.log.debug('v2 getChangeControls API Call')
data = self.clnt.get(
- '/changeControl/getChangeControls.do?searchText=%s'
- '&startIndex=%d&endIndex=%d' % (qplus(query), start, end),
+ f"/changeControl/getChangeControls.do?searchText={qplus(query)}"
+ f"&startIndex={start}&endIndex={end}",
timeout=self.request_timeout)
if 'data' not in data:
return None
@@ -2524,7 +2480,7 @@ class CvpApi(object):
Returns:
tasks (list): The list of available tasks
'''
- self.log.debug('change_control_available_tasks: query: %s' % query)
+ self.log.debug(f"change_control_available_tasks: query: {query}")
if self.clnt.apiversion is None:
self.get_cvp_info()
if self.clnt.apiversion >= 3.0:
@@ -2534,8 +2490,8 @@ class CvpApi(object):
self.log.debug('v2 getTasksByStatus API Call')
data = self.clnt.get(
- '/changeControl/getTasksByStatus.do?searchText=%s'
- '&startIndex=%d&endIndex=%d' % (qplus(query), start, end),
+ f"/changeControl/getTasksByStatus.do?searchText={qplus(query)}"
+ f"&startIndex={start}&endIndex={end}",
timeout=self.request_timeout)
if 'data' not in data:
return None
@@ -2644,9 +2600,8 @@ class CvpApi(object):
if self.clnt.apiversion is None:
self.get_cvp_info()
if self.clnt.apiversion < 3.0:
- self.log.debug('Wrong method for API version %s.'
- ' Use create_change_control method',
- self.clnt.apiversion)
+ self.log.debug(f"Wrong method for API version {self.clnt.apiversion}."
+ f" Use create_change_control method")
self.log.warning('create_change_control_v3:'
' Use old change control APIs for old versions')
return None
@@ -2655,7 +2610,7 @@ class CvpApi(object):
stages = []
if sequential:
for index, task in enumerate(tasks):
- stage_id = 'stage%d' % index
+ stage_id = f"stage{index}"
stage = {'stage': [{
'id': stage_id,
'action': {
@@ -2669,7 +2624,7 @@ class CvpApi(object):
else:
stage_rows = []
for index, task in enumerate(tasks):
- stage_id = 'stage%d' % index
+ stage_id = f"stage{index}"
stage_row = {
'id': stage_id,
'action': {
@@ -2704,8 +2659,7 @@ class CvpApi(object):
Ex: {"data": "success"}
'''
- self.log.debug('add_notes_to_change_control: cc_id %s, notes %s'
- % (cc_id, notes))
+ self.log.debug(f"add_notes_to_change_control: cc_id {cc_id}, notes {notes}")
if self.clnt.apiversion is None:
self.get_cvp_info()
if self.clnt.apiversion >= 3.0:
@@ -2760,9 +2714,8 @@ class CvpApi(object):
if self.clnt.apiversion is None:
self.get_cvp_info()
if self.clnt.apiversion < 3.0:
- self.log.debug('Approval methods not valid for API version %s.'
- ' Functionality did not exist',
- self.clnt.apiversion)
+ self.log.debug(f"Approval methods not valid for API version {self.clnt.apiversion}."
+ f" Functionality did not exist")
return None
self.log.debug('v3 Approve change control API Call')
@@ -2781,9 +2734,8 @@ class CvpApi(object):
if self.clnt.apiversion is None:
self.get_cvp_info()
if self.clnt.apiversion < 3.0:
- self.log.debug('Approval methods not valid for API version %s.'
- ' Functionality did not exist',
- self.clnt.apiversion)
+ self.log.debug(f"Approval methods not valid for API version {self.clnt.apiversion}."
+ f" Functionality did not exist")
return None
self.log.debug('v3 Delete Approval for change control API Call')
@@ -2896,7 +2848,7 @@ class CvpApi(object):
'timeZone': '',
'type': 'Custom'}
'''
- self.log.debug('get_change_control_info: %s', cc_id)
+ self.log.debug(f"get_change_control_info: {cc_id}")
if self.clnt.apiversion is None:
self.get_cvp_info()
if self.clnt.apiversion >= 3.0:
@@ -2910,8 +2862,8 @@ class CvpApi(object):
self.log.debug('v2 getChangeControlInformation.do API Call')
try:
resp = self.clnt.get(
- '/changeControl/getChangeControlInformation.do?'
- 'startIndex=0&endIndex=0&ccId=%s' % cc_id,
+ f"/changeControl/getChangeControlInformation.do?"
+ f"startIndex=0&endIndex=0&ccId={cc_id}",
timeout=self.request_timeout)
except CvpApiError as error:
if 'No data found' in error.msg:
@@ -2940,14 +2892,13 @@ class CvpApi(object):
},
u'state': u'Completed'}}]
'''
- self.log.debug('get_change_control_status: %s', cc_id)
+ self.log.debug(f"get_change_control_status: {cc_id}")
if self.clnt.apiversion is None:
self.get_cvp_info()
if self.clnt.apiversion < 3.0:
- self.log.debug('get_change_control_status method not supported'
- ' for API version %s. Use old'
- ' get_change_control_info method'
- % self.clnt.apiversion)
+ self.log.debug(f"get_change_control_status method not supported"
+ f" for API version {self.clnt.apiversion}. Use old"
+ f" get_change_control_info method")
return None
self.log.debug(
@@ -2972,8 +2923,7 @@ class CvpApi(object):
Ex: {u'data': {u'status': u'success', u'taskIds': []}}
'''
- info = ('App %s resetting device %s and moving it to Undefined'
- % (app_name, device['fqdn']))
+ info = (f"App {app_name} resetting device {device['fqdn']} and moving it to Undefined")
self.log.debug(info)
if 'parentContainerId' in device:
@@ -3001,8 +2951,7 @@ class CvpApi(object):
self._add_temp_action(data)
except CvpApiError as error:
if 'Data already exists' in str(error):
- self.log.debug('Device %s already in container Undefined'
- % device['fqdn'])
+ self.log.debug(f"Device {device['fqdn']} already in container Undefined")
if create_task:
return self._save_topology_v2([])
return None
@@ -3028,7 +2977,7 @@ class CvpApi(object):
Ex: {u'data': {u'status': u'success', u'taskIds': [u'32']}}
'''
- info = 'Deploy device %s to container %s' % (device['fqdn'], container)
+ info = f"Deploy device {device['fqdn']} to container {container}"
self.log.debug(info)
container_info = self.get_container_by_name(container)
# Add action for moving device to specified container
@@ -3037,8 +2986,8 @@ class CvpApi(object):
# Get proposed configlets device will inherit from container it is
# being moved to.
- prop_conf = self.clnt.get('/provisioning/getTempConfigsByNetElementId.'
- 'do?netElementId=%s' % device['key'])
+ prop_conf = self.clnt.get(f"/provisioning/getTempConfigsByNetElementId."
+ f"do?netElementId={device['key']}")
new_configlets = prop_conf['proposedConfiglets']
if configlets:
new_configlets.extend(configlets)
@@ -3061,30 +3010,33 @@ class CvpApi(object):
devices (list): list of device Serial Numbers for which the
token should be generated. The default is all devices.
duration (string): the token's validity time (max 1 month),
- accepted formats are: "24h", "86400s", "60m"
+ accepted formats for the legacy endpoint: "24h", "86400s", "60m"
+ accepted format for the new endpoint: "86400s" (only seconds)
Returns:
response (list) on CVaaS: A list that contains the generated
enrollment token.
Ex: [{'enrollmentToken':{'token': <token>, 'groups': [],
'reenrollDevices': <devices list>,
- 'validFor': <duration e.g 24h>, 'field_mask': None}}]
+ 'validFor': <duration e.g 86400s>, 'field_mask': None}}]
response (dict) on CV on-prem: A dictionary that contains the
generated enrollment token.
Ex: {'data': <token>}
'''
+ endpoint_legacy = '/cvpservice/enroll/createToken'
+ endpoint = '/api/resources/admin.Enrollment/AddEnrollmentToken'
if not devices:
devices = ["*"]
# For on-prem check the version as it is only supported from 2021.2.0+
if not self.clnt.is_cvaas:
if self.clnt.apiversion is None:
self.get_cvp_info()
+ # TODO: update this check when 2024.2.0 is released
if self.clnt.apiversion >= 6.0:
self.log.debug('v6 /cvpservice/enroll/createToken')
data = {"reenrollDevices": devices, "duration": duration}
- return self.clnt.post('/cvpservice/enroll/createToken',
- data=data, timeout=self.request_timeout)
+ return self.clnt.post(endpoint_legacy, data=data, timeout=self.request_timeout)
self.log.warning(
'Enrollment Tokens only supported on CVP 2021.2.0+')
return None
@@ -3092,9 +3044,7 @@ class CvpApi(object):
"enrollmentToken": {"reenrollDevices": devices,
"validFor": duration}
}
- return self.clnt.post(
- '/api/v3/services/admin.Enrollment/AddEnrollmentToken',
- data=data, timeout=self.request_timeout)
+ return self.clnt.post(endpoint, data=data, timeout=self.request_timeout)
def get_all_tags(self, element_type='ELEMENT_TYPE_UNSPECIFIED', workspace_id=''):
''' Get all device and/or interface tags from the mainline workspace or all other workspaces
@@ -3121,8 +3071,9 @@ class CvpApi(object):
}
]
}
- self.log.debug('v6 {}'.format(tag_url))
+ self.log.debug(f"v6 {tag_url}")
return self.clnt.post(tag_url, data=payload)
+ return None
def get_tag_edits(self, workspace_id):
''' Show all tags edits in a workspace
@@ -3151,6 +3102,7 @@ class CvpApi(object):
}
self.log.debug('v6 ' + tag_url + ' ' + str(payload))
return self.clnt.post(tag_url, data=payload)
+ return None
def get_tag_assignment_edits(self, workspace_id):
''' Show all tags assignment edits in a workspace
@@ -3179,6 +3131,7 @@ class CvpApi(object):
}
self.log.debug('v6 ' + tag_url + ' ' + str(payload))
return self.clnt.post(tag_url, data=payload)
+ return None
def tag_config(self, element_type, workspace_id, tag_label, tag_value, remove=False):
''' Create/Delete device or interface tags.
@@ -3213,8 +3166,9 @@ class CvpApi(object):
},
"remove": remove
}
- self.log.debug('v6 {} '.format(tag_url) + str(payload))
+ self.log.debug(f"v6 {tag_url} " + str(payload))
return self.clnt.post(tag_url, data=payload)
+ return None
def tag_assignment_config(self, element_type, workspace_id, tag_label,
tag_value, device_id, interface_id, remove=False):
@@ -3257,8 +3211,9 @@ class CvpApi(object):
},
"remove": remove
}
- self.log.debug('v6 {} '.format(tag_url) + str(payload))
+ self.log.debug(f"v6 {tag_url} " + str(payload))
return self.clnt.post(tag_url, data=payload)
+ return None
def get_all_workspaces(self):
''' Get state information for all workspaces
@@ -3271,8 +3226,9 @@ class CvpApi(object):
if self.cvp_version_compare('>=', 6.0, msg):
workspace_url = '/api/resources/workspace/v1/Workspace/all'
payload = {}
- self.log.debug('v6 {}'.format(workspace_url))
+ self.log.debug(f"v6 {workspace_url}")
return self.clnt.post(workspace_url, data=payload)
+ return None
def get_workspace(self, workspace_id):
''' Get state information for all workspaces
@@ -3283,10 +3239,10 @@ class CvpApi(object):
msg = 'Workspace Resource APIs are supported from 2021.2.0 or newer.'
# For on-prem check the version as it is only supported from 2021.2.0+
if self.cvp_version_compare('>=', 6.0, msg):
- workspace_url = '/api/resources/workspace/v1/Workspace?key.workspaceId={}'.format(
- workspace_id)
- self.log.debug('v6 {}'.format(workspace_url))
+ workspace_url = f"/api/resources/workspace/v1/Workspace?key.workspaceId={workspace_id}"
+ self.log.debug(f"v6 {workspace_url}")
return self.clnt.get(workspace_url)
+ return None
def workspace_config(self, workspace_id, display_name,
description='', request='REQUEST_UNSPECIFIED',
@@ -3333,6 +3289,7 @@ class CvpApi(object):
}
self.log.debug('v6 ' + str(workspace_url) + ' ' + str(payload))
return self.clnt.post(workspace_url, data=payload)
+ return None
def workspace_build_status(self, workspace_id, build_id):
''' Verify the state of the workspace build process.
@@ -3349,10 +3306,11 @@ class CvpApi(object):
msg = 'Workspace Resource APIs are supported from 2021.2.0 or newer.'
# For on-prem check the version as it is only supported from 2021.2.0+
if self.cvp_version_compare('>=', 6.0, msg):
- params = 'key.workspaceId={}&key.buildId={}'.format(workspace_id, build_id)
+ params = f"key.workspaceId={workspace_id}&key.buildId={build_id}"
workspace_url = '/api/resources/workspace/v1/WorkspaceBuild?' + params
- self.log.debug('v6 {}'.format(workspace_url + params))
+ self.log.debug(f"v6 {workspace_url + params}")
return self.clnt.get(workspace_url, timeout=self.request_timeout)
+ return None
def change_control_get_one(self, cc_id, cc_time=None):
''' Get the configuration and status of a change control using Resource APIs.
@@ -3380,11 +3338,11 @@ class CvpApi(object):
# For on-prem check the version as it is only supported from 2021.2.0+
if self.cvp_version_compare('>=', 6.0, msg):
if cc_time is None:
- params = 'key.id={}'.format(cc_id)
+ params = f"key.id={cc_id}"
else:
- params = 'key.id={}&time={}'.format(cc_id, cc_time)
+ params = f"key.id={cc_id}&time={cc_time}"
cc_url = '/api/resources/changecontrol/v1/ChangeControl?' + params
- self.log.debug('v6 {}'.format(cc_url))
+ self.log.debug(f"v6 {cc_url}")
try:
response = self.clnt.get(cc_url, timeout=self.request_timeout)
except Exception as error:
@@ -3392,6 +3350,7 @@ class CvpApi(object):
return None
raise error
return response
+ return None
def change_control_get_all(self):
''' Get the configuration and status of all Change Controls using Resource APIs.
@@ -3404,8 +3363,9 @@ class CvpApi(object):
# For on-prem check the version as it is only supported from 2021.2.0+
if self.cvp_version_compare('>=', 6.0, msg):
cc_url = '/api/resources/changecontrol/v1/ChangeControl/all'
- self.log.debug('v6 {}'.format(cc_url))
+ self.log.debug(f"v6 {cc_url}")
return self.clnt.get(cc_url, timeout=self.request_timeout)
+ return None
def change_control_approval_get_one(self, cc_id, cc_time=None):
''' Get the state of a specific Change Control's approve config using Resource APIs.
@@ -3426,9 +3386,9 @@ class CvpApi(object):
# For on-prem check the version as it is only supported from 2021.2.0+
if self.cvp_version_compare('>=', 6.0, msg):
if cc_time is None:
- params = 'key.id={}'.format(cc_id)
+ params = f"key.id={cc_id}"
else:
- params = 'key.id={}&time={}'.format(cc_id, cc_time)
+ params = f"key.id={cc_id}&time={cc_time}"
cc_url = '/api/resources/changecontrol/v1/ApproveConfig?' + params
cc_status = self.change_control_get_one(cc_id)
if cc_status is None:
@@ -3439,6 +3399,7 @@ class CvpApi(object):
" state to be populated.")
return None
return self.clnt.get(cc_url, timeout=self.request_timeout)
+ return None
def change_control_approval_get_all(self):
''' Get state information for all Change Control Approvals using Resource APIs.
@@ -3451,8 +3412,9 @@ class CvpApi(object):
# For on-prem check the version as it is only supported from 2021.2.0+
if self.cvp_version_compare('>=', 6.0, msg):
cc_url = '/api/resources/changecontrol/v1/ApproveConfig/all'
- self.log.debug('v6 {}'.format(cc_url))
+ self.log.debug(f"v6 {cc_url}")
return self.clnt.get(cc_url, timeout=self.request_timeout)
+ return None
def change_control_approve(self, cc_id, notes="", approve=True):
''' Approve/Unapprove a change control using Resource APIs.
@@ -3498,10 +3460,11 @@ class CvpApi(object):
msg = 'Change Control Resource APIs are supported from 2021.2.0 or newer.'
# For on-prem check the version as it is only supported from 2021.2.0+
if self.cvp_version_compare('>=', 6.0, msg):
- params = 'key.id={}'.format(cc_id)
+ params = f"key.id={cc_id}"
cc_url = '/api/resources/changecontrol/v1/ChangeControlConfig?' + params
- self.log.debug('v6 {}'.format(cc_url))
+ self.log.debug(f"v6 {cc_url}")
return self.clnt.delete(cc_url, timeout=self.request_timeout)
+ return None
def change_control_create_with_custom_stages(self, custom_cc=None):
''' Create a Change Control with custom stage hierarchy using Resource APIs.
@@ -3636,6 +3599,7 @@ class CvpApi(object):
cc_url = '/api/resources/changecontrol/v1/ChangeControlConfig'
self.log.debug('v6 ' + str(cc_url) + ' ' + str(payload))
return self.clnt.post(cc_url, data=payload)
+ return None
def change_control_create_for_tasks(self, cc_id, name, tasks, series=True):
''' Create a simple Change Control for tasks using Resource APIs.
@@ -3659,7 +3623,7 @@ class CvpApi(object):
stages = {'values': {'root': {'name': 'root', 'rows': {'values': []}}}}
if series:
for index, task in enumerate(tasks):
- stage_id = 'stage%d' % index
+ stage_id = f"stage{index}"
stages['values']['root']['rows']['values'].append({'values': [stage_id]})
stages['values'][stage_id] = {
'action': {
@@ -3676,7 +3640,7 @@ class CvpApi(object):
else:
stages['values']['root']['rows']['values'].append({'values': []})
for index, task in enumerate(tasks):
- stage_id = 'stage%d' % index
+ stage_id = f"stage{index}"
stages['values']['root']['rows']['values'][0]['values'].append(stage_id)
stages['values'][stage_id] = {
'action': {
@@ -3707,6 +3671,7 @@ class CvpApi(object):
cc_url = '/api/resources/changecontrol/v1/ChangeControlConfig'
self.log.debug('v6 ' + str(cc_url) + ' ' + str(payload))
return self.clnt.post(cc_url, data=payload, timeout=self.request_timeout)
+ return None
def change_control_start(self, cc_id, notes=""):
''' Start a Change Control using Resource APIs.
@@ -3734,6 +3699,7 @@ class CvpApi(object):
cc_url = '/api/resources/changecontrol/v1/ChangeControlConfig'
self.log.debug('v6 ' + str(cc_url) + ' ' + str(payload))
return self.clnt.post(cc_url, data=payload, timeout=self.request_timeout)
+ return None
def change_control_stop(self, cc_id, notes=""):
''' Stop a Change Control using Resource APIs.
@@ -3762,6 +3728,7 @@ class CvpApi(object):
cc_url = '/api/resources/changecontrol/v1/ChangeControlConfig'
self.log.debug('v6 ' + str(cc_url) + ' ' + str(payload))
return self.clnt.post(cc_url, data=payload, timeout=self.request_timeout)
+ return None
def change_control_schedule(self, cc_id, schedule_time, notes=""):
''' Schedule a Change Control using Resource APIs.
@@ -3793,6 +3760,7 @@ class CvpApi(object):
cc_url = '/api/resources/changecontrol/v1/ChangeControlConfig'
self.log.debug('v8 ' + str(cc_url) + ' ' + str(payload))
return self.clnt.post(cc_url, data=payload, timeout=self.request_timeout)
+ return None
def device_decommissioning(self, device_id, request_id):
''' Decommission a device using Resource APIs.
@@ -3802,7 +3770,7 @@ class CvpApi(object):
request_id (string): Key identifies the request to decommission the device.
Recommended to generate uuid with str(uuid.uuid4()).
Returns:
- response (dict): Returns None if the device is not found else returns A dict that contains...
+ response (dict): Returns None if device is not found else return dict that contains
Ex: {'value': {'key': {'requestId': '4a4ba5a2-9886-4cd5-84d6-bdaf85a9f091'},
'deviceId': 'BAD032986065E8DC14CBB6472EC314A6'},
'time': '2022-02-12T02:58:30.765459650Z'}
@@ -3827,10 +3795,9 @@ class CvpApi(object):
self.log.debug('v7 ' + str(url) + ' ' + str(payload))
return self.clnt.post(url, data=payload, timeout=self.request_timeout)
else:
- self.log.warning(
- 'Device with %s serial number does not exist (or is not registered) to decommission'
- % device_id)
- return None
+ self.log.warning("Device with %s serial number does not exist (or is not registered)"
+ " to decommission", device_id)
+ return None
def device_decommissioning_status_get_one(self, request_id):
''' Get the decommission status of a device using Resource APIs.
@@ -3847,10 +3814,11 @@ class CvpApi(object):
msg = 'Decommissioning via Resource APIs are supported from 2021.3.0 or newer.'
# For on-prem check the version as it is only supported from 2021.3.0+
if self.cvp_version_compare('>=', 7.0, msg):
- params = 'key.requestId={}'.format(request_id)
+ params = f"key.requestId={request_id}"
url = '/api/resources/inventory/v1/DeviceDecommissioning?' + params
self.log.debug('v7 ' + str(url))
return self.clnt.get(url, timeout=self.request_timeout)
+ return None
def device_decommissioning_status_get_all(self, status="DECOMMISSIONING_STATUS_UNSPECIFIED"):
''' Get the decommissioning status of all devices using Resource APIs.
@@ -3882,6 +3850,7 @@ class CvpApi(object):
url = '/api/resources/inventory/v1/DeviceDecommissioning/all'
self.log.debug('v7 ' + str(url))
return self.clnt.post(url, data=payload, timeout=self.request_timeout)
+ return None
def add_role(self, name, description, moduleList):
''' Add new local role to the CVP UI.
@@ -3918,9 +3887,10 @@ class CvpApi(object):
rolekey (str): role key on CVP
Returns:
response (dict): Returns a dict that contains the role.
- Ex: {'name': 'Test Role', 'key': 'role_1599019487020581247', 'description': 'Test'...}
+ Ex: {'name': 'Test Role', 'key': 'role_1599019487020581247',
+ 'description': 'Test'...}
'''
- return self.clnt.get('/role/getRole.do?roleId={}'.format(rolekey),
+ return self.clnt.get(f"/role/getRole.do?roleId={rolekey}",
timeout=self.request_timeout)
def get_roles(self):
@@ -3962,8 +3932,9 @@ class CvpApi(object):
msg = 'Service Account Resource APIs are supported from 2021.3.0+.'
if self.cvp_version_compare('>=', 7.0, msg):
url = '/api/v3/services/arista.serviceaccount.v1.TokenService/GetAll'
- self.log.debug('v7 {}'.format(url))
+ self.log.debug(f"v7 {url}")
return self.clnt.post(url)
+ return None
def svc_account_token_get_one(self, token_id):
''' Get a service account token's state using Resource APIs
@@ -3979,8 +3950,9 @@ class CvpApi(object):
if self.cvp_version_compare('>=', 7.0, msg):
payload = {"key": {"id": token_id}}
url = '/api/v3/services/arista.serviceaccount.v1.TokenService/GetOne'
- self.log.debug('v7 {} {}'.format(url, payload))
+ self.log.debug(f"v7 {url} {payload}")
return self.clnt.post(url, data=payload)
+ return None
def svc_account_token_delete(self, token_id):
''' Delete a service account token using Resource APIs.
@@ -3996,8 +3968,9 @@ class CvpApi(object):
if self.cvp_version_compare('>=', 7.0, msg):
payload = {"key": {"id": token_id}}
url = '/api/v3/services/arista.serviceaccount.v1.TokenConfigService/Delete'
- self.log.debug('v7 {} {}'.format(url, payload))
+ self.log.debug(f"v7 {url} {payload}")
return self.clnt.post(url, data=payload)
+ return None
def svc_account_token_set(self, username, duration, description):
''' Create a service account token using Resource APIs.
@@ -4020,8 +3993,9 @@ class CvpApi(object):
msg = 'Service Account Resource APIs are supported from 2021.3.0+.'
if self.cvp_version_compare('>=', 7.0, msg):
url = '/api/v3/services/arista.serviceaccount.v1.TokenConfigService/Set'
- self.log.debug('v7 {} {}'.format(url, payload))
+ self.log.debug(f"v7 {url} {payload}")
return self.clnt.post(url, data=payload)
+ return None
def svc_account_get_all(self):
''' Get all service account states using Resource APIs.
@@ -4036,8 +4010,9 @@ class CvpApi(object):
msg = 'Service Account Resource APIs are supported from 2021.3.0+.'
if self.cvp_version_compare('>=', 7.0, msg):
url = '/api/v3/services/arista.serviceaccount.v1.AccountService/GetAll'
- self.log.debug('v7 {} '.format(url))
+ self.log.debug(f"v7 {url}")
return self.clnt.post(url)
+ return None
def svc_account_get_one(self, username):
''' Get a service account's state using Resource APIs
@@ -4054,8 +4029,9 @@ class CvpApi(object):
if self.cvp_version_compare('>=', 7.0, msg):
payload = {"key": {"name": username}}
url = '/api/v3/services/arista.serviceaccount.v1.AccountService/GetOne'
- self.log.debug('v7 {} {}'.format(url, payload))
+ self.log.debug(f"v7 {url} {payload}")
return self.clnt.post(url, data=payload)
+ return None
def svc_account_set(self, username, description, roles, status):
''' Create a service account using Resource APIs.
@@ -4087,16 +4063,17 @@ class CvpApi(object):
if role['key'] in roles or role['name'] in roles:
role_ids.append(role['key'])
if len(roles) != len(role_ids):
- self.log.warning('Not all provided roles {} are valid. '
- 'Only using the found valid roles {}'.format(roles, role_ids))
+ self.log.warning(f"Not all provided roles {roles} are valid. "
+ f"Only using the found valid roles {role_ids}")
payload = {'value': {'description': description,
'groups': {'values': role_ids},
'key': {'name': username},
'status': status}}
url = '/api/v3/services/arista.serviceaccount.v1.AccountConfigService/Set'
- self.log.debug('v7 {} {}'.format(url, payload))
+ self.log.debug(f"v7 {url} {payload}")
return self.clnt.post(url, data=payload)
+ return None
def svc_account_delete(self, username):
''' Delete a service account using Resource APIs.
@@ -4112,8 +4089,9 @@ class CvpApi(object):
if self.cvp_version_compare('>=', 7.0, msg):
payload = {"key": {"name": username}}
url = '/api/v3/services/arista.serviceaccount.v1.AccountConfigService/Delete'
- self.log.debug('v7 {} {}'.format(url, payload))
+ self.log.debug(f"v7 {url} {payload}")
return self.clnt.post(url, data=payload)
+ return None
def svc_account_delete_expired_tokens(self):
''' Delete all service account tokens using Resource APIs.
diff --git a/cvprac/cvp_client.py b/cvprac/cvp_client.py
index 602f21a..a02676f 100644
--- a/cvprac/cvp_client.py
+++ b/cvprac/cvp_client.py
@@ -29,6 +29,9 @@
# OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN
# IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
#
+
+# pylint: disable=too-many-branches,too-many-statements,too-many-locals,too-many-lines
+
''' RESTful API Client class for Cloudvision(R) Portal
This module provides a RESTful API client for Cloudvision(R) Portal (CVP)
@@ -96,18 +99,24 @@ import json
import logging
from logging.handlers import SysLogHandler
from itertools import cycle
-from pkg_resources import parse_version
+from packaging.version import parse
import requests
-from requests.exceptions import ConnectionError, HTTPError, Timeout, \
- ReadTimeout, TooManyRedirects, JSONDecodeError
+from requests.exceptions import ( # pylint: disable=redefined-builtin
+ ConnectionError,
+ HTTPError,
+ Timeout,
+ ReadTimeout,
+ TooManyRedirects,
+ JSONDecodeError
+)
from cvprac.cvp_api import CvpApi
from cvprac.cvp_client_errors import CvpApiError, CvpLoginError, \
CvpRequestError, CvpSessionLogOutError
-class CvpClient(object):
+class CvpClient():
''' Use this class to create a persistent connection to CVP.
'''
# pylint: disable=too-many-instance-attributes
@@ -233,28 +242,37 @@ class CvpClient(object):
' Appending 0. Updated Version String - %s',
".".join(version_components))
full_version = ".".join(version_components)
- if parse_version(full_version) >= parse_version('2023.1.0'):
+ if parse(full_version) >= parse('2024.1.0'):
+ self.log.info('Setting API version to v12')
+ self.apiversion = 12.0
+ elif parse(full_version) >= parse('2023.3.0'):
+ self.log.info('Setting API version to v11')
+ self.apiversion = 11.0
+ elif parse(full_version) >= parse('2023.2.0'):
+ self.log.info('Setting API version to v10')
+ self.apiversion = 10.0
+ elif parse(full_version) >= parse('2023.1.0'):
self.log.info('Setting API version to v9')
self.apiversion = 9.0
- elif parse_version(full_version) >= parse_version('2022.1.0'):
+ elif parse(full_version) >= parse('2022.1.0'):
self.log.info('Setting API version to v8')
self.apiversion = 8.0
- elif parse_version(full_version) >= parse_version('2021.3.0'):
+ elif parse(full_version) >= parse('2021.3.0'):
self.log.info('Setting API version to v7')
self.apiversion = 7.0
- elif parse_version(full_version) >= parse_version('2021.2.0'):
+ elif parse(full_version) >= parse('2021.2.0'):
self.log.info('Setting API version to v6')
self.apiversion = 6.0
- elif parse_version(full_version) >= parse_version('2020.2.4'):
+ elif parse(full_version) >= parse('2020.2.4'):
self.log.info('Setting API version to v5')
self.apiversion = 5.0
- elif parse_version(full_version) >= parse_version('2020.1.1'):
+ elif parse(full_version) >= parse('2020.1.1'):
self.log.info('Setting API version to v4')
self.apiversion = 4.0
- elif parse_version(full_version) >= parse_version('2019.0.0'):
+ elif parse(full_version) >= parse('2019.0.0'):
self.log.info('Setting API version to v3')
self.apiversion = 3.0
- elif parse_version(full_version) >= parse_version('2018.2.0'):
+ elif parse(full_version) >= parse('2018.2.0'):
self.log.info('Setting API version to v2')
self.apiversion = 2.0
else:
@@ -378,13 +396,12 @@ class CvpClient(object):
self.error_msg = '\n'
for _ in range(0, num_nodes):
host = next(self.node_pool)
- self.url_prefix = ('https://%s:%d/web' % (host, self.port or 443))
- self.url_prefix_short = ('https://%s:%d'
- % (host, self.port or 443))
+ self.url_prefix = f"https://{host}:{self.port or 443}/web"
+ self.url_prefix_short = f"https://{host}:{self.port or 443}"
error = self._reset_session()
if error is None:
break
- self.error_msg += '%s: %s\n' % (host, error)
+ self.error_msg += f"{host}: {error}\n"
def _reset_session(self):
''' Get a new request session and try logging into the current
@@ -428,23 +445,20 @@ class CvpClient(object):
if 'Unauthorized' in response.reason:
# Check for 'Unauthorized' User error because this is how
# CVP responds to a logged out users requests in 2018.x.
- msg = '%s: Request Error: %s' % (prefix, response.reason)
+ msg = f"{prefix}: Request Error: {response.reason}"
self.log.error(msg)
raise CvpApiError(msg)
if 'User is unauthorized' in response.text:
# Check for 'User is unauthorized' response text because this
# is how CVP responds to a logged out users requests in 2019.x.
- msg = '%s: Request Error: User is unauthorized' % prefix
+ msg = f"{prefix}: Request Error: User is unauthorized"
self.log.error(msg)
raise CvpApiError(msg)
- else:
- msg = '%s: Request Error: %s - %s' % (prefix, response.reason,
- response.text)
- self.log.error(msg)
- raise CvpRequestError(msg)
+ msg = f"{prefix}: Request Error: {response.reason} - {response.text}"
+ raise CvpRequestError(msg)
if 'LOG OUT MESSAGE' in response.text:
- msg = ('%s: Request Error: session logged out' % prefix)
+ msg = f"{prefix}: Request Error: session logged out"
raise CvpSessionLogOutError(msg)
joutput = json_decoder(response.text)
@@ -460,9 +474,9 @@ class CvpClient(object):
# Build the error message from all the errors.
err_msg = error_list[0]
for idx in range(1, len(error_list)):
- err_msg = '%s\n%s' % (err_msg, error_list[idx])
+ err_msg = f"{err_msg}\n{error_list[idx]}"
- msg = ('%s: Request Error: %s' % (prefix, err_msg))
+ msg = f"{prefix}: Request Error: {err_msg}"
self.log.error(msg)
raise CvpApiError(msg)
@@ -477,8 +491,7 @@ class CvpClient(object):
response status is not OK.
'''
if not response.ok:
- msg = '%s: Request Error: %s - %s' % (prefix, response.reason,
- response.text)
+ msg = f"{prefix}: Request Error: {response.reason} - {response.text}"
self.log.error(msg)
raise CvpRequestError(msg)
@@ -512,7 +525,7 @@ class CvpClient(object):
self.headers.pop('APP_SESSION_ID', None)
if self.api_token is not None:
return self._set_headers_api_token()
- elif self.is_cvaas:
+ if self.is_cvaas:
raise CvpLoginError('CVaaS only supports API token authentication.'
' Please create an API token and provide it'
' via the api_token parameter in combination'
@@ -551,7 +564,7 @@ class CvpClient(object):
headers=self.headers,
timeout=self.connect_timeout,
verify=self.cert)
- self._is_good_response(response, 'Authenticate: %s' % url)
+ self._is_good_response(response, f"Authenticate: {url}")
self.cookies = response.cookies
self.headers['APP_SESSION_ID'] = response.json()['sessionId']
@@ -561,18 +574,20 @@ class CvpClient(object):
'''
# If using an API token there is no need to run a Login API.
# Simply add the token into the headers or cookies
- self.headers['Authorization'] = 'Bearer %s' % self.api_token
+ self.headers['Authorization'] = f"Bearer {self.api_token}"
# Alternative to adding token to headers it can be added to
# cookies as shown below.
# self.cookies = {'access_token': self.api_token}
url = self.url_prefix_short + '/api/v1/rest/'
- response = self.session.get(url,
- cookies=self.cookies,
- headers=self.headers,
- timeout=self.connect_timeout,
- verify=self.cert)
+ response = self.session.get(
+ url,
+ cookies=self.cookies,
+ headers=self.headers,
+ timeout=self.connect_timeout,
+ verify=self.cert
+ )
# Verify that the generic request was successful
- self._is_good_response(response, 'Authenticate: %s' % url)
+ self._is_good_response(response, f"Authenticate: {url}")
def logout(self):
'''
@@ -584,7 +599,7 @@ class CvpClient(object):
self.log.info('User logged out.')
self.session = None
else:
- err = 'Error trying to logout %s' % response
+ err = f"Error trying to logout {response}"
self.log.error(err)
def _make_request(self, req_type, url, timeout, data=None,
@@ -700,8 +715,8 @@ class CvpClient(object):
try:
resp_data = response.json()
- if (resp_data is not None and 'result' in resp_data
- and '/resources/' in full_url):
+ if (resp_data is not None and 'result' in resp_data and
+ '/resources/' in full_url):
# Resource APIs use JSON streaming and will return
# multiple JSON objects during GetAll type API
# calls. We are wrapping the multiple objects into
@@ -725,10 +740,8 @@ class CvpClient(object):
' response data. Attempt to decode')
decoded_data = json_decoder(response.text)
return {'data': decoded_data}
- else:
- self.log.error('Unknown format for JSONDecodeError - %s',
- err_str)
- raise error
+ self.log.error("Unknown format for JSONDecodeError - %s", err_str)
+ raise error
def _send_request(self, req_type, full_url, timeout, data=None,
files=None):
@@ -795,7 +808,7 @@ class CvpClient(object):
timeout=timeout,
verify=self.cert)
else:
- fhs = dict()
+ fhs = {}
fhs['Accept'] = self.headers['Accept']
if 'APP_SESSION_ID' in self.headers:
fhs['APP_SESSION_ID'] = self.headers[
@@ -830,8 +843,7 @@ class CvpClient(object):
continue
try:
- self._is_good_response(response, '%s: %s ' %
- (req_type, full_url))
+ self._is_good_response(response, f"{req_type}: {full_url} ")
except CvpSessionLogOutError as error:
self.log.debug(error)
# Retry the request to the same node if there was a CVP session
@@ -840,11 +852,10 @@ class CvpClient(object):
# be retried on the same node.
if req_try + 1 == self.NUM_RETRY_REQUESTS:
raise error
- else:
- self._reset_session()
- if not self.session:
- raise error
- continue
+ self._reset_session()
+ if not self.session:
+ raise error
+ continue
except CvpApiError as error:
self.log.debug(error)
if ('Unauthorized' in error.msg or
@@ -859,14 +870,12 @@ class CvpClient(object):
# will be retried on the same node.
if req_try + 1 == self.NUM_RETRY_REQUESTS:
raise error
- else:
- self._reset_session()
- if not self.session:
- raise error
- continue
- else:
- # pylint: disable=raising-bad-type
- raise error
+ self._reset_session()
+ if not self.session:
+ raise error
+ continue
+ # pylint: disable=raising-bad-type
+ raise error
return response
def get(self, url, timeout=30):
diff --git a/dev-requirements.txt b/dev-requirements.txt
index 2395cb7..a0adc90 100644
--- a/dev-requirements.txt
+++ b/dev-requirements.txt
@@ -1,6 +1,5 @@
check-manifest
coverage
-mock
pdoc
pep8
pyflakes
diff --git a/docs/labs/lab06-provisioning/vc_task_retrigger.py b/docs/labs/lab06-provisioning/vc_task_retrigger.py
index b5586ee..cfee3a1 100644
--- a/docs/labs/lab06-provisioning/vc_task_retrigger.py
+++ b/docs/labs/lab06-provisioning/vc_task_retrigger.py
@@ -7,7 +7,7 @@
import argparse
import ssl
import sys
-from pkg_resources import parse_version
+from packaging.version import parse
from getpass import getpass
from cvprac.cvp_client import CvpClient
import requests.packages.urllib3
@@ -56,7 +56,7 @@ def main():
# Get the current CVP version
cvp_release = clnt.api.get_cvp_info()['version']
- if parse_version(cvp_release) < parse_version('2020.3.0'):
+ if parse(cvp_release) < parse('2020.3.0'):
# For older CVP, we manually trigger a compliance check
try:
clnt.api.check_compliance('root', 'container')
diff --git a/docs/release-notes-1.4.0.rst b/docs/release-notes-1.4.0.rst
new file mode 100644
index 0000000..b6c81fb
--- /dev/null
+++ b/docs/release-notes-1.4.0.rst
@@ -0,0 +1,20 @@
+######
+v1.4.0
+######
+
+2024-5-6
+
+Enhancements
+^^^^^^^^^^^^
+
+* Move from pkg_resources to packaging for Python 3.12 support. (`271 <https://github.com/aristanetworks/cvprac/pull/271>`_) [`mharista <https://github.com/mharista>`_]
+* Add support for searchTopology V3 endpoint. (`275 <https://github.com/aristanetworks/cvprac/pull/275>`_) [`mharista <https://github.com/mharista>`_]
+
+Fixed
+^^^^^
+
+* Add missing url encoding for get_user. (`264 <https://github.com/aristanetworks/cvprac/pull/264>`_) [`noredistribution <https://github.com/noredistribution>`_]
+* Updated the enrollment endpoint for CVaaS. (`269 <https://github.com/aristanetworks/cvprac/pull/269>`_) [`noredistribution <https://github.com/noredistribution>`_]
+* Add timeout to get_configlets_and_mappers(). (`270 <https://github.com/aristanetworks/cvprac/pull/270>`_) [`noredistribution <https://github.com/noredistribution>`_]
+* Update setup.py to reference python3 only. (`272 <https://github.com/aristanetworks/cvprac/pull/272>`_) [`mharista <https://github.com/mharista>`_]
+* Python3 lint/format fixes. (`273 <https://github.com/aristanetworks/cvprac/pull/273>`_) [`mharista <https://github.com/mharista>`_]
diff --git a/requirements.txt b/requirements.txt
index 9029d8d..108c9fb 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1 +1,2 @@
requests[socks]>=2.27.0
+packaging>=23.2
diff --git a/setup.py b/setup.py
index 8c2c06b..ddb6654 100644
--- a/setup.py
+++ b/setup.py
@@ -100,8 +100,13 @@ setup(
# Specify the Python versions you support here. In particular, ensure
# that you indicate whether you support Python 2, Python 3 or both.
- 'Programming Language :: Python :: 2',
- 'Programming Language :: Python :: 2.7',
+ 'Programming Language :: Python :: 3 :: Only',
+ 'Programming Language :: Python :: 3.7',
+ 'Programming Language :: Python :: 3.8',
+ 'Programming Language :: Python :: 3.9',
+ 'Programming Language :: Python :: 3.10',
+ 'Programming Language :: Python :: 3.11',
+ 'Programming Language :: Python :: 3.12',
],
# What does your project relate to?
@@ -111,7 +116,7 @@ setup(
# your project is installed. For an analysis of "install_requires" vs pip's
# requirements files see:
# https://packaging.python.org/en/latest/requirements.html
- install_requires=['requests[socks]>=2.27.0'],
+ install_requires=['requests[socks]>=2.27.0', 'packaging>=23.2'],
# List additional groups of dependencies here (e.g. development
# dependencies). You can install these using the following syntax,