How to use is_failed method in avocado

Best Python code snippet using avocado_python

nsxt_upgrade_run.py

Source:nsxt_upgrade_run.py Github

copy

Full Screen

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

Full Screen

Full Screen

task.py

Source:task.py Github

copy

Full Screen

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")...

Full Screen

Full Screen

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.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run avocado automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful