def nvme_function_validation()

in microsoft/testsuites/nvme/nvme.py [0:0]

• 54 lines of code
• 4 McCabe index (conditional complexity)

    def nvme_function_validation(self, node: Node) -> None:
        nvme = node.features[Nvme]
        nvme_namespaces = nvme.get_namespaces()
        nvme_cli = node.tools[Nvmecli]
        cat = node.tools[Cat]
        mount = node.tools[Mount]
        for namespace in nvme_namespaces:
            # 1. Get the number of errors from nvme-cli before operations.
            error_count_before_operations = nvme_cli.get_error_count(namespace)

            # 2. Create a partition, filesystem and mount it.
            _format_mount_disk(node, namespace, FileSystem.ext4)

            # 3. Create a txt file on the partition, content is 'TestContent'.
            mount_point = namespace.rpartition("/")[-1]
            cmd_result = node.execute(
                f"echo TestContent > {mount_point}/testfile.txt", shell=True, sudo=True
            )
            cmd_result.assert_exit_code(
                message=f"{mount_point}/testfile.txt may not exist."
            )

            # 4. Create a file 'data' on the partition, get the md5sum value.
            cmd_result = node.execute(
                f"dd if=/dev/zero of={mount_point}/data bs=10M count=100",
                shell=True,
                sudo=True,
            )
            cmd_result.assert_exit_code(
                message=f"{mount_point}/data is not created successfully, "
                "please check the disk space."
            )
            initial_md5 = node.execute(
                f"md5sum {mount_point}/data", shell=True, sudo=True
            )
            initial_md5.assert_exit_code(
                message=f"{mount_point}/data not exist or md5sum command enounter"
                " unexpected error."
            )

            # 5. Umount and remount the partition.
            mount.umount(namespace, mount_point, erase=False)
            mount.mount(f"{namespace}p1", mount_point)

            # 6. Get the txt file content, compare the value.
            file_content = cat.run(f"{mount_point}/testfile.txt", shell=True, sudo=True)
            assert_that(
                file_content.stdout,
                f"content of {mount_point}/testfile.txt should keep consistent "
                "after umount and re-mount.",
            ).is_equal_to("TestContent")

            # 6. Get md5sum value of file 'data', compare with initial value.
            final_md5 = node.execute(
                f"md5sum {mount_point}/data", shell=True, sudo=True
            )
            assert_that(
                initial_md5.stdout,
                f"md5sum of {mount_point}/data should keep consistent "
                "after umount and re-mount.",
            ).is_equal_to(final_md5.stdout)

            # 7. Compare the number of errors from nvme-cli after operations.
            error_count_after_operations = nvme_cli.get_error_count(namespace)
            assert_that(
                error_count_before_operations,
                "error-log should not increase after operations.",
            ).is_equal_to(error_count_after_operations)

            mount.umount(disk_name=namespace, point=mount_point)