Fix(cv_facts_v3): return authorization error when svc token expires

ansible_collections/arista/cvp/plugins/module_utils/facts_tools.py

@@ -24,11 +24,13 @@
 import re
 import os
 from typing import List
+from ansible.module_utils.basic import AnsibleModule
 from concurrent.futures import ThreadPoolExecutor
 import ansible_collections.arista.cvp.plugins.module_utils.logger   # noqa # pylint: disable=unused-import
 from ansible_collections.arista.cvp.plugins.module_utils.resources.api.fields import Api
 from ansible_collections.arista.cvp.plugins.module_utils.resources.modules.fields import FactsResponseFields
 import ansible_collections.arista.cvp.plugins.module_utils.tools_schema as schema   # noqa # pylint: disable=unused-import
+from ansible_collections.arista.cvp.plugins.module_utils.tools_cv import authorization_error
 try:
     from cvprac.cvp_client import CvpClient  # noqa # pylint: disable=unused-import
     from cvprac.cvp_client_errors import CvpApiError, CvpRequestError  # noqa # pylint: disable=unused-import
@@ -322,8 +324,9 @@ class CvFactsTools():
     CvFactsTools Object to operate Facts from Cloudvision
     """
 
-    def __init__(self, cv_connection):
+    def __init__(self, cv_connection, ansible_module: AnsibleModule = None):
         self.__cv_client = cv_connection
+        self.__ansible = ansible_module
         self._cache = {FactsResponseFields.CACHE_CONTAINERS: None, FactsResponseFields.CACHE_MAPPERS: None}
         self._max_worker = min(32, (os.cpu_count() or 1) + 4)
         self.__init_facts()
@@ -592,6 +595,7 @@ def __fact_devices(self, filter: str = '.*', verbose: str = 'short'):
             cv_devices = self.__cv_client.api.get_inventory()
         except CvpApiError as error_msg:
             MODULE_LOGGER.error('Error when collecting devices facts: %s', str(error_msg))
+            authorization_error(self.__ansible.fail_json, error_msg)
         MODULE_LOGGER.info('Extract device data using filter %s', str(filter))
         facts_builder = CvFactResource()
         for device in cv_devices:
@@ -614,6 +618,7 @@ def __fact_containers(self, filter: str = '.*'):
             cv_containers = self.__cv_client.api.get_containers()
         except CvpApiError as error_msg:
             MODULE_LOGGER.error('Error when collecting containers facts: %s', str(error_msg))
+            authorization_error(self.__ansible.fail_json, error_msg)
         facts_builder = CvFactResource()
         for container in cv_containers['data']:
             if container[Api.generic.NAME] != 'Tenant':
@@ -637,7 +642,11 @@ def __fact_configlets(self, filter: str = '.*', configlets_per_call: int = 10):
         configlets_per_call : int, optional
             Number of configlets to retrieve per API call, by default 10
         """
-        max_range_calc = self.__cv_client.api.get_configlets(start=0, end=1)['total'] + 1
+        try:
+            max_range_calc = self.__cv_client.api.get_configlets(start=0, end=1)['total'] + 1
+        except CvpApiError as error_msg:
+            MODULE_LOGGER.error('Error when collecting configlets facts: %s', str(error_msg))
+            authorization_error(self.__ansible.fail_json, error_msg)
         futures_list = []
         results = []
         with ThreadPoolExecutor(max_workers=self._max_worker) as executor:
@@ -678,6 +687,7 @@ def __fact_images(self, filter: str = '.*'):
             cv_images = self.__cv_client.api.get_images()
         except CvpApiError as error_msg:
             MODULE_LOGGER.error('Error when collecting images facts: %s', str(error_msg))
+            authorization_error(self.__ansible.fail_json, error_msg)
 
         facts_builder = CvFactResource()
         for image in cv_images['data']:
@@ -710,18 +720,21 @@ def __fact_tasks(self, filter: str = '.*', verbose: str = 'short'):
                 cv_tasks = cv_tasks['data']
             except CvpApiError as error_msg:
                 MODULE_LOGGER.error('Error when collecting task facts: %s', str(error_msg))
+                authorization_error(self.__ansible.fail_json, error_msg)
         elif isinstance(filter, str):
             # filter by task status
             try:
                 cv_tasks = self.__cv_client.api.get_tasks_by_status(filter)
             except CvpApiError as error_msg:
                 MODULE_LOGGER.error('Error when collecting %s task facts: %s', filter, str(error_msg))
+                authorization_error(self.__ansible.fail_json, error_msg)
         elif isinstance(filter, int):
             # filter by task_id
             try:
                 cv_tasks = self.__cv_client.api.get_task_by_id(filter)
             except CvpApiError as error_msg:
                 MODULE_LOGGER.error('Error when collecting %s task facts: %s', filter, str(error_msg))
+                authorization_error(self.__ansible.fail_json, error_msg)
 
         for task in cv_tasks:
             MODULE_LOGGER.debug('Got following information for task: %s', str(task))

ansible_collections/arista/cvp/plugins/module_utils/tools_cv.py

@@ -207,3 +207,19 @@ def cv_update_configlets_on_device(module, device_facts, add_configlets, del_con
         LOGGER.info(" * cv_update_configlets_on_device - device_addition result: %s", str(device_addition))
     LOGGER.info(" * cv_update_configlets_on_device - final result: %s", str(response))
     return response
+
+
+def authorization_error(module, error_msg):
+    """
+    Function to handle authorization error and fail module with correct message.
+
+    Parameters
+    ----------
+    module : AnsibleModule
+        Ansible module information
+    error : str
+        Error message to check if it is an authorization error or not.
+
+    """
+    if "Unauthorized" in str(error_msg):
+        module(msg="Unauthorized access to CloudVision. Please check your credentials or service account token validty!")

ansible_collections/arista/cvp/plugins/modules/cv_facts_v3.py

@@ -183,7 +183,7 @@ def main():
     cv_client = tools_cv.cv_connect(ansible_module)
 
     # Instantiate ansible results
-    facts_collector = CvFactsTools(cv_connection=cv_client)
+    facts_collector = CvFactsTools(cv_connection=cv_client, ansible_module=ansible_module)
     facts = facts_collector.facts(scope=ansible_module.params['facts'], regex_filter=ansible_module.params['regexp_filter'],
                                   verbose=ansible_module.params['verbose'])
     result = dict(changed=False, data=facts, failed=False)