How to use get_partition_with_mount_point method in lisa

Best Python code snippet using lisa_python

azure_image_standard.py

Source:azure_image_standard.py

...858    def verify_resource_disk_readme_file(self, log: Logger, node: RemoteNode) -> None:859        resource_disk_mount_point = get_resource_disk_mount_point(log, node)860        # verify that resource disk is mounted861        # function returns successfully if disk matching mount point is present862        node.features[Disk].get_partition_with_mount_point(resource_disk_mount_point)863        # verify lost+found folder exists864        fold_path = f"{resource_disk_mount_point}/lost+found"865        folder_exists = node.tools[Ls].path_exists(fold_path, sudo=True)866        assert_that(folder_exists, f"{fold_path} should be present").is_true()867        # verify DATALOSS_WARNING_README.txt file exists868        file_path = f"{resource_disk_mount_point}/DATALOSS_WARNING_README.txt"869        file_exists = node.tools[Ls].path_exists(file_path, sudo=True)870        assert_that(file_exists, f"{file_path} should be present").is_true()871        # verify 'WARNING: THIS IS A TEMPORARY DISK' contained in872        # DATALOSS_WARNING_README.txt file.873        read_text = node.tools[Cat].read(file_path, force_run=True, sudo=True)874        assert_that(875            read_text,876            f"'WARNING: THIS IS A TEMPORARY DISK' should be present in {file_path}",877        ).contains("WARNING: THIS IS A TEMPORARY DISK")878    @TestCaseMetadata(879        description="""880        This test will check that resource disk is formatted correctly.881        Steps:882        1. Get the mount point for the resource disk. If `/var/log/cloud-init.log`883        file is present, mount location is `/mnt`, otherwise it is obtained from884        `ResourceDisk.MountPoint` entry in `waagent.conf` configuration file.885        2. Verify that resource disk file system type should not be 'ntfs'.886        """,887        priority=1,888        requirement=simple_requirement(889            disk=AzureDiskOptionSettings(has_resource_disk=True),890            supported_platform_type=[AZURE],891        ),892    )893    def verify_resource_disk_file_system(self, log: Logger, node: RemoteNode) -> None:894        resource_disk_mount_point = get_resource_disk_mount_point(log, node)895        node.features[Disk].get_partition_with_mount_point(resource_disk_mount_point)896        disk_info = node.tools[Lsblk].find_disk_by_mountpoint(resource_disk_mount_point)897        for partition in disk_info.partitions:898            # by default, resource disk comes with ntfs type899            # waagent or cloud-init will format it unless there are some commands hung900            # or interrupt901            assert_that(902                partition.fstype,903                "Resource disk file system type should not equal to ntfs",...

storage.py

Source:storage.py

...90        # os disk(root disk) is the entry with mount point `/' in the output91        # of `mount` command92        os_disk = (93            node.features[Disk]94            .get_partition_with_mount_point(self.os_disk_mount_point)95            .disk96        )97        mtab = node.tools[Cat].run("/etc/mtab").stdout98        resource_disk_from_mtab = get_matched_str(99            mtab, self._get_mtab_mount_point_regex(resource_disk_mount_point)100        )101        assert (102            resource_disk_from_mtab103        ), f"resource disk mountpoint not found {resource_disk_mount_point}"104        assert_that(105            resource_disk_from_mtab, "Resource disk should not be equal to os disk"106        ).is_not_equal_to(os_disk)107    @TestCaseMetadata(108        description="""109        This test will check that the swap is correctly configured on the VM.110        Steps:111        1. Check if swap file/partition is configured by checking the output of112        `swapon -s` and `lsblk`.113        2. Check swap status in `waagent.conf`.114        3. Verify that truth value in step 1 and step 2 match.115        """,116        priority=1,117        requirement=simple_requirement(supported_platform_type=[AZURE]),118    )119    def verify_swap(self, node: RemoteNode) -> None:120        is_swap_enabled_wa_agent = node.tools[Waagent].is_swap_enabled()121        is_swap_enabled_distro = node.tools[Swap].is_swap_enabled()122        assert_that(123            is_swap_enabled_distro,124            "swap configuration from waagent.conf and distro should match",125        ).is_equal_to(is_swap_enabled_wa_agent)126    @TestCaseMetadata(127        description="""128        This test will check that the file IO operations are working correctly129        Steps:130        1. Get the mount point for the resource disk. If `/var/log/cloud-init.log`131        file is present, mount location is `/mnt`, otherwise it is obtained from132        `ResourceDisk.MountPoint` entry in `waagent.conf` configuration file.133        2. Verify that resource disk is mounted from the output of `mount` command.134        3. Write a text file to the resource disk.135        4. Read the text file and verify that content is same.136        """,137        priority=1,138        requirement=simple_requirement(139            disk=AzureDiskOptionSettings(has_resource_disk=True),140            supported_platform_type=[AZURE],141        ),142    )143    def verify_resource_disk_io(self, log: Logger, node: RemoteNode) -> None:144        resource_disk_mount_point = get_resource_disk_mount_point(log, node)145        # verify that resource disk is mounted146        # function returns successfully if disk matching mount point is present147        node.features[Disk].get_partition_with_mount_point(resource_disk_mount_point)148        file_path = f"{resource_disk_mount_point}/sample.txt"149        original_text = "Writing to resource disk!!!"150        # write content to the file151        node.tools[Echo].write_to_file(152            original_text, node.get_pure_path(file_path), sudo=True153        )154        # read content from the file155        read_text = node.tools[Cat].read(file_path, force_run=True, sudo=True)156        assert_that(157            read_text,158            "content read from file should be equal to content written to file",159        ).is_equal_to(original_text)160    @TestCaseMetadata(161        description="""162        This test will verify that identifier of root partition matches163        from different sources.164        Steps:165        1. Get the partition identifier from `blkid` command.166        2. Verify that the partition identifier from `blkid` is present in dmesg.167        3. Verify that the partition identifier from `blkid` is present in fstab output.168        """,169        priority=1,170        requirement=simple_requirement(171            supported_platform_type=[AZURE],172        ),173    )174    def verify_os_partition_identifier(self, log: Logger, node: RemoteNode) -> None:175        # get information of root disk from blkid176        os_partition = (177            node.features[Disk]178            .get_partition_with_mount_point(self.os_disk_mount_point)179            .name180        )181        os_partition_info = node.tools[Blkid].get_partition_info_by_name(os_partition)182        # verify that root=<name> or root=uuid=<uuid> or root=partuuid=<part_uuid> is183        # present in dmesg184        dmesg = node.tools[Dmesg].run(sudo=True).stdout185        if (186            not get_matched_str(187                dmesg,188                re.compile(189                    rf".*BOOT_IMAGE=.*root={os_partition_info.name}",190                ),191            )192            and not get_matched_str(...

disks.py

Source:disks.py

...15    def can_disable(cls) -> bool:16        return False17    def enabled(self) -> bool:18        return True19    def get_partition_with_mount_point(self, mount_point: str) -> PartitionInfo:20        partition_info = self._node.tools[Mount].get_partition_info()21        matched_partitions = [22            partition23            for partition in partition_info24            if partition.mount_point == mount_point25        ]26        assert_that(27            matched_partitions,28            f"Exactly one partition with mount point {mount_point} should be present",29        ).is_length(1)30        partition = matched_partitions[0]31        self._log.debug(f"disk: {partition}, mount_point: {mount_point}")32        return partition33    def get_raw_data_disks(self) -> List[str]:...

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.