How to use enable_public_key method in lisa

Best Python code snippet using lisa_python

infinibandsuit.py

Source:infinibandsuit.py

...199        client_ip = client_ib_interfaces[0].ip_addr200        # Test relies on machines being able to ssh into each other201        server_ssh = server_node.tools[Ssh]202        client_ssh = client_node.tools[Ssh]203        server_ssh.enable_public_key(client_ssh.generate_key_pairs())204        client_ssh.enable_public_key(server_ssh.generate_key_pairs())205        server_ssh.add_known_host(client_ip)206        client_ssh.add_known_host(server_ip)207        # Note: Using bash because script is not supported by Dash208        # sh points to dash on Ubuntu209        server_node.execute(210            "bash /opt/intel/oneapi/mpi/2021.1.1/bin/mpirun "211            f"-hosts {server_ip},{server_ip} -iface {server_nic_name} -ppn 1 -n 2 "212            "-env I_MPI_FABRICS=shm:ofi -env SECS_PER_SAMPLE=600 "213            "-env FI_PROVIDER=mlx -env I_MPI_DEBUG=5 -env I_MPI_PIN_DOMAIN=numa "214            "/opt/intel/oneapi/mpi/2021.1.1/bin/IMB-MPI1 pingpong",215            expected_exit_code=0,216            expected_exit_code_failure_message="Failed intra-node pingpong test "217            "with intel mpi",218        )219        server_node.execute(220            "bash /opt/intel/oneapi/mpi/2021.1.1/bin/mpirun "221            f"-hosts {server_ip},{client_ip} -iface {server_nic_name} -ppn 1 -n 2 "222            "-env I_MPI_FABRICS=shm:ofi -env SECS_PER_SAMPLE=600 "223            "-env FI_PROVIDER=mlx -env I_MPI_DEBUG=5 -env I_MPI_PIN_DOMAIN=numa "224            "/opt/intel/oneapi/mpi/2021.1.1/bin/IMB-MPI1 pingpong",225            expected_exit_code=0,226            expected_exit_code_failure_message="Failed inter-node pingpong test "227            "with intel mpi",228        )229        tests = ["IMB-MPI1 allreduce", "IMB-RMA", "IMB-NBC"]230        for test in tests:231            server_node.execute(232                "bash /opt/intel/oneapi/mpi/2021.1.1/bin/mpirun "233                f"-hosts {server_ip},{client_ip} -iface {server_nic_name} -ppn 22 "234                "-n 44 -env I_MPI_FABRICS=shm:ofi -env SECS_PER_SAMPLE=600 "235                "-env FI_PROVIDER=mlx -env I_MPI_DEBUG=5 -env I_MPI_PIN_DOMAIN=numa "236                f"/opt/intel/oneapi/mpi/2021.1.1/bin/{test}",237                expected_exit_code=0,238                expected_exit_code_failure_message=f"Failed {test} test with intel mpi",239                timeout=1200,240            )241    @TestCaseMetadata(242        description="""243            This test case will244            1. Ensure RDMA is setup245            2. Install Open MPI246            3. Set up ssh keys of server/client connection247            4. Run MPI pingpong tests248            5. Run other MPI tests249            """,250        priority=4,251        requirement=simple_requirement(252            supported_features=[Infiniband],253            min_count=2,254        ),255    )256    def verify_open_mpi(self, environment: Environment, log: Logger) -> None:257        server_node = environment.nodes[0]258        client_node = environment.nodes[1]259        # Ensure RDMA is setup260        try:261            run_in_parallel(262                [263                    lambda: client_node.features[Infiniband],264                    lambda: server_node.features[Infiniband],265                ]266            )267        except (UnsupportedDistroException, UnsupportedKernelException) as err:268            raise SkippedException(err)269        server_ib = server_node.features[Infiniband]270        client_ib = client_node.features[Infiniband]271        run_in_parallel([server_ib.install_open_mpi, client_ib.install_open_mpi])272        server_node.execute("ldconfig", sudo=True)273        client_node.execute("ldconfig", sudo=True)274        # Restart the ssh sessions for changes to /etc/security/limits.conf275        # to take effect276        server_node.close()277        client_node.close()278        # Get the ip adresses and device name of ib device279        server_ib_interfaces = server_ib.get_ib_interfaces()280        client_ib_interfaces = client_ib.get_ib_interfaces()281        server_ip = server_ib_interfaces[0].ip_addr282        client_ip = client_ib_interfaces[0].ip_addr283        # Test relies on machines being able to ssh into each other284        server_ssh = server_node.tools[Ssh]285        client_ssh = client_node.tools[Ssh]286        server_ssh.enable_public_key(client_ssh.generate_key_pairs())287        client_ssh.enable_public_key(server_ssh.generate_key_pairs())288        server_ssh.add_known_host(client_ip)289        client_ssh.add_known_host(server_ip)290        # Ping Pong test291        find = server_node.tools[Find]292        find_results = find.find_files(293            server_node.get_pure_path("/usr"), "IMB-MPI1", sudo=True294        )295        assert_that(len(find_results)).described_as(296            "Could not find location of IMB-MPI1 for Open MPI"297        ).is_greater_than(0)298        test_path = find_results[0]299        assert_that(test_path).described_as(300            "Could not find location of IMB-MPI1 for Open MPI"301        ).is_not_empty()302        server_node.execute(303            f"/usr/local/bin/mpirun --host {server_ip},{server_ip} "304            "-n 2 --mca btl self,vader,openib --mca btl_openib_cq_size 4096 "305            "--mca btl_openib_allow_ib 1 --mca "306            f"btl_openib_warn_no_device_params_found 0 {test_path} pingpong",307            expected_exit_code=0,308            expected_exit_code_failure_message="Failed intra-node ping pong test "309            "with Open MPI",310        )311        # IMB-MPI Tests312        find_results = find.find_files(313            server_node.get_pure_path("/usr"), "IMB-MPI1", sudo=True314        )315        assert_that(len(find_results)).described_as(316            "Could not find location of Open MPI test: IMB-MPI1"317        ).is_greater_than(0)318        test_path = find_results[0]319        assert_that(test_path).described_as(320            "Could not find location of Open MPI test: IMB-MPI1"321        ).is_not_empty()322        server_node.execute(323            f"/usr/local/bin/mpirun --host {server_ip},{client_ip} "324            "-n 2 --mca btl self,vader,openib --mca btl_openib_cq_size 4096 "325            "--mca btl_openib_allow_ib 1 --mca "326            f"btl_openib_warn_no_device_params_found 0 {test_path}",327            expected_exit_code=0,328            expected_exit_code_failure_message="Failed " "IMB-MPI1 test with Open MPI",329        )330    @TestCaseMetadata(331        description="""332            This test case will333            1. Ensure RDMA is setup334            2. Install IBM MPI335            3. Set up ssh keys of server/client connection336            4. Run MPI pingpong tests337            """,338        priority=4,339        requirement=simple_requirement(340            supported_features=[Infiniband],341            min_count=2,342        ),343    )344    def verify_ibm_mpi(self, environment: Environment, log: Logger) -> None:345        server_node = environment.nodes[0]346        client_node = environment.nodes[1]347        # Ensure RDMA is setup348        try:349            run_in_parallel(350                [351                    lambda: client_node.features[Infiniband],352                    lambda: server_node.features[Infiniband],353                ]354            )355        except (UnsupportedDistroException, UnsupportedKernelException) as err:356            raise SkippedException(err)357        server_ib = server_node.features[Infiniband]358        client_ib = client_node.features[Infiniband]359        run_in_parallel([server_ib.install_ibm_mpi, client_ib.install_ibm_mpi])360        # Restart the ssh sessions for changes to /etc/security/limits.conf361        # to take effect362        server_node.close()363        client_node.close()364        # Get the ip adresses and device name of ib device365        server_ib_interfaces = server_ib.get_ib_interfaces()366        client_ib_interfaces = client_ib.get_ib_interfaces()367        server_ip = server_ib_interfaces[0].ip_addr368        client_ip = client_ib_interfaces[0].ip_addr369        # Test relies on machines being able to ssh into each other370        server_ssh = server_node.tools[Ssh]371        client_ssh = client_node.tools[Ssh]372        server_ssh.enable_public_key(client_ssh.generate_key_pairs())373        client_ssh.enable_public_key(server_ssh.generate_key_pairs())374        server_ssh.add_known_host(client_ip)375        client_ssh.add_known_host(server_ip)376        server_node.execute(377            "/opt/ibm/platform_mpi/bin/mpirun "378            f"-hostlist {server_ip}:1,{server_ip}:1 -np 2 -e "379            f"MPI_IB_PKEY={server_ib.get_pkey()} -ibv /opt/ibm/platform_mpi/help/"380            "ping_pong 4096",381            expected_exit_code=0,382            expected_exit_code_failure_message="Infiniband intra-node ping pong "383            "test failed with IBM MPI",384        )385        server_node.execute(386            "/opt/ibm/platform_mpi/bin/mpirun "387            f"-hostlist {server_ip}:1,{client_ip}:1 -np 2 -e "388            f"MPI_IB_PKEY={server_ib.get_pkey()} -ibv /opt/ibm/platform_mpi/help/"389            "ping_pong 4096",390            expected_exit_code=0,391            expected_exit_code_failure_message="Infiniband inter-node ping pong "392            "test failed with IBM MPI",393        )394    @TestCaseMetadata(395        description="""396            This test case will397            1. Ensure RDMA is setup398            2. Install MVAPICH MPI399            3. Set up ssh keys of server/client connection400            4. Run MPI pingpong tests401            5. Run other MPI tests402            """,403        priority=4,404        requirement=simple_requirement(405            supported_features=[Infiniband],406            min_count=2,407        ),408    )409    def verify_mvapich_mpi(self, environment: Environment, log: Logger) -> None:410        server_node = environment.nodes[0]411        client_node = environment.nodes[1]412        # Ensure RDMA is setup413        try:414            run_in_parallel(415                [416                    lambda: client_node.features[Infiniband],417                    lambda: server_node.features[Infiniband],418                ]419            )420        except (UnsupportedDistroException, UnsupportedKernelException) as err:421            raise SkippedException(err)422        server_ib = server_node.features[Infiniband]423        client_ib = client_node.features[Infiniband]424        run_in_parallel([server_ib.install_mvapich_mpi, client_ib.install_mvapich_mpi])425        # Restart the ssh sessions for changes to /etc/security/limits.conf426        # to take effect427        server_node.close()428        client_node.close()429        # Get the ip adresses and device name of ib device430        server_ib_interfaces = server_ib.get_ib_interfaces()431        client_ib_interfaces = client_ib.get_ib_interfaces()432        server_ip = server_ib_interfaces[0].ip_addr433        client_ip = client_ib_interfaces[0].ip_addr434        # Test relies on machines being able to ssh into each other435        server_ssh = server_node.tools[Ssh]436        client_ssh = client_node.tools[Ssh]437        server_ssh.enable_public_key(client_ssh.generate_key_pairs())438        client_ssh.enable_public_key(server_ssh.generate_key_pairs())439        server_ssh.add_known_host(client_ip)440        client_ssh.add_known_host(server_ip)441        # Run MPI tests442        find = server_node.tools[Find]443        test_names = ["IMB-MPI1", "IMB-RMA", "IMB-NBC"]444        for test in test_names:445            find_results = find.find_files(446                server_node.get_pure_path("/usr"), test, sudo=True447            )448            assert_that(len(find_results)).described_as(449                f"Could not find location of MVAPICH MPI test: {test}"450            ).is_greater_than(0)451            test_path = find_results[0]452            assert_that(test_path).described_as(...

common.py

Source:common.py

...103    source_node = cast(RemoteNode, environment.nodes[0])104    dest_node = cast(RemoteNode, environment.nodes[1])105    source_ssh = source_node.tools[Ssh]106    dest_ssh = dest_node.tools[Ssh]107    dest_ssh.enable_public_key(source_ssh.generate_key_pairs())108    # generate 200Mb file109    source_node.execute("dd if=/dev/urandom of=large_file bs=100 count=0 seek=2M")110    max_retry_times = 10111    for _, source_nic_info in vm_nics[source_node.name].items():112        matched_dest_nic_name = ""113        for dest_nic_name, dest_nic_info in vm_nics[dest_node.name].items():114            # only when IPs are in the same subnet, IP1 of machine A can connect to115            # IP2 of machine B116            # e.g. eth2 IP is 10.0.2.3 on machine A, eth2 IP is 10.0.3.4 on machine117            # B, use nic name doesn't work in this situation118            if (119                dest_nic_info.ip_addr.rsplit(".", maxsplit=1)[0]120                == source_nic_info.ip_addr.rsplit(".", maxsplit=1)[0]121            ):...

ssh.py

Source:ssh.py

...31            str(self.node.get_pure_path("~/.ssh/id_rsa.pub")),32            force_run=True,33        )34        return public_key35    def enable_public_key(self, public_key: str) -> None:36        self.node.tools[Echo].write_to_file(37            public_key,38            self.node.get_pure_path("~/.ssh/authorized_keys"),39            append=True,40        )41    def add_known_host(self, ip: str) -> None:42        self.node.execute(f"ssh-keyscan -H {ip} >> ~/.ssh/known_hosts", shell=True)43    def get_default_sshd_config_path(self) -> str:44        file_name = "sshd_config"45        default_path = f"/etc/ssh/{file_name}"46        if self.node.shell.exists(self.node.get_pure_path(default_path)):47            return default_path48        find = self.node.tools[Find]49        result = find.find_files(self.node.get_pure_path("/"), file_name, sudo=True)...

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.