Release 1.4.0

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/

VERSION

@@ -1 +1 @@
-1.3.2
+1.4.0

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.'

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 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 @@ def set_version(self, version):
                               ' 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 @@ def _create_session(self, all_nodes=False):
         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 @@ def _is_good_response(self, response, prefix):
             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 @@ def _is_good_response(self, response, prefix):
                 # 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 @@ def _check_response_status(self, response, prefix):
                 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 @@ def _login(self):
         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 @@ def _login_on_prem(self):
                                      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 @@ def _set_headers_api_token(self):
         '''
         # 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 @@ def logout(self):
             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 @@ def _make_request(self, req_type, url, timeout, data=None,
 
         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 @@ def _make_request(self, req_type, url, timeout, data=None,
                                ' 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 @@ def _send_request(self, req_type, full_url, timeout, data=None,
                                                      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 @@ def _send_request(self, req_type, full_url, timeout, data=None,
                 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 @@ def _send_request(self, req_type, full_url, timeout, data=None,
                 # 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 @@ def _send_request(self, req_type, full_url, timeout, data=None,
                     # 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):

dev-requirements.txt

@@ -1,6 +1,5 @@
 check-manifest
 coverage
-mock
 pdoc
 pep8
 pyflakes

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')

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>`_]

requirements.txt

@@ -1 +1,2 @@
 requests[socks]>=2.27.0
+packaging>=23.2

setup.py

@@ -100,8 +100,13 @@ def get_long_description():
 
         # 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 @@ def get_long_description():
     # 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,

test/lib/systestlib.py

@@ -55,7 +55,7 @@ class DutSystemTest(unittest.TestCase):
     ''' DutSystemTest class that provides information about the DUTs used.
     '''
     def __init__(self, *args, **kwargs):
-        super(DutSystemTest, self).__init__(*args, **kwargs)
+        super().__init__(*args, **kwargs)
 
     @classmethod
     def setUpClass(cls):
@@ -67,6 +67,6 @@ def setUpClass(cls):
         '''
         cls.duts = {}
         filename = get_fixture('cvp_nodes.yaml')
-        with open(filename, 'r') as stream:
+        with open(filename, 'r', encoding="utf-8") as stream:
             cls.duts = yaml.safe_load(stream)
             stream.close()