Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use get_vendor_from_pci_id method in autotest

Best Python code snippet using autotest_python

kvm_utils.py

Source:kvm_utils.py

...784    status, full_id = commands.getstatusoutput(cmd)785    if status != 0:786        return None787    return full_id788def get_vendor_from_pci_id(pci_id):789    """790    Check out the device vendor ID according to pci_id.791    @param pci_id: PCI ID of a device.792    """793    cmd = "lspci -n | awk '/%s/ {print $3}'" % pci_id794    return re.sub(":", " ", commands.getoutput(cmd))795class KvmLoggingConfig(logging_config.LoggingConfig):796    """797    Used with the sole purpose of providing convenient logging setup798    for the KVM test auxiliary programs.799    """800    def configure_logging(self, results_dir=None, verbose=False):801        super(KvmLoggingConfig, self).configure_logging(use_console=True,802                                                        verbose=verbose)803class PciAssignable(object):804    """805    Request PCI assignable devices on host. It will check whether to request806    PF (physical Functions) or VF (Virtual Functions).807    """808    def __init__(self, type="vf", driver=None, driver_option=None,809                 names=None, devices_requested=None):810        """811        Initialize parameter 'type' which could be:812        vf: Virtual Functions813        pf: Physical Function (actual hardware)814        mixed:  Both includes VFs and PFs815        If pass through Physical NIC cards, we need to specify which devices816        to be assigned, e.g. 'eth1 eth2'.817        If pass through Virtual Functions, we need to specify how many vfs818        are going to be assigned, e.g. passthrough_count = 8 and max_vfs in819        config file.820        @param type: PCI device type.821        @param driver: Kernel module for the PCI assignable device.822        @param driver_option: Module option to specify the maximum number of823                VFs (eg 'max_vfs=7')824        @param names: Physical NIC cards correspondent network interfaces,825                e.g.'eth1 eth2 ...'826        @param devices_requested: Number of devices being requested.827        """828        self.type = type829        self.driver = driver830        self.driver_option = driver_option831        if names:832            self.name_list = names.split()833        if devices_requested:834            self.devices_requested = int(devices_requested)835        else:836            self.devices_requested = None837    def _get_pf_pci_id(self, name, search_str):838        """839        Get the PF PCI ID according to name.840        @param name: Name of the PCI device.841        @param search_str: Search string to be used on lspci.842        """843        cmd = "ethtool -i %s | awk '/bus-info/ {print $2}'" % name844        s, pci_id = commands.getstatusoutput(cmd)845        if not (s or "Cannot get driver information" in pci_id):846            return pci_id[5:]847        cmd = "lspci | awk '/%s/ {print $1}'" % search_str848        pci_ids = [id for id in commands.getoutput(cmd).splitlines()]849        nic_id = int(re.search('[0-9]+', name).group(0))850        if (len(pci_ids) - 1) < nic_id:851            return None852        return pci_ids[nic_id]853    def _release_dev(self, pci_id):854        """855        Release a single PCI device.856        @param pci_id: PCI ID of a given PCI device.857        """858        base_dir = "/sys/bus/pci"859        full_id = get_full_pci_id(pci_id)860        vendor_id = get_vendor_from_pci_id(pci_id)861        drv_path = os.path.join(base_dir, "devices/%s/driver" % full_id)862        if 'pci-stub' in os.readlink(drv_path):863            cmd = "echo '%s' > %s/new_id" % (vendor_id, drv_path)864            if os.system(cmd):865                return False866            stub_path = os.path.join(base_dir, "drivers/pci-stub")867            cmd = "echo '%s' > %s/unbind" % (full_id, stub_path)868            if os.system(cmd):869                return False870            driver = self.dev_drivers[pci_id]871            cmd = "echo '%s' > %s/bind" % (full_id, driver)872            if os.system(cmd):873                return False874        return True875    def get_vf_devs(self):876        """877        Catch all VFs PCI IDs.878        @return: List with all PCI IDs for the Virtual Functions avaliable879        """880        if not self.sr_iov_setup():881            return []882        cmd = "lspci | awk '/Virtual Function/ {print $1}'"883        return commands.getoutput(cmd).split()884    def get_pf_devs(self):885        """886        Catch all PFs PCI IDs.887        @return: List with all PCI IDs for the physical hardware requested888        """889        pf_ids = []890        for name in self.name_list:891            pf_id = self._get_pf_pci_id(name, "Ethernet")892            if not pf_id:893                continue894            pf_ids.append(pf_id)895        return pf_ids896    def get_devs(self, count):897        """898        Check out all devices' PCI IDs according to their name.899        @param count: count number of PCI devices needed for pass through900        @return: a list of all devices' PCI IDs901        """902        if self.type == "vf":903            vf_ids = self.get_vf_devs()904        elif self.type == "pf":905            vf_ids = self.get_pf_devs()906        elif self.type == "mixed":907            vf_ids = self.get_vf_devs()908            vf_ids.extend(self.get_pf_devs())909        return vf_ids[0:count]910    def get_vfs_count(self):911        """912        Get VFs count number according to lspci.913        """914        # FIXME: Need to think out a method of identify which915        # 'virtual function' belongs to which physical card considering916        # that if the host has more than one 82576 card. PCI_ID?917        cmd = "lspci | grep 'Virtual Function' | wc -l"918        return int(commands.getoutput(cmd))919    def check_vfs_count(self):920        """921        Check VFs count number according to the parameter driver_options.922        """923        # Network card 82576 has two network interfaces and each can be924        # virtualized up to 7 virtual functions, therefore we multiply925        # two for the value of driver_option 'max_vfs'.926        expected_count = int((re.findall("(\d)", self.driver_option)[0])) * 2927        return (self.get_vfs_count == expected_count)928    def is_binded_to_stub(self, full_id):929        """930        Verify whether the device with full_id is already binded to pci-stub.931        @param full_id: Full ID for the given PCI device932        """933        base_dir = "/sys/bus/pci"934        stub_path = os.path.join(base_dir, "drivers/pci-stub")935        if os.path.exists(os.path.join(stub_path, full_id)):936            return True937        return False938    def sr_iov_setup(self):939        """940        Ensure the PCI device is working in sr_iov mode.941        Check if the PCI hardware device drive is loaded with the appropriate,942        parameters (number of VFs), and if it's not, perform setup.943        @return: True, if the setup was completed successfuly, False otherwise.944        """945        re_probe = False946        s, o = commands.getstatusoutput('lsmod | grep %s' % self.driver)947        if s:948            re_probe = True949        elif not self.check_vfs_count():950            os.system("modprobe -r %s" % self.driver)951            re_probe = True952        else:953            return True954        # Re-probe driver with proper number of VFs955        if re_probe:956            cmd = "modprobe %s %s" % (self.driver, self.driver_option)957            logging.info("Loading the driver '%s' with option '%s'" %958                                   (self.driver, self.driver_option))959            s, o = commands.getstatusoutput(cmd)960            if s:961                return False962            return True963    def request_devs(self):964        """965        Implement setup process: unbind the PCI device and then bind it966        to the pci-stub driver.967        @return: a list of successfully requested devices' PCI IDs.968        """969        base_dir = "/sys/bus/pci"970        stub_path = os.path.join(base_dir, "drivers/pci-stub")971        self.pci_ids = self.get_devs(self.devices_requested)972        logging.debug("The following pci_ids were found: %s", self.pci_ids)973        requested_pci_ids = []974        self.dev_drivers = {}975        # Setup all devices specified for assignment to guest976        for pci_id in self.pci_ids:977            full_id = get_full_pci_id(pci_id)978            if not full_id:979                continue980            drv_path = os.path.join(base_dir, "devices/%s/driver" % full_id)981            dev_prev_driver= os.path.realpath(os.path.join(drv_path,982                                              os.readlink(drv_path)))983            self.dev_drivers[pci_id] = dev_prev_driver984            # Judge whether the device driver has been binded to stub985            if not self.is_binded_to_stub(full_id):986                logging.debug("Binding device %s to stub", full_id)987                vendor_id = get_vendor_from_pci_id(pci_id)988                stub_new_id = os.path.join(stub_path, 'new_id')989                unbind_dev = os.path.join(drv_path, 'unbind')990                stub_bind = os.path.join(stub_path, 'bind')991                info_write_to_files = [(vendor_id, stub_new_id),992                                       (full_id, unbind_dev),993                                       (full_id, stub_bind)]994                for content, file in info_write_to_files:995                    try:996                        utils.open_write_close(file, content)997                    except IOError:998                        logging.debug("Failed to write %s to file %s", content,999                                      file)1000                        continue1001                if not self.is_binded_to_stub(full_id):...

test_setup.py

Source:test_setup.py

...357        @param pci_id: PCI ID of a given PCI device.358        """359        base_dir = "/sys/bus/pci"360        full_id = utils_misc.get_full_pci_id(pci_id)361        vendor_id = utils_misc.get_vendor_from_pci_id(pci_id)362        drv_path = os.path.join(base_dir, "devices/%s/driver" % full_id)363        if 'pci-stub' in os.readlink(drv_path):364            error.context("Release device %s to host" % pci_id, logging.info)365            driver = self.dev_unbind_drivers[pci_id]366            cmd = "echo '%s' > %s/new_id" % (vendor_id, driver)367            logging.info("Run command in host: %s" % cmd)368            if os.system(cmd):369                return False370            stub_path = os.path.join(base_dir, "drivers/pci-stub")371            cmd = "echo '%s' > %s/unbind" % (full_id, stub_path)372            logging.info("Run command in host: %s" % cmd)373            if os.system(cmd):374                return False375            driver = self.dev_unbind_drivers[pci_id]376            cmd = "echo '%s' > %s/bind" % (full_id, driver)377            logging.info("Run command in host: %s" % cmd)378            if os.system(cmd):379                return False380        return True381    def get_vf_status(self, vf_id):382        """383        Check whether one vf is assigned to VM.384        vf_id: vf id to check.385        @return: Return True if vf has already assinged to VM. Else386        return false.387        """388        base_dir = "/sys/bus/pci"389        tub_path = os.path.join(base_dir, "drivers/pci-stub")390        vf_res_path = os.path.join(tub_path, "0000\:%s/resource*" % vf_id)391        cmd = "lsof %s" % vf_res_path392        output = utils.system_output(cmd, timeout=60, ignore_status=True)393        if 'qemu' in output:394            return True395        else:396            return False397    def get_vf_devs(self):398        """399        Catch all VFs PCI IDs.400        @return: List with all PCI IDs for the Virtual Functions avaliable401        """402        if self.setup:403            if not self.sr_iov_setup():404                return []405        self.setup = None406        cmd = "lspci | awk '/%s/ {print $1}'" % self.vf_filter_re407        return utils.system_output(cmd, verbose=False).split()408    def get_pf_devs(self):409        """410        Catch all PFs PCI IDs.411        @return: List with all PCI IDs for the physical hardware requested412        """413        pf_ids = []414        for name in self.name_list:415            pf_id = self._get_pf_pci_id(name, "%s" % self.pf_filter_re)416            if not pf_id:417                continue418            pf_ids.append(pf_id)419        return pf_ids420    def get_devs(self, count, type_list=None):421        """422        Check out all devices' PCI IDs according to their name.423        @param count: count number of PCI devices needed for pass through424        @return: a list of all devices' PCI IDs425        """426        base_dir = "/sys/bus/pci"427        if type_list is None:428            type_list = self.type_list429        vf_ids = self.get_vf_devs()430        pf_ids = self.get_pf_devs()431        vf_d = []432        for pf_id in pf_ids:433            for vf_id in vf_ids:434                if vf_id[:2] == pf_id[:2] and\435                    (int(vf_id[-1]) & 1 == int(pf_id[-1])):436                    vf_d.append(vf_id)437        for vf_id in vf_ids:438            if self.get_vf_status(vf_id):439                vf_d.append(vf_id)440        for vf in vf_d:441            vf_ids.remove(vf)442        dev_ids = []443        for i in range(count):444            if type_list:445                try:446                    d_type = type_list[i]447                except IndexError:448                    d_type = "vf"449            if d_type == "vf":450                vf_id = vf_ids.pop(0)451                dev_ids.append(vf_id)452                self.dev_unbind_drivers[vf_id] = os.path.join(base_dir,453                                                 "drivers/%svf" % self.driver)454            elif d_type == "pf":455                pf_id = pf_ids.pop(0)456                dev_ids.append(pf_id)457                self.dev_unbind_drivers[pf_id] = os.path.join(base_dir,458                                                 "drivers/%s" % self.driver)459        if len(dev_ids) != count:460            logging.error("Did not get enough PCI Device")461        return dev_ids462    def get_vfs_count(self):463        """464        Get VFs count number according to lspci.465        """466        # FIXME: Need to think out a method of identify which467        # 'virtual function' belongs to which physical card considering468        # that if the host has more than one 82576 card. PCI_ID?469        cmd = "lspci | grep '%s' | wc -l" % self.vf_filter_re470        return int(utils.system_output(cmd, verbose=False))471    def check_vfs_count(self):472        """473        Check VFs count number according to the parameter driver_options.474        """475        # Network card 82576 has two network interfaces and each can be476        # virtualized up to 7 virtual functions, therefore we multiply477        # two for the value of driver_option 'max_vfs'.478        expected_count = int((re.findall("(\d)", self.driver_option)[0])) * 2479        return (self.get_vfs_count == expected_count)480    def is_binded_to_stub(self, full_id):481        """482        Verify whether the device with full_id is already binded to pci-stub.483        @param full_id: Full ID for the given PCI device484        """485        base_dir = "/sys/bus/pci"486        stub_path = os.path.join(base_dir, "drivers/pci-stub")487        if os.path.exists(os.path.join(stub_path, full_id)):488            return True489        return False490    @error.context_aware491    def sr_iov_setup(self):492        """493        Ensure the PCI device is working in sr_iov mode.494        Check if the PCI hardware device drive is loaded with the appropriate,495        parameters (number of VFs), and if it's not, perform setup.496        @return: True, if the setup was completed successfuly, False otherwise.497        """498        # Check if the host support interrupt remapping499        error.context("Set up host env for PCI assign test", logging.info)500        kvm_re_probe = False501        o = utils.system_output("cat /var/log/dmesg")502        ecap = re.findall("ecap\s+(.\w+)", o)503        if ecap and int(ecap[0], 16) & 8 == 0:504            if self.kvm_params is not None:505                if self.auai_path and self.kvm_params[self.auai_path] == "N":506                    kvm_re_probe = True507            else:508                kvm_re_probe = True509        # Try to re probe kvm module with interrupt remapping support510        if kvm_re_probe:511            kvm_arch = kvm_control.get_kvm_arch()512            utils.system("modprobe -r %s" % kvm_arch)513            utils.system("modprobe -r kvm")514            cmd = "modprobe kvm allow_unsafe_assigned_interrupts=1"515            if self.kvm_params is not None:516                for i in self.kvm_params:517                    if "allow_unsafe_assigned_interrupts" not in i:518                        if self.kvm_params[i] == "Y":519                            params_name = os.path.split(i)[1]520                            cmd += " %s=1" % params_name521            error.context("Loading kvm with: %s" % cmd, logging.info)522            try:523                utils.system(cmd)524            except Exception:525                logging.debug("Can not enable the interrupt remapping support")526            utils.system("modprobe %s" % kvm_arch)527        re_probe = False528        s, o = commands.getstatusoutput('lsmod | grep %s' % self.driver)529        if s:530            re_probe = True531        elif not self.check_vfs_count():532            os.system("modprobe -r %s" % self.driver)533            re_probe = True534        else:535            return True536        # Re-probe driver with proper number of VFs537        if re_probe:538            cmd = "modprobe %s %s" % (self.driver, self.driver_option)539            error.context("Loading the driver '%s' with command '%s'" %\540                          (self.driver, cmd), logging.info)541            s, o = commands.getstatusoutput(cmd)542            utils.system("/etc/init.d/network restart", ignore_status=True)543            if s:544                return False545            return True546    def sr_iov_cleanup(self):547        """548        Clean up the sriov setup549        Check if the PCI hardware device drive is loaded with the appropriate,550        parameters (none of VFs), and if it's not, perform cleanup.551        @return: True, if the setup was completed successfuly, False otherwise.552        """553        # Check if the host support interrupt remapping554        error.context("Clean up host env after PCI assign test", logging.info)555        kvm_re_probe = False556        if self.kvm_params is not None:557            if (self.auai_path and558               open(self.auai_path, "r").read().strip() == "Y"):559                if self.kvm_params and self.kvm_params[self.auai_path] == "N":560                    kvm_re_probe = True561        else:562            kvm_re_probe = True563        # Try to re probe kvm module with interrupt remapping support564        if kvm_re_probe:565            kvm_arch = kvm_control.get_kvm_arch()566            utils.system("modprobe -r %s" % kvm_arch)567            utils.system("modprobe -r kvm")568            cmd = "modprobe kvm"569            if self.kvm_params:570                for i in self.kvm_params:571                    if self.kvm_params[i] == "Y":572                        params_name = os.path.split(i)[1]573                        cmd += " %s=1" % params_name574            logging.info("Loading kvm with command: %s" % cmd)575            try:576                utils.system(cmd)577            except Exception:578                logging.debug("Failed to reload kvm")579            cmd = "modprobe %s" % kvm_arch580            logging.info("Loading %s with command: %s" % (kvm_arch, cmd))581            utils.system(cmd)582        re_probe = False583        s = commands.getstatusoutput('lsmod | grep %s' % self.driver)[0]584        if s:585            cmd = "modprobe -r %s" % self.driver586            logging.info("Running host command: %s" % cmd)587            os.system(cmd)588            re_probe = True589        else:590            return True591        # Re-probe driver with proper number of VFs592        if re_probe:593            cmd = "modprobe %s" % self.driver594            msg = "Loading the driver '%s' without option" % self.driver595            error.context(msg, logging.info)596            s = commands.getstatusoutput(cmd)[0]597            utils.system("/etc/init.d/network restart", ignore_status=True)598            if s:599                return False600            return True601    def request_devs(self, count=None):602        """603        Implement setup process: unbind the PCI device and then bind it604        to the pci-stub driver.605        @param count: count number of PCI devices needed for pass through606        @return: a list of successfully requested devices' PCI IDs.607        """608        if count is None:609            count = self.devices_requested610        base_dir = "/sys/bus/pci"611        stub_path = os.path.join(base_dir, "drivers/pci-stub")612        pci_ids = self.get_devs(count)613        logging.info("The following pci_ids were found: %s", pci_ids)614        requested_pci_ids = []615        # Setup all devices specified for assignment to guest616        for pci_id in pci_ids:617            full_id = utils_misc.get_full_pci_id(pci_id)618            if not full_id:619                continue620            drv_path = os.path.join(base_dir, "devices/%s/driver" % full_id)621            dev_prev_driver = os.path.realpath(os.path.join(drv_path,622                                               os.readlink(drv_path)))623            self.dev_drivers[pci_id] = dev_prev_driver624            # Judge whether the device driver has been binded to stub625            if not self.is_binded_to_stub(full_id):626                error.context("Bind device %s to stub" % full_id, logging.info)627                vendor_id = utils_misc.get_vendor_from_pci_id(pci_id)628                stub_new_id = os.path.join(stub_path, 'new_id')629                unbind_dev = os.path.join(drv_path, 'unbind')630                stub_bind = os.path.join(stub_path, 'bind')631                info_write_to_files = [(vendor_id, stub_new_id),632                                       (full_id, unbind_dev),633                                       (full_id, stub_bind)]634                for content, file in info_write_to_files:635                    try:636                        utils.open_write_close(file, content)637                    except IOError:638                        logging.debug("Failed to write %s to file %s", content,639                                      file)640                        continue641                if not self.is_binded_to_stub(full_id):...

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.