KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _run_ansible_module method in avocado

Best Python code snippet using avocado_python

service.py

Source:service.py

...96        pattern = ','.join(self.inventory_groups)97        return set([98            host.name for host in self.environment.inventory_manager.get_hosts(pattern)99        ])100    def _run_ansible_module(self, host_pattern, module, module_args):101        return run_ansible_module(102            self.environment,103            self.ansible_context,104            host_pattern,105            module,106            module_args,107            True,108            None,109            False,110        )111class SubServicesMixin(six.with_metaclass(ABCMeta)):112    @abstractproperty113    def managed_services(self):114        """115        :return: List of sub-services managed by this class. e.g. Celery workers, Pillowtop processes116        """117        raise NotImplementedError118    def get_options(self):119        options = super(SubServicesMixin, self).get_options()120        managed_services = self.managed_services121        if managed_services:122            options["Sub-services (use with --only)"] = [ServiceOption(service) for service in managed_services]123        return options124class SupervisorService(SubServicesMixin, ServiceBase):125    def execute_action(self, action, host_pattern=None, process_pattern=None):126        if host_pattern:127            self.environment.inventory_manager.subset(host_pattern)128        process_host_mapping = self._get_processes_by_host(process_pattern)129        if not process_host_mapping:130            raise NoHostsMatch131        all_processes_by_host = get_all_supervisor_processes_by_host(self.environment)132        process_host_mapping = optimize_process_operations(133            all_processes_by_host, process_host_mapping134        )135        non_zero_exits = []136        for hosts, processes in process_host_mapping.items():137            if action == 'status' and 'all' in processes:138                command = 'supervisorctl status'139            else:140                command = 'supervisorctl {} {}'.format(141                    action,142                    ' '.join(processes)143                )144            exit_code = self._run_ansible_module(145                ','.join(hosts),146                'shell',147                command148            )149            if exit_code != 0:150                non_zero_exits.append(exit_code)151        return non_zero_exits[0] if non_zero_exits else 0152    @abstractmethod153    def _get_processes_by_host(self, process_pattern=None):154        """155        Given the process pattern return the matching processes and hosts.156        :param process_pattern: process pattern from the args or None157        :return: dict mapping hostname -> [process name list]158        """159        raise NotImplementedError160def _service_status_helper(service_name):161    if service_name in MONIT_MANAGED_SERVICES:162        return 'monit status {}'.format(service_name)163    return 'service {} status'.format(service_name)164def _ansible_module_helper(service_name):165    if service_name in MONIT_MANAGED_SERVICES:166        return 'monit'167    return 'service'168class AnsibleService(ServiceBase):169    """Service that is controlled via the ansible 'service' module"""170    @property171    def service_name(self):172        """Override if different"""173        return self.name174    def execute_action(self, action, host_pattern=None, process_pattern=None):175        if host_pattern:176            self.environment.inventory_manager.subset(self.inventory_groups)177            hosts = self.environment.inventory_manager.get_hosts(host_pattern)178            if not hosts:179                raise NoHostsMatch180        host_pattern = host_pattern or ','.join(self.inventory_groups)181        if action == 'status':182            command = _service_status_helper(self.service_name)183            return self._run_ansible_module(host_pattern, 'shell', command)184        service_args = 'name={} state={}'.format(self.service_name, STATES[action])185        return self._run_ansible_module(host_pattern, _ansible_module_helper(self.service_name), service_args)186class MultiAnsibleService(SubServicesMixin, AnsibleService):187    """Service that is made up of multiple other services e.g. CouchDB2"""188    @abstractproperty189    def service_process_mapping(self):190        """191        Return a mapping of service names (as passed in by the user) to192        a tuple of ('linux_service_name', 'inventory_group1,inventory_group2')193        """194        raise NotImplementedError195    @property196    def inventory_groups(self):197        return [198            group199            for service, inventory_group in self.service_process_mapping.values()200            for group in inventory_group.split(',')201        ]202    @property203    def managed_services(self):204        return [name for name in self.service_process_mapping]205    def check_status(self, host_pattern=None, process_pattern=None):206        def _status(service_name, run_on):207            command = _service_status_helper(service_name)208            return self._run_ansible_module(run_on, 'shell', command)209        return self._run_action_on_hosts(_status, host_pattern, process_pattern)210    def execute_action(self, action, host_pattern=None, process_pattern=None):211        if action == 'status':212            return self.check_status(host_pattern, process_pattern)213        def _change_state(service_name, run_on, action=action):214            service_args = 'name={} state={}'.format(service_name, STATES[action])215            return self._run_ansible_module(run_on, _ansible_module_helper(service_name), service_args)216        return self._run_action_on_hosts(_change_state, host_pattern, process_pattern)217    def _run_action_on_hosts(self, action_fn, host_pattern, process_pattern):218        if host_pattern:219            self.environment.inventory_manager.subset(host_pattern)220        non_zero_exits = []221        ran = False222        for service, run_on in self._get_service_host_groups(process_pattern):223            hosts = self.environment.inventory_manager.get_hosts(run_on)224            run_on = ','.join([host.name for host in hosts]) if host_pattern else run_on225            if hosts:226                ran = True227                exit_code = action_fn(service, run_on)228                if exit_code != 0:229                    non_zero_exits.append(exit_code)...

module.py

Source:module.py

...13     * kwargs: passed as arguments to ansible module14    """15    name = "ansible-module"16    description = f"Runner for dependencies of type {name}"17    def _run_ansible_module(self, runnable, queue):18        args = " ".join([f"{k}={v}" for k, v in runnable.kwargs.items()])19        if args:20            args_cmdline = f"-a '{args}'"21        else:22            args_cmdline = ""23        try:24            proc_result = run(f"ansible -m {runnable.uri} {args_cmdline} localhost")25            result = "pass"26            stdout = proc_result.stdout27            stderr = proc_result.stderr28        except CmdError as e:29            result = "fail"30            stdout = ""31            stderr = str(e)...

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.