KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference
How to use _add_host method in tempest

Best Python code snippet using tempest_python
dense.py
Source:dense.py
...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)...
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.