How to use _add_host method in tempest

Best Python code snippet using tempest_python

dense.py

Source:dense.py Github

copy

Full Screen

...159 display.warning("The 'dense' callback plugin requires OrderedDict which is not available in this version of python, disabling.")160 self.disabled = True161 def __del__(self):162 sys.stdout.write(vt100.restore + vt100.reset + '\n' + vt100.save + vt100.clearline)163 def _add_host(self, result, status):164 name = result._host.get_name()165 # Add a new status in case a failed task is ignored166 if status == 'failed' and result._task.ignore_errors:167 status = 'ignored'168 # Check if we have to update an existing state (when looping over items)169 if name not in self.hosts:170 self.hosts[name] = dict(state=status)171 elif states.index(self.hosts[name]['state']) < states.index(status):172 self.hosts[name]['state'] = status173 # Store delegated hostname, if needed174 delegated_vars = result._result.get('_ansible_delegated_vars', None)175 if delegated_vars:176 self.hosts[name]['delegate'] = delegated_vars['ansible_host']177 # Print progress bar178 self._display_progress(result)179# # Ensure that tasks with changes/failures stay on-screen, and during diff-mode180# if status in ['changed', 'failed', 'unreachable'] or (result.get('_diff_mode', False) and result._resultget('diff', False)):181 # Ensure that tasks with changes/failures stay on-screen182 if status in ['changed', 'failed', 'unreachable']:183 self.keep = True184 if self._display.verbosity == 1:185 # Print task title, if needed186 self._display_task_banner()187 self._display_results(result, status)188 def _clean_results(self, result):189 # Remove non-essential attributes190 for attr in self.removed_attributes:191 if attr in result:192 del(result[attr])193 # Remove empty attributes (list, dict, str)194 for attr in result.copy():195 if isinstance(result[attr], (MutableSequence, MutableMapping, binary_type, text_type)):196 if not result[attr]:197 del(result[attr])198 def _handle_exceptions(self, result):199 if 'exception' in result:200 # Remove the exception from the result so it's not shown every time201 del result['exception']202 if self._display.verbosity == 1:203 return "An exception occurred during task execution. To see the full traceback, use -vvv."204 def _display_progress(self, result=None):205 # Always rewrite the complete line206 sys.stdout.write(vt100.restore + vt100.reset + vt100.clearline + vt100.nolinewrap + vt100.underline)207 sys.stdout.write('%s %d:' % (self.type, self.count[self.type]))208 sys.stdout.write(vt100.reset)209 sys.stdout.flush()210 # Print out each host in its own status-color211 for name in self.hosts:212 sys.stdout.write(' ')213 if self.hosts[name].get('delegate', None):214 sys.stdout.write(self.hosts[name]['delegate'] + '>')215 sys.stdout.write(colors[self.hosts[name]['state']] + name + vt100.reset)216 sys.stdout.flush()217# if result._result.get('diff', False):218# sys.stdout.write('\n' + vt100.linewrap)219 sys.stdout.write(vt100.linewrap)220# self.keep = True221 def _display_task_banner(self):222 if not self.shown_title:223 self.shown_title = True224 sys.stdout.write(vt100.restore + vt100.reset + vt100.clearline + vt100.underline)225 sys.stdout.write('%s %d: %s' % (self.type, self.count[self.type], self.task.get_name().strip()))226 sys.stdout.write(vt100.restore + vt100.reset + '\n' + vt100.save + vt100.clearline)227 sys.stdout.flush()228 else:229 sys.stdout.write(vt100.restore + vt100.reset + vt100.clearline)230 self.keep = False231 def _display_results(self, result, status):232 # Leave the previous task on screen (as it has changes/errors)233 if self._display.verbosity == 0 and self.keep:234 sys.stdout.write(vt100.restore + vt100.reset + '\n' + vt100.save + vt100.clearline)235 else:236 sys.stdout.write(vt100.restore + vt100.reset + vt100.clearline)237 self.keep = False238 self._clean_results(result._result)239 dump = ''240 if result._task.action == 'include':241 return242 elif status == 'ok':243 return244 elif status == 'ignored':245 dump = self._handle_exceptions(result._result)246 elif status == 'failed':247 dump = self._handle_exceptions(result._result)248 elif status == 'unreachable':249 dump = result._result['msg']250 if not dump:251 dump = self._dump_results(result._result)252 if result._task.loop and 'results' in result._result:253 self._process_items(result)254 else:255 sys.stdout.write(colors[status] + status + ': ')256 delegated_vars = result._result.get('_ansible_delegated_vars', None)257 if delegated_vars:258 sys.stdout.write(vt100.reset + result._host.get_name() + '>' + colors[status] + delegated_vars['ansible_host'])259 else:260 sys.stdout.write(result._host.get_name())261 sys.stdout.write(': ' + dump + '\n')262 sys.stdout.write(vt100.reset + vt100.save + vt100.clearline)263 sys.stdout.flush()264 if status == 'changed':265 self._handle_warnings(result._result)266 def v2_playbook_on_play_start(self, play):267 # Leave the previous task on screen (as it has changes/errors)268 if self._display.verbosity == 0 and self.keep:269 sys.stdout.write(vt100.restore + vt100.reset + '\n' + vt100.save + vt100.clearline + vt100.bold)270 else:271 sys.stdout.write(vt100.restore + vt100.reset + vt100.clearline + vt100.bold)272 # Reset at the start of each play273 self.keep = False274 self.count.update(dict(handler=0, task=0))275 self.count['play'] += 1276 self.play = play277 # Write the next play on screen IN UPPERCASE, and make it permanent278 name = play.get_name().strip()279 if not name:280 name = 'unnamed'281 sys.stdout.write('PLAY %d: %s' % (self.count['play'], name.upper()))282 sys.stdout.write(vt100.restore + vt100.reset + '\n' + vt100.save + vt100.clearline)283 sys.stdout.flush()284 def v2_playbook_on_task_start(self, task, is_conditional):285 # Leave the previous task on screen (as it has changes/errors)286 if self._display.verbosity == 0 and self.keep:287 sys.stdout.write(vt100.restore + vt100.reset + '\n' + vt100.save + vt100.clearline + vt100.underline)288 else:289 # Do not clear line, since we want to retain the previous output290 sys.stdout.write(vt100.restore + vt100.reset + vt100.underline)291 # Reset at the start of each task292 self.keep = False293 self.shown_title = False294 self.hosts = OrderedDict()295 self.task = task296 self.type = 'task'297 # Enumerate task if not setup (task names are too long for dense output)298 if task.get_name() != 'setup':299 self.count['task'] += 1300 # Write the next task on screen (behind the prompt is the previous output)301 sys.stdout.write('%s %d.' % (self.type, self.count[self.type]))302 sys.stdout.write(vt100.reset)303 sys.stdout.flush()304 def v2_playbook_on_handler_task_start(self, task):305 # Leave the previous task on screen (as it has changes/errors)306 if self._display.verbosity == 0 and self.keep:307 sys.stdout.write(vt100.restore + vt100.reset + '\n' + vt100.save + vt100.clearline + vt100.underline)308 else:309 sys.stdout.write(vt100.restore + vt100.reset + vt100.clearline + vt100.underline)310 # Reset at the start of each handler311 self.keep = False312 self.shown_title = False313 self.hosts = OrderedDict()314 self.task = task315 self.type = 'handler'316 # Enumerate handler if not setup (handler names may be too long for dense output)317 if task.get_name() != 'setup':318 self.count[self.type] += 1319 # Write the next task on screen (behind the prompt is the previous output)320 sys.stdout.write('%s %d.' % (self.type, self.count[self.type]))321 sys.stdout.write(vt100.reset)322 sys.stdout.flush()323 def v2_playbook_on_cleanup_task_start(self, task):324 # TBD325 sys.stdout.write('cleanup.')326 sys.stdout.flush()327 def v2_runner_on_failed(self, result, ignore_errors=False):328 self._add_host(result, 'failed')329 def v2_runner_on_ok(self, result):330 if result._result.get('changed', False):331 self._add_host(result, 'changed')332 else:333 self._add_host(result, 'ok')334 def v2_runner_on_skipped(self, result):335 self._add_host(result, 'skipped')336 def v2_runner_on_unreachable(self, result):337 self._add_host(result, 'unreachable')338 def v2_runner_on_include(self, included_file):339 pass340 def v2_runner_on_file_diff(self, result, diff):341 sys.stdout.write(vt100.bold)342 self.super_ref.v2_runner_on_file_diff(result, diff)343 sys.stdout.write(vt100.reset)344 def v2_on_file_diff(self, result):345 sys.stdout.write(vt100.bold)346 self.super_ref.v2_on_file_diff(result)347 sys.stdout.write(vt100.reset)348 # Old definition in v2.0349 def v2_playbook_item_on_ok(self, result):350 self.v2_runner_item_on_ok(result)351 def v2_runner_item_on_ok(self, result):352 if result._result.get('changed', False):353 self._add_host(result, 'changed')354 else:355 self._add_host(result, 'ok')356 # Old definition in v2.0357 def v2_playbook_item_on_failed(self, result):358 self.v2_runner_item_on_failed(result)359 def v2_runner_item_on_failed(self, result):360 self._add_host(result, 'failed')361 # Old definition in v2.0362 def v2_playbook_item_on_skipped(self, result):363 self.v2_runner_item_on_skipped(result)364 def v2_runner_item_on_skipped(self, result):365 self._add_host(result, 'skipped')366 def v2_playbook_on_no_hosts_remaining(self):367 if self._display.verbosity == 0 and self.keep:368 sys.stdout.write(vt100.restore + vt100.reset + '\n' + vt100.save + vt100.clearline)369 else:370 sys.stdout.write(vt100.restore + vt100.reset + vt100.clearline)371 self.keep = False372 sys.stdout.write(vt100.white + vt100.redbg + 'NO MORE HOSTS LEFT')373 sys.stdout.write(vt100.restore + vt100.reset + '\n' + vt100.save + vt100.clearline)374 sys.stdout.flush()375 def v2_playbook_on_include(self, included_file):376 pass377 def v2_playbook_on_stats(self, stats):378 if self._display.verbosity == 0 and self.keep:379 sys.stdout.write(vt100.restore + vt100.reset + '\n' + vt100.save + vt100.clearline)...

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 tempest 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