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

How to use setup_rdma method in lisa

Best Python code snippet using lisa_python

infiniband.py

Source:infiniband.py

...43    #                     link_layer:             InfiniBand44    _ib_info_pattern = re.compile(r"(\s*(?P<id>\S*):\s*(?P<value>.*)\n?)")45    def _initialize(self, *args: Any, **kwargs: Any) -> None:46        super()._initialize(*args, **kwargs)47        self.setup_rdma()48    def enabled(self) -> bool:49        return True50    @classmethod51    def name(cls) -> str:52        return FEATURE_NAME_INFINIBAND53    @classmethod54    def can_disable(cls) -> bool:55        return True56    def is_over_sriov(self) -> bool:57        raise NotImplementedError58    # nd stands for network direct59    # example SKU: Standard_H16mr60    def is_over_nd(self) -> bool:61        raise NotImplementedError62    def _is_legacy_device(self) -> bool:63        lspci = self._node.tools[Lspci]64        device_list = lspci.get_devices()65        return any("ConnectX-3" in device.device_info for device in device_list)66    def get_ib_interfaces(self) -> List[IBDevice]:67        """Gets the list of Infiniband devices68        excluding any ethernet devices69        and get their corresponding network interface70        Returns list of IBDevice(ib_device_name, nic_name, ip_addr)71        Example IBDevice("mlx5_ib0", "ib0", "172.16.1.23")"""72        ib_devices = []73        device_info = self._get_ib_device_info()74        for device in device_info:75            if device["link_layer"].strip() == "InfiniBand" and "node_guid" in device:76                device_name = device["hca_id"].strip()77                guid = device["node_guid"].strip()78                # Get the last three bytes of guid79                # Example80                # guid = 0015:5dff:fe33:ff0c81                # mpat = 33:ff:0c (This will match the ib device)82                mpat = f"{guid[12:17]}:{guid[17:19]}"83                for (nic_name, nic_info) in self._node.nics.nics.items():84                    result = self._node.execute(f"/sbin/ip addr show {nic_name}")85                    if mpat in result.stdout and "ib" in nic_name:86                        assert_that(nic_info.ip_addr).described_as(87                            f"NIC {nic_name} does not have an ip address."88                        ).is_not_empty()89                        ib_devices.append(90                            IBDevice(device_name, nic_name, nic_info.ip_addr)91                        )92        assert_that(ib_devices).described_as(93            "Failed to get any InfiniBand device / interface pairs"94        ).is_not_empty()95        return ib_devices96    def _get_ib_device_info(self) -> List[Dict[str, str]]:97        device_info = []98        devices = self._get_ib_device_names()99        for device_name in devices:100            result = self._node.execute(101                f"ibv_devinfo -d {device_name}",102                expected_exit_code=0,103                expected_exit_code_failure_message="Failed to get device info from "104                f"ibv_devinfo for infiniband device {device_name}",105            )106            d = {107                match.group("id"): match.group("value")108                for match in self._ib_info_pattern.finditer(result.stdout)109            }110            if "hca_id" in d:111                device_info.append(d)112        assert_that(device_info).described_as(113            "Failed to get device info for any InfiniBand devices"114        ).is_not_empty()115        return device_info116    def _get_ib_device_names(self) -> List[str]:117        node = self._node118        result = node.execute(119            "ls /sys/class/infiniband",120            expected_exit_code=0,121            expected_exit_code_failure_message="Failed to get InfiniBand"122            " devices from /sys/class/infiniband",123        )124        assert_that(result.stdout).described_as(125            "No infiniband devices found in /sys/class/infiniband"126        ).is_not_empty()127        return result.stdout.split()128    def _get_mofed_version(self) -> str:129        node = self._node130        default = "5.4-3.0.3.0"131        if self._is_legacy_device():132            return "4.9-5.1.0.0"133        if (134            isinstance(node.os, Ubuntu) and node.os.information.version >= "20.4.0"135        ) or (isinstance(node.os, Redhat) and node.os.information.version >= "8.2.0"):136            return "5.7-1.0.2.0"137        return default138    def get_pkey(self) -> str:139        ib_device_name = self.get_ib_interfaces()[0].ib_device_name140        cat = self._node.tools[Cat]141        return cat.read(f"/sys/class/infiniband/{ib_device_name}/ports/1/pkeys/0")142    def setup_rdma(self) -> None:143        node = self._node144        # Dependencies145        kernel = node.tools[Uname].get_linux_information().kernel_version_raw146        ubuntu_required_packages = [147            "build-essential",148            "numactl",149            "rpm",150            "libnuma-dev",151            "libmpc-dev",152            "libmpfr-dev",153            "libxml2-dev",154            "m4",155            "byacc",156            "python-dev",...

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.