Best Python code snippet using lisa_python
infiniband.py
Source:infiniband.py  
...58    # 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",157            "python-setuptools",158            "tcl",159            "environment-modules",160            "tk",161            "texinfo",162            "libudev-dev",163            "binutils",164            "binutils-dev",165            "selinux-policy-dev",166            "flex",167            "libnl-3-dev",168            "libnl-route-3-dev",169            "libnl-3-200",170            "bison",171            "libnl-route-3-200",172            "gfortran",173            "cmake",174            "libnl-3-dev",175            "libnl-route-3-dev",176            "libsecret-1-0",177            "dkms",178            "python-setuptools",179            "g++",180            "libc6-i386",181            "lib32gcc-8-dev",182        ]183        redhat_required_packages = [184            "git",185            "zip",186            "python3",187            "kernel-rpm-macros",188            "gdb-headless",189            "elfutils-libelf-devel",190            "rpm-build",191            "make",192            "gcc",193            "tcl",194            "tk",195            "gcc-gfortran",196            "tcsh",197            "kernel-modules-extra",198            "createrepo",199            "libtool",200            "fuse-libs",201            "gcc-c++",202            "glibc",203            "libgcc",204            "byacc",205            "libevent",206            "pciutils",207            "lsof",208        ]209        if isinstance(node.os, CentOs):210            node.execute(211                "yum install -y https://partnerpipelineshare.blob.core.windows.net"212                f"/kernel-devel-rpms/kernel-devel-{kernel}.rpm",213                sudo=True,214            )215        if isinstance(node.os, Redhat):216            if node.os.information.version.major >= 9:217                redhat_required_packages.append("perl-CPAN")218                redhat_required_packages.append("perl-Pod-Html")219            for package in [220                "python36-devel",221                "python3-devel",222                "python-devel",223                "python2-devel",224            ]:225                if node.os.is_package_in_repo(package):226                    redhat_required_packages.append(package)227            node.os.install_packages(redhat_required_packages)228            try:229                node.os.install_packages("kernel-devel-$(uname -r)")230            except MissingPackagesException:231                node.log.debug(232                    "kernel-devel-$(uname -r) not found. Trying kernel-devel"233                )234                node.os.install_packages("kernel-devel")235        elif isinstance(node.os, Ubuntu) and node.os.information.version >= "18.4.0":236            node.os.install_packages(ubuntu_required_packages)237        else:238            raise UnsupportedDistroException(239                node.os,240                "Only CentOS 7.6-8.3 and Ubuntu 18.04-20.04 distros are "241                "supported by the HCP team",242            )243        # Turn off firewall244        firewall = node.tools[Firewall]245        firewall.stop()246        # Disable SELinux247        sed = node.tools[Sed]248        sed.substitute(249            regexp="SELINUX=enforcing",250            replacement="SELINUX=disabled",251            file="/etc/selinux/config",252            sudo=True,253        )254        # Install OFED255        mofed_version = self._get_mofed_version()256        if isinstance(node.os, Redhat):257            os_class = "rhel"258        else:259            os_class = node.os.name.lower()260        os_version = node.os.information.release.split(".")261        mofed_folder = (262            f"MLNX_OFED_LINUX-{mofed_version}-{os_class}"263            f"{os_version[0]}."264            f"{os_version[1]}-x86_64"265        )266        tarball_name = f"{mofed_folder}.tgz"267        mlnx_ofed_download_url = (268            f"https://partnerpipelineshare.blob.core.windows.net/ofed/{tarball_name}"269        )270        wget = node.tools[Wget]271        wget.get(url=mlnx_ofed_download_url, file_path=".", filename=tarball_name)272        tar = node.tools[Tar]273        tar.extract(file=tarball_name, dest_dir=".", gzip=True)274        extra_params = ""275        if isinstance(node.os, Redhat):276            ls = node.tools[Ls]277            kernel_dirs = ls.list_dir("/usr/src/kernels")278            if f"/usr/src/kernels/{kernel}" in kernel_dirs:279                kernel_src = f"/usr/src/kernels/{kernel}"280            elif kernel_dirs:281                kernel_src = kernel_dirs[0]282            else:283                raise UnsupportedKernelException(284                    node.os, "Cannot install OFED drivers without kernel-devel package"285                )286            extra_params = (287                f"--kernel {kernel} --kernel-sources {kernel_src}  "288                f"--skip-repo --without-fw-update"289            )290        if not self._is_legacy_device():291            extra_params += " --skip-unsupported-devices-check"292        node.execute(293            f"./{mofed_folder}/mlnxofedinstall --add-kernel-support {extra_params}",294            expected_exit_code=0,295            expected_exit_code_failure_message="SetupRDMA: failed to install "296            "MOFED Drivers",297            sudo=True,298            timeout=1200,299        )300        node.execute(301            "/etc/init.d/openibd force-restart",302            expected_exit_code=0,303            expected_exit_code_failure_message="SetupRDMA: failed to restart driver",304            sudo=True,...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!!
