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 Github

copy

Full Screen

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

Full Screen

Full Screen

test_setup.py

Source:test_setup.py Github

copy

Full Screen

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

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:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run autotest 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?

Helpful

NotHelpful