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

How to use get_svc_name method in avocado

Best Python code snippet using avocado_python

multipath.py

Source:multipath.py

...21from . import distro22from . import process23from . import service24from . import wait25def get_svc_name():26    """27    Gets the multipath service name based on distro.28    """29    if distro.detect().name == 'Ubuntu':30        return "multipath-tools"31    return "multipathd"32def form_conf_mpath_file(blacklist="", defaults_extra=""):33    """34    Form a multipath configuration file, and restart multipath service.35    :param blacklist: Entry in conf file to indicate blacklist section.36    :param defaults_extra: Extra entry in conf file in defaults section.37    """38    conf_file = "/etc/multipath.conf"39    with open(conf_file, "w") as mpath_fp:40        mpath_fp.write("defaults {\n")41        mpath_fp.write("    find_multipaths yes\n")42        mpath_fp.write("    user_friendly_names yes\n")43        if defaults_extra:44            mpath_fp.write("    %s\n" % defaults_extra)45        mpath_fp.write("}\n")46        if blacklist:47            mpath_fp.write("blacklist {\n")48            mpath_fp.write("    %s\n" % blacklist)49            mpath_fp.write("}\n")50    logging.debug(open(conf_file, "r").read())51    # The reason for sleep here is to give some time for change in52    # multipath.conf file to take effect.53    time.sleep(5)54    mpath_svc = service.SpecificServiceManager(get_svc_name())55    mpath_svc.restart()56    wait.wait_for(mpath_svc.status, timeout=10)57def device_exists(path):58    """59    Checks if a given path exists.60    :return: True if path exists, False if does not exist.61    """62    cmd = "multipath -l %s" % path63    if process.system(cmd, ignore_status=True, sudo=True) != 0:64        return False65    return True66def get_mpath_name(wwid):67    """68    Get multipath name for a given wwid....

PyTorchServingHandler.py

Source:PyTorchServingHandler.py

...63        if model_node is not None:64            up_models = [node.model_name for node in model_node.up_models]65        up_model_serving_svc = {}66        for model_name in up_models:67            up_model_serving_svc[model_name] = get_svc_name(68                self.project_name, self.task_name, model_name)69        upnodes = list(up_model_serving_svc.keys())70        responses = asyncMsg([71            "http://{}:{}/predictions/{}".format(72                up_model_serving_svc[model_name], inference_port, model_name)73            for model_name in upnodes74        ], data, 3, False, False)75        return upnodes, responses76    def handle(self, data, context):77        """è¿è¡modelçé¢æµ78        1. æ ¹æ®æ°æ®åºæ¥è¯¢ä¸æ¸¸æ¨¡å79        2. è¥æï¼åè·åä¸æ¸¸æ¨¡åçè¾åº80        3. å°jsonè§£æçæ°æ®ä¸ä¸æ¸¸æ¨¡åçè¾åºensemble_outoutç»å°servingå½æ°81        """...

Systemd.py

Source:Systemd.py

...9    name = 'Systemd'10    __execs__ = ['/bin/systemctl']11    __handles__ = [('Service', 'systemd')]12    __req__ = {'Service': ['name', 'status']}13    def get_svc_name(self, service):14        """Append .service to name if name doesn't specify a unit type."""15        svc = service.get('name')16        if svc.endswith(('.service', '.socket', '.device', '.mount',17                         '.automount', '.swap', '.target', '.path',18                         '.timer', '.snapshot', '.slice', '.scope')):19            return svc20        else:21            return '%s.service' % svc22    def get_svc_command(self, service, action):23        return "/bin/systemctl %s %s" % (action, self.get_svc_name(service))24    def VerifyService(self, entry, _):25        """Verify Service status for entry."""26        entry.set('target_status', entry.get('status'))  # for reporting27        bootstatus = self.get_bootstatus(entry)28        if bootstatus is None:29            # bootstatus is unspecified and status is ignore30            return True31        if self.cmd.run(self.get_svc_command(entry, 'is-enabled')):32            current_bootstatus = 'on'33        else:34            current_bootstatus = 'off'35        if entry.get('status') == 'ignore':36            return current_bootstatus == bootstatus37        cmd = self.get_svc_command(entry, 'show') + ' -p ActiveState'38        rv = self.cmd.run(cmd)39        if rv.stdout.strip() in ('ActiveState=active',40                                 'ActiveState=activating',41                                 'ActiveState=reloading'):42            current_status = 'on'43        else:44            current_status = 'off'45        entry.set('current_status', current_status)46        return (entry.get('status') == current_status and47                bootstatus == current_bootstatus)48    def InstallService(self, entry):49        """Install Service entry."""50        self.logger.info("Installing Service %s" % (entry.get('name')))51        bootstatus = self.get_bootstatus(entry)52        if bootstatus is None:53            # bootstatus is unspecified and status is ignore54            return True55        # Enable or disable the service56        if bootstatus == 'on':57            cmd = self.get_svc_command(entry, 'enable')58        else:59            cmd = self.get_svc_command(entry, 'disable')60        if not self.cmd.run(cmd).success:61            # Return failure immediately and do not start/stop the service.62            return False63        # Start or stop the service, depending on the current service_mode64        cmd = None65        if Bcfg2.Options.setup.service_mode == 'disabled':66            # 'disabled' means we don't attempt to modify running svcs67            pass68        elif Bcfg2.Options.setup.service_mode == 'build':69            # 'build' means we attempt to stop all services started70            if entry.get('current_status') == 'on':71                cmd = self.get_svc_command(entry, 'stop')72        else:73            if entry.get('status') == 'on':74                cmd = self.get_svc_command(entry, 'start')75            elif entry.get('status') == 'off':76                cmd = self.get_svc_command(entry, 'stop')77        if cmd:78            return self.cmd.run(cmd).success79        else:80            return True81    def FindExtra(self):82        """Find Extra Systemd Service entries."""83        specified = [self.get_svc_name(entry)84                     for entry in self.getSupportedEntries()]85        extra = set()86        for fname in glob.glob("/etc/systemd/system/*.wants/*"):87            name = os.path.basename(fname)88            if name not in specified:89                extra.add(name)90        return [Bcfg2.Client.XML.Element('Service', name=name, type='systemd')...

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.