Best Python code snippet using avocado_python
multipath_test.py
Source:multipath_test.py  
...100        """101        Tests Multipath.102        """103        msg = ""104        multipath.form_conf_mpath_file()105        plcy = "path_selector \"%s 0\"" % self.policy106        multipath.form_conf_mpath_file(defaults_extra=plcy)107        for path_dic in self.mpath_list:108            operation = ''109            self.log.debug("operating on paths: %s", path_dic["paths"])110            # Path Selector policy111            self.log.info("changing Selector policy")112            for policy in self.policies:113                cmd = "path_selector \"%s 0\"" % policy114                multipath.form_conf_mpath_file(defaults_extra=cmd)115                time.sleep(5)116                if multipath.get_policy(path_dic["wwid"]) != policy:117                    msg += "%s for %s fails\n" % (policy, path_dic["wwid"])118            def is_mpath_available():119                if operation == 'block':120                    if multipath.device_exists(path_dic["wwid"]) is True:121                        self.log.info("%s=False, not blocked" % test_mpath)122                        return False123                    return True124                elif operation == 'recover':125                    if multipath.device_exists(path_dic["wwid"]) is False:126                        self.log.info("%s=False, not recovered" % test_mpath)127                        return False128                    return True129            # mutipath -f mpathX130            test_mpath = path_dic["name"]131            self.log.info("flushing %s" % path_dic["name"])132            operation = 'block'133            multipath.flush_path(path_dic["name"])134            if wait.wait_for(is_mpath_available, timeout=10):135                self.log.info("flush of %s success" % path_dic["name"])136            else:137                msg += "Flush of %s fails\n" % path_dic["name"]138            self.log.info("Recovering %s after flush.." % path_dic["name"])139            operation = 'recover'140            self.mpath_svc.restart()141            wait.wait_for(self.mpath_svc.status, timeout=10)142            if wait.wait_for(is_mpath_available, timeout=10):143                self.log.info("recovery of %s success after \144                               flush" % path_dic["name"])145            else:146                msg += "Recovery of %s fails after flush\n" % path_dic["name"]147            # Blacklisting wwid148            test_mpath = path_dic["wwid"]149            self.log.info("Black listing WWIDs")150            cmd = "wwid %s" % path_dic["wwid"]151            multipath.form_conf_mpath_file(blacklist=cmd, defaults_extra=plcy)152            operation = 'block'153            if wait.wait_for(is_mpath_available, timeout=10):154                self.log.info("Blocklist of %s success" % path_dic["wwid"])155            else:156                msg += "Blacklist of %s fails\n" % path_dic["wwid"]157            operation = 'recover'158            self.log.info("Recovering WWID after Black list.....")159            multipath.form_conf_mpath_file(defaults_extra=plcy)160            if wait.wait_for(is_mpath_available, timeout=10):161                self.log.info("recovery of %s success" % path_dic["wwid"])162            else:163                msg += "Recovery of %s fails after blocklist\n" \164                        % path_dic["wwid"]165            def is_path_available():166                if operation == 'block':167                    if multipath.device_exists(test_path) is True:168                        return False169                    return True170                elif operation == 'recover':171                    if multipath.device_exists(test_path) is False:172                        self.log.info("%s=False still not recover" % test_path)173                        return False174                    return True175            # Blacklisting sdX176            self.log.info("Black listing individual paths")177            test_path = ''178            for disk in path_dic["paths"]:179                test_path = disk180                self.log.info("Black_listing %s" % disk)181                operation = 'block'182                cmd = "devnode %s" % disk183                multipath.form_conf_mpath_file(blacklist=cmd,184                                               defaults_extra=plcy)185                if wait.wait_for(is_path_available, timeout=10):186                    self.log.info("block list of %s success" % disk)187                else:188                    msg += "block list of %s fails\n" % disk189                operation = 'recover'190                multipath.form_conf_mpath_file(defaults_extra=plcy)191                if wait.wait_for(is_path_available, timeout=10):192                    self.log.info("recovery of %s success" % disk)193                else:194                    msg += "Recovery of %s fails after blacklist %s\n" \195                            % (disk, path_dic["wwid"])196            multipath.form_conf_mpath_file(defaults_extra=plcy)197        if msg:198            self.fail("Some tests failed. Find details below:\n%s" % msg)199    def test_fail_reinstate_individual_paths(self):200        '''201        Failing and reinstating individual paths eg: sdX202        '''203        err_paths = []204        self.log.info(" Failing and reinstating the individual paths")205        for dic_path in self.mpath_list:206            for path in dic_path['paths']:207                if multipath.fail_path(path) is False:208                    self.log.info("could not fail %s in indvdl path:", path)209                    err_paths.append(path)210                elif multipath.reinstate_path(path) is False:...multipath.py
Source:multipath.py  
...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:...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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!
