How to use _run_ansible_module method in avocado

Best Python code snippet using avocado_python Github


Full Screen

...96 pattern = ','.join(self.inventory_groups)97 return set([98 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([ 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)...

Full Screen

Full Screen Github


Full Screen

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

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:


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?