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

How to use get_lower_nics method in lisa

Best Python code snippet using lisa_python

nic.py

Source:nic.py

...125    def get_unpaired_devices(self) -> List[str]:126        return [x.upper for x in self.nics.values() if not x.lower]127    def get_upper_nics(self) -> List[str]:128        return list(self.nics.keys())129    def get_lower_nics(self) -> List[str]:130        return [x.lower for x in self.nics.values() if x.lower]131    def get_device_slots(self) -> List[str]:132        return [x.pci_slot for x in self.nics.values() if x.pci_slot]133    # update the current nic driver in the NicInfo instance134    # grabs the driver short name and the driver sysfs path135    def get_nic_driver(self, nic_name: str) -> str:136        # get the current driver for the nic from the node137        # sysfs provides a link to the driver entry at device/driver138        nic = self.get_nic(nic_name)139        cmd = f"readlink -f /sys/class/net/{nic_name}/device/driver"140        # ex return value:141        # /sys/bus/vmbus/drivers/hv_netvsc142        found_link = self._node.execute(cmd, expected_exit_code=0).stdout143        assert_that(found_link).described_as(144            f"sysfs check for NIC device {nic_name} driver returned no output"145        ).is_not_equal_to("")146        nic.driver_sysfs_path = PurePosixPath(found_link)147        driver_name = nic.driver_sysfs_path.name148        assert_that(driver_name).described_as(149            f"sysfs entry contained no filename for device driver: {found_link}"150        ).is_not_equal_to("")151        nic.bound_driver = driver_name152        return driver_name153    def get_nic(self, nic_name: str) -> NicInfo:154        return self.nics[nic_name]155    def get_nic_by_index(self, index: int = -1) -> NicInfo:156        # get nic by index, default is -1 to give a non-primary nic157        # when there are more than one nic on the system158        number_of_nics = len(self.get_upper_nics())159        assert_that(number_of_nics).is_greater_than(0)160        try:161            nic_name = self.get_upper_nics()[index]162        except IndexError:163            raise LisaException(164                f"Attempted get_upper_nics()[{index}], only "165                f"{number_of_nics} nics are registered in node.nics. "166                f"Had upper interfaces: {self.get_upper_nics()}"167            )168        try:169            nic = self.nics[nic_name]170        except KeyError:171            raise LisaException(172                f"NicInfo for interface {nic_name} not found! "173                f"Had upper interfaces: {self.get_upper_nics()}"174            )175        return nic176    def nic_info_is_present(self, nic_name: str) -> bool:177        return nic_name in self.get_upper_nics() or nic_name in self.get_lower_nics()178    def unbind(self, nic: NicInfo) -> None:179        # unbind nic from current driver and return the old sysfs path180        echo = self._node.tools[Echo]181        # if sysfs path is not set, fetch the current driver182        if not nic.driver_sysfs_path:183            self.get_nic_driver(nic.upper)184        unbind_path = nic.driver_sysfs_path.joinpath("unbind")185        echo.write_to_file(186            nic.dev_uuid,187            unbind_path,188            sudo=True,189        )190    def bind(self, nic: NicInfo, driver_module_path: str) -> None:191        echo = self._node.tools[Echo]192        nic.driver_sysfs_path = PurePosixPath(driver_module_path)193        bind_path = nic.driver_sysfs_path.joinpath("bind")194        echo.write_to_file(195            nic.dev_uuid,196            self._node.get_pure_path(f"{str(bind_path)}"),197            sudo=True,198        )199        nic.bound_driver = nic.driver_sysfs_path.name200    def load_interface_info(self, nic_name: Optional[str] = None) -> None:201        command = "/sbin/ip addr show"202        if nic_name:203            command += f" {nic_name}"204        result = self._node.execute(205            command,206            shell=True,207            expected_exit_code=0,208            expected_exit_code_failure_message=(209                f"Could not run {command} on node {self._node.name}"210            ),211        )212        entries = find_groups_in_lines(213            result.stdout, self.__ip_addr_show_regex, single_line=False214        )215        found_nics = []216        for entry in entries:217            self._node.log.debug(f"Found nic info: {entry}")218            nic_name = entry["name"]219            mac = entry["mac"]220            ip_addr = entry["ip_addr"]221            if nic_name in self.get_upper_nics():222                nic_entry = self.nics[nic_name]223                nic_entry.ip_addr = ip_addr224                nic_entry.mac_addr = mac225                found_nics.append(nic_name)226        if not nic_name:227            assert_that(sorted(found_nics)).described_as(228                f"Could not locate nic info for all nics. "229                f"Nic set was {self.nics.keys()} and only found info for {found_nics}"230            ).is_equal_to(sorted(self.nics.keys()))231    def reload(self) -> None:232        self.nics.clear()233        self._initialize()234    @retry(tries=15, delay=3, backoff=1.15)235    def wait_for_sriov_enabled(self) -> None:236        lspci = self._node.tools[Lspci]237        # check for VFs on the guest238        vfs = lspci.get_devices_by_type(constants.DEVICE_TYPE_SRIOV, force_run=True)239        assert_that(len(vfs)).described_as(240            "Could not identify any SRIOV NICs on the test node."241        ).is_not_zero()242        # check if the NIC driver has finished setting up the243        # failsafe pair, reload if not244        if not self.get_lower_nics():245            self.reload()246        if not self.get_lower_nics():247            assert_that(self.get_lower_nics()).described_as(248                "Did not detect any upper/lower sriov paired nics!: "249                f"upper: {self.get_upper_nics()} "250                f"lower: {self.get_lower_nics()} "251                f"unpaired: {self.get_unpaired_devices()}"252                f"vfs: {','.join([str(pci) for pci in vfs])}"253            ).is_not_empty()254    def _initialize(self, *args: Any, **kwargs: Any) -> None:255        self._node.log.debug("loading nic information...")256        self.nic_names = self._get_nic_names()257        self._get_node_nic_info()258        self._get_default_nic()259        self.load_interface_info()260        self._get_nic_uuids()261        for nic in self.get_upper_nics():262            self.get_nic_driver(nic)263    def _get_nic_names(self) -> List[str]:264        # identify all of the nics on the device, excluding tunnels and loopbacks etc.265        all_nics = self._node.execute(266            "ls /sys/class/net/",267            shell=True,268            sudo=True,269        ).stdout.split()270        virtual_nics = self._node.execute(271            "ls /sys/devices/virtual/net",272            shell=True,273            sudo=True,274        ).stdout.split()275        # remove virtual nics from the list276        non_virtual_nics = [x for x in all_nics if x not in virtual_nics]277        # verify if the nics names are not empty278        for item in non_virtual_nics:279            assert_that(item).described_as(280                "nic name could not be found"281            ).is_not_equal_to("")282        return non_virtual_nics283    def _get_nic_device(self, nic_name: str) -> str:284        slot_info_result = self._node.execute(285            f"readlink /sys/class/net/{nic_name}/device"286        )287        slot_info_result.assert_exit_code()288        base_device_result = self._node.execute(f"basename {slot_info_result.stdout}")289        base_device_result.assert_exit_code()290        # todo check addr matches expectation291        return base_device_result.stdout292    def _get_nic_uuid(self, nic_name: str) -> str:293        full_dev_path = self._node.execute(294            f"readlink /sys/class/net/{nic_name}/device",295            expected_exit_code_failure_message=(296                f"could not get sysfs device info for {nic_name}"297            ),298        )299        uuid = os.path.basename(full_dev_path.stdout.strip())300        self._node.log.debug(f"{nic_name} UUID:{uuid}")301        return uuid302    def _get_nic_uuids(self) -> None:303        for nic in self.get_upper_nics():304            self.nics[nic].dev_uuid = self._get_nic_uuid(nic)305    def _get_nic_numa_node(self, name: str) -> int:306        result = self._node.execute(307            f"cat /sys/class/net/{name}/device/numa_node",308            sudo=True,309            shell=True,310            expected_exit_code=0,311            expected_exit_code_failure_message=(312                f"Could not get numa information for nic {name}"313            ),314        )315        numa = int(result.stdout.strip())316        if numa == -1:317            numa = 0318        return numa319    def _get_node_nic_info(self) -> None:320        # Identify which nics are slaved to master devices.321        # This should be really simple with /usr/bin/ip but experience shows322        # the tool isn't super consistent across distros in this regard323        # use sysfs to gather upper/lower nic pairings and pci slot info324        nic_info_fetch_cmd = "ls -la /sys/class/net/*/lower*/device"325        self._node.log.debug(f"Gathering NIC information on {self._node.name}.")326        result = self._node.execute(327            nic_info_fetch_cmd,328            shell=True,329        )330        if result.exit_code != 0:331            nic_info_fetch_cmd = "ls -la /sys/class/net/*/device"332            result = self._node.execute(333                nic_info_fetch_cmd,334                shell=True,335                expected_exit_code=0,336                expected_exit_code_failure_message="Could not grab NIC device info.",337            )338        for line in result.stdout.splitlines():339            sriov_match = self.__nic_lower_regex.search(line)340            if sriov_match:341                upper_nic, lower_nic, pci_slot = sriov_match.groups()342                nic_info = NicInfo(upper_nic, lower_nic, pci_slot)343                self.append(nic_info)344            sriov_match = self.__nic_vf_slot_regex.search(line)345            if sriov_match:346                lower_nic, pci_slot = sriov_match.groups()347                ip = self._node.tools[Ip]348                lower_nic_mac = ip.get_mac(lower_nic)349                for nic_name in [x for x in self.nic_names if x != lower_nic]:350                    upper_nic_mac = ip.get_mac(nic_name)351                    if upper_nic_mac == lower_nic_mac:352                        upper_nic = nic_name353                        nic_info = NicInfo(upper_nic, lower_nic, pci_slot)354                        self.append(nic_info)355                        break356        # Collects NIC info for any unpaired NICS357        for nic_name in [358            x359            for x in self.nic_names360            if x not in self.get_upper_nics() and x not in self.get_lower_nics()361        ]:362            nic_info = NicInfo(nic_name)363            self.append(nic_info)364        assert_that(len(self)).described_as(365            "During Lisa nic info initialization, Nics class could not "366            f"find any nics attached to {self._node.name}."367        ).is_greater_than(0)368    def _get_default_nic(self) -> None:369        cmd = "/sbin/ip route"370        ip_route_result = self._node.execute(cmd, shell=True, sudo=True)371        ip_route_result.assert_exit_code()372        assert_that(ip_route_result.stdout).is_not_empty()373        dev_match = self.__dev_regex.search(ip_route_result.stdout)374        if not dev_match or not dev_match.groups():...

common.py

Source:common.py

...26            f"version {node.os.information.version}"27        )28def verify_hibernation(node: RemoteNode, log: Logger) -> None:29    node_nic = node.nics30    lower_nics_before_hibernation = node_nic.get_lower_nics()31    upper_nics_before_hibernation = node_nic.get_upper_nics()32    hibernation_setup_tool = node.tools[HibernationSetup]33    entry_before_hibernation = hibernation_setup_tool.check_entry()34    exit_before_hibernation = hibernation_setup_tool.check_exit()35    received_before_hibernation = hibernation_setup_tool.check_received()36    uevent_before_hibernation = hibernation_setup_tool.check_uevent()37    startstop = node.features[StartStop]38    hibernation_setup_tool.start()39    startstop.stop(state=features.StopState.Hibernate)40    is_ready = True41    timeout = 90042    timer = create_timer()43    while timeout > timer.elapsed(False):44        is_ready, _ = wait_tcp_port_ready(45            node.public_address,46            node.public_port,47            log=log,48            timeout=10,49        )50        if not is_ready:51            break52    if is_ready:53        raise LisaException("VM still can be accessed after hibernation")54    startstop.start()55    entry_after_hibernation = hibernation_setup_tool.check_entry()56    exit_after_hibernation = hibernation_setup_tool.check_exit()57    received_after_hibernation = hibernation_setup_tool.check_received()58    uevent_after_hibernation = hibernation_setup_tool.check_uevent()59    assert_that(60        entry_after_hibernation - entry_before_hibernation,61        "not find 'hibernation entry'.",62    ).is_equal_to(1)63    assert_that(64        exit_after_hibernation - exit_before_hibernation,65        "not find 'hibernation exit'.",66    ).is_equal_to(1)67    assert_that(68        received_after_hibernation - received_before_hibernation,69        "not find 'Hibernation request received'.",70    ).is_equal_to(1)71    assert_that(72        uevent_after_hibernation - uevent_before_hibernation,73        "not find 'Sent hibernation uevent'.",74    ).is_equal_to(1)75    node_nic = node.nics76    node_nic.initialize()77    lower_nics_after_hibernation = node_nic.get_lower_nics()78    upper_nics_after_hibernation = node_nic.get_upper_nics()79    assert_that(80        len(lower_nics_after_hibernation),81        "sriov nics count changes after hibernation.",82    ).is_equal_to(len(lower_nics_before_hibernation))83    assert_that(84        len(upper_nics_after_hibernation),85        "synthetic nics count changes after hibernation.",86    ).is_equal_to(len(upper_nics_before_hibernation))87def run_storage_workload(node: RemoteNode) -> Decimal:88    fio = node.tools[Fio]89    fiodata = node.get_pure_path("./fiodata")90    core_count = node.tools[Lscpu].get_core_count()91    if node.shell.exists(fiodata):...

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.