Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use is_failed method in avocado

Best Python code snippet using avocado_python

nsxt_upgrade_run.py

Source:nsxt_upgrade_run.py

1#!/usr/bin/env python2#3# Copyright 2019 VMware, Inc.4# SPDX-License-Identifier: BSD-2-Clause OR GPL-3.0-only5#6# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING,7# BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.8# IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,9# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;10# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,11# WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE,12# EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.13from __future__ import absolute_import, division, print_function14__metaclass__ = type15ANSIBLE_METADATA = {'metadata_version': '1.1',16                    'status': ['preview'],17                    'supported_by': 'community'}18DOCUMENTATION = '''19---20module: nsxt_upgrade_run21short_description: 'Start the upgrade'22description: 'Upgrade will start as per the upgrade plan.'23version_added: '2.7'24author: 'Kommireddy Akhilesh'25options:26    hostname:27        description: 'Deployed NSX manager hostname.'28        required: true29        type: str30    username:31        description: 'The username to authenticate with the NSX manager.'32        required: true33        type: str34    password:35        description: 'The password to authenticate with the NSX manager.'36        required: true37        type: str38    paused_upgrade:39        description: 'Mode of upgrade'40        required: true41        type: bool42'''43EXAMPLES = '''44- name: Runs the upgrade45  nsxt_upgrade_run:46      hostname: "10.192.167.137"47      username: "admin"48      password: "Admin!23Admin"49      validate_certs: False50      paused_upgrade: True51'''52RETURN = '''# '''53import json, time54from ansible.module_utils.basic import AnsibleModule55from ansible.module_utils.vmware_nsxt import vmware_argument_spec, request56from ansible.module_utils.common_utils import get_attribute_from_endpoint, clean_and_get_params, get_upgrade_orchestrator_node57from ansible.module_utils._text import to_native58def get_upgrade_status(module, manager_url, mgr_username, mgr_password, validate_certs):59  '''60  Get the current status of upgrade at the start.61  Doesn't upgrade if any component is in progress 62  or system is already upgraded.63  '''64  no_of_checks = 065  while True:66    upgrade_status = get_attribute_from_endpoint(module, manager_url, '/upgrade/status-summary',67                     mgr_username, mgr_password, validate_certs, 'overall_upgrade_status', 68                     False)69    no_of_checks = no_of_checks + 170    if upgrade_status == 'IN_PROGRESS' or upgrade_status == 'PAUSING':71      if no_of_checks > 2:72        module.fail_json(msg='Upgrade is in state: %s, can\'t continue' % upgrade_status)73    elif upgrade_status == 'SUCCESS':74      module.exit_json(changed=False, message='Upgrade state is SUCCESS. No need to'75                    ' continue.')76    else:77      return upgrade_status78    time.sleep(20)79def decide_next_step(module, manager_url, mgr_username, mgr_password, 80                     validate_certs, can_continue, is_failed):81  '''82  params:83  - can_continue: if upgrade can be continued 84  - is_failed: Is there any component Failure85  return:86  - Decides the next operation to be done based on 87    can_continue and is_failed values 88  '''89  if can_continue and is_failed:90    return91  elif can_continue and not is_failed:92    return93  elif not can_continue and is_failed:94    raise Exception('Upgrade failed. Please run upgrade status summary'95                    ' to see the reason of upgrade failure.')96  else:97    time.sleep(15)98    try:99      upgrade_status = get_attribute_from_endpoint(module, manager_url, '/upgrade/summary',100                        mgr_username, mgr_password, validate_certs, 'upgrade_status',101                        False)102    except Exception as err:103      return104    if upgrade_status == 'SUCCESS':105      module.exit_json(changed=True, message='System has been upgraded successfully!!!')106    elif upgrade_status == 'IN_PROGRESS' or upgrade_status == 'PAUSING' or upgrade_status == 'PAUSED':107      return108    else:109      module.fail_json(msg='All components till last one are upgraded. Still upgrade status'110        ' is %s. Please run upgrade status summary to see the reason.' % upgrade_status)111def check_continuity(module, manager_url, mgr_username, mgr_password, validate_certs):112  '''113  Returns:114  Based on the output of upgrade status summary API, gets the115  checks and returns if upgrade can be continued and116  if there is any component fail in the upgrade117  '''118  try:119    component_status_list = get_attribute_from_endpoint(module, manager_url, 120                          '/upgrade/status-summary', mgr_username, mgr_password,121                          validate_certs, 'component_status', False)122  except Exception as err:123    can_continue = True124    is_failed = True125    return can_continue, is_failed126  try:127    can_continue = True128    for component_status in component_status_list:129      if component_status['status'] == 'IN_PROGRESS' or \130         component_status['status'] == 'PAUSING':131        can_continue = False132        break133    if not can_continue:134      return can_continue, False135    else:136      is_failed = False137      found_not_started = False138      for component_status in component_status_list[::-1]:139        if component_status['status'] == 'NOT_STARTED':140          found_not_started = True141        elif component_status['status'] == 'PAUSED':142          can_continue = True143          is_failed = False144          return can_continue, is_failed145        elif component_status['status'] == 'SUCCESS':146          if not found_not_started:147            can_continue = False148            is_failed = False149            return can_continue, is_failed150          else:151            can_continue = True152            is_failed = False153            return can_continue, is_failed154        elif component_status['status'] == 'FAILED':155          can_continue = False156          is_failed = True157          return can_continue, is_failed158        elif component_status['status'] == 'IN_PROGRESS' or \159        component_status['status'] == 'PAUSING':160          can_continue = False161          is_failed = False162          return can_continue, is_failed163        else:164          return True, True165  except Exception as err:166    can_continue = True167    is_failed = True168    return can_continue, is_failed169def main():170  argument_spec = vmware_argument_spec()171  argument_spec.update(paused_upgrade=dict(type='bool', required=True))172  module = AnsibleModule(argument_spec=argument_spec, supports_check_mode=True)173  mgr_hostname = module.params['hostname']174  mgr_username = module.params['username']175  mgr_password = module.params['password']176  validate_certs = module.params['validate_certs']177  paused_upgrade = module.params['paused_upgrade']178  179  headers = dict(Accept="application/json")180  headers['Content-Type'] = 'application/json'181  182  mgr_hostname = get_upgrade_orchestrator_node(module, mgr_hostname, mgr_username, 183                                            mgr_password, headers, validate_certs)184  185  manager_url = 'https://{}/api/v1'.format(mgr_hostname)186  if module.check_mode:187    if paused_upgrade:188      module.exit_json(changed=False, debug_out='NSX-T will upgrade with pauses.')189    else:190      module.exit_json(changed=False, debug_out='NSX-T will upgrade without pauses.')191  # If paused_upgrade is not true i.e auto mode192  if not paused_upgrade:193    while True:194      upgrade_status = get_upgrade_status(module, manager_url, mgr_username,195                                          mgr_password, validate_certs)196      if upgrade_status == 'NOT_STARTED':197        try:198          (rc, resp) = request(manager_url+ '/upgrade/plan?action=start', 199                         data='', headers=headers, method='POST', 200                         url_username=mgr_username, url_password=mgr_password, 201                         validate_certs=validate_certs, ignore_errors=True)202        except Exception as err:203          module.fail_json(msg="Failed while upgrading. Error[%s]." % to_native(err))204      else:205        try:206          (rc, resp) = request(manager_url+ '/upgrade/plan?action=continue', 207                         data='', headers=headers, method='POST', 208                         url_username=mgr_username, url_password=mgr_password, 209                         validate_certs=validate_certs, ignore_errors=True)210        except Exception as err:211          module.fail_json(msg="Failed while upgrading. Error[%s]." % to_native(err))212      time.sleep(10)213      while True:214        try:215          can_continue, is_failed = check_continuity(module, manager_url, mgr_username,216                                                     mgr_password, validate_certs)217          decide_next_step(module, manager_url, mgr_username, mgr_password, 218                           validate_certs, can_continue, is_failed)219          if can_continue and not is_failed:220            break221          time.sleep(10)222        except Exception as err:223          module.fail_json(msg='Upgrade failed. Error: [%s]' % to_native(err))224  else:225    # Paused upgrade i.e manual mode226    upgrade_status = get_upgrade_status(module, manager_url, mgr_username,227                                          mgr_password, validate_certs)228    if upgrade_status == 'NOT_STARTED':229      try:230        (rc, resp) = request(manager_url+ '/upgrade/plan?action=start', 231                     data='', headers=headers, method='POST', 232                     url_username=mgr_username, url_password=mgr_password, 233                     validate_certs=validate_certs, ignore_errors=True)234      except Exception as err:235        module.fail_json(msg="Failed while upgrading. Error[%s]." % to_native(err))236    else:237      try:238        (rc, resp) = request(manager_url+ '/upgrade/plan?action=continue', 239                     data='', headers=headers, method='POST', 240                     url_username=mgr_username, url_password=mgr_password, 241                     validate_certs=validate_certs, ignore_errors=True)242      except Exception as err:243        module.fail_json(msg="Failed while upgrading. Error[%s]." % to_native(err))244    time.sleep(10)245    while True:246      try:247        can_continue, is_failed = check_continuity(module, manager_url, mgr_username,248                                                   mgr_password, validate_certs)249        decide_next_step(module, manager_url, mgr_username, mgr_password, 250                         validate_certs, can_continue, is_failed)251        if can_continue and not is_failed:252          break253        time.sleep(10)254      except Exception as err:255        module.fail_json(msg='Upgrade failed. Error: [%s]' % to_native(err))256    module.exit_json(changed=True, message='A component has been upgraded successfully.'257                                           ' Whole system is not. Please run the module'258                                           ' again till the time whole system is'259                                           ' not upgraded.')260if __name__ == '__main__':...

task.py

Source:task.py

1from src.config.logger import logger2from src.helpers.error_helper import NotFound, Conflict3import json4class Task:5    def __init__(self, trello_card_controller, zac_task_controller, routine_controller, model_helper):6        logger.info("Initializing Tasks")7        self._trello_card_controller = trello_card_controller8        self._zac_task_controller = zac_task_controller9        self._routine_controller = routine_controller10        self._model_helper = model_helper11    def get(self, user, query):12        logger.info("Initializing get tasks")13        trello_user = user["trello_user"]14        telegram_user = user["telegram_user"]15        trello_query = query16        trello_query["user"] = trello_user17        trello_cards = self._trello_card_controller.get(trello_query)18        is_conclude = query.is_conclude if "is_conclude" in query else False19        is_failed = query.is_failed if "is_failed" in query else False20        zac_query = query21        zac_query["user"] = telegram_user22        zac_query["isConclude"] = is_conclude23        zac_query["is_failed"] = is_failed24        zac_tasks = self._zac_task_controller.get(zac_query)25        cards = self._model_helper.trello_cards(trello_cards)26        tasks = self._model_helper.zac_tasks(zac_tasks)27        tasks.extend(cards)28        tasks.sort(key=lambda x: x.get("date") or x.get("due"))29        return {"tasks": tasks}30    def get_scheduled(self, user, query):31        logger.info("Initializing get scheduled tasks")32        trello_user = user["trello_user"]33        telegram_user = user["telegram_user"]34        trello_query = query35        trello_query["user"] = trello_user36        trello_cards = self._trello_card_controller.get(trello_query)37        is_conclude = query.is_conclude if "is_conclude" in query else False38        is_failed = query.is_failed if "is_failed" in query else False39        zac_query = query40        zac_query["user"] = telegram_user41        # zac_query["isConclude"] = is_conclude42        # zac_query["is_failed"] = is_failed43        zac_tasks = self._zac_task_controller.get(zac_query)44        is_active = query.is_active if "is_active" in query else True45        routine_query = query46        routine_query["user"] = telegram_user47        routine_query["days"] = json.loads(query["days"])48        routine_query["isActive"] = is_active49        zac_routines = self._routine_controller.get(routine_query)50        cards = self._model_helper.trello_cards(trello_cards)51        tasks = self._model_helper.zac_tasks(zac_tasks)52        routines = self._model_helper.zac_routines(zac_routines)53        tasks.extend(cards)54        tasks.sort(key=lambda x: x.get("date") or x.get("due"))55        return {"tasks": tasks, "routines": routines}56    def conclude(self, user, id):57        logger.info("Initializing conclude task")58        telegram_user = user["telegram_user"]59        filter = { "id": id, "user": telegram_user }60        tasks = self._zac_task_controller.get(filter)61        if len(tasks) == 0:62            logger.error("Task not found")63            raise NotFound("Task not found")64        task = tasks[0]65        if task["isConclude"] or ("is_failed" in task and task["is_failed"]):66            logger.error("Task is already concluded or failed")67            raise Conflict("Task is already concluded or failed")68        self._zac_task_controller.update(filter, {"isConclude": True})69    def fail(self, user, id):70        logger.info("Initializing fail task")71        telegram_user = user["telegram_user"]72        filter = { "id": id, "user": telegram_user }73        tasks = self._zac_task_controller.get(filter)74        if len(tasks) == 0:75            logger.error("Task not found")76            raise NotFound("Task not found")77        task = tasks[0]78        if task["isConclude"] or ("is_failed" in task and task["is_failed"]):79            logger.error("Task is already concluded or failed")80            raise Conflict("Task is already concluded or failed")...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.