How to use _format_disk method in lisa

Best Python code snippet using lisa_python

lenovo.py

Source:lenovo.py

...229            return (None, None)230        columns = sel[0].keys()231        values = [[line[col] for col in columns] for line in sel]232        return columns, values233    def _format_disk(self, disk):234        return " | ".join(235            map(236                str,237                [238                    disk["DRIVE_INDEX"],239                    "ctrl{}".format(disk["CONTROLLER_INDEX"]),240                    "slot{}".format(disk["SLOT_NUMBER"]),241                    "{}gb".format(disk["SIZE"] // 1024),242                    DISK_TYPE.get(disk["MEDIA_TYPE"], "N/A"),243                    DISK_STATE.get(disk["DEVICE_STATE"], "unknown"),244                    DISK_INTERFACE.get(disk["INTF_TYPE"], "unknown"),245                    DISK_SPEED.get(disk["LINK_SPEED"], "unknown"),246                    disk["VENDOR_ID"],247                ],248            )249        )250    def check_disks(self):251        pre = "{} disks".format(self.oob_info["identifier"])252        expected = self.parsed_args.expected253        disks = self._get_disks()254        disks_ok, disks_crit = [], []255        for d in disks:256            if d["DEVICE_STATE"] != 0 or d["SIZE"] <= 0:257                disks_crit.append(d)258            else:259                disks_ok.append(d)260        status, msg, lines = nagios.OK, [], []261        if disks_crit:262            status = nagios.CRITICAL263            lines.append("{} disks CRITICAL:".format(len(disks_crit)))264            lines.extend(map(lambda d: "- {}".format(self._format_disk(d)), disks_crit))265            msg.append("{} disks CRITICAL".format(len(disks_crit)))266        msg.append("{} disks OK".format(len(disks_ok)))267        if expected is not None and len(disks_ok) != expected:268            msg.append("expected {}".format(expected))269            status = max(status, nagios.WARNING)270            lines.append("{} disks OK:".format(len(disks_ok)))271            lines.extend(map(lambda d: "- {}".format(self._format_disk(d)), disks_ok))272            if len(disks_ok) < expected:273                status = max(status, nagios.CRITICAL)274        nagios.result(status, msg, lines, pre=pre)275    def _get_image_info(self):276        return self._get_rpc("getimageinfo", "firmware versions")277    def _format_fw(self, fw):278        return "- " + " | ".join(279            [280                DEV_ID.get(fw["DEV_TYPE"], "unknown"),281                fw["CURIMG_VER"],282            ]283        )284    def get_firmware(self):285        firmwares = self._get_image_info()...

disk

Source:disk

...65    for cmd in cmds:66        print("==>", " ".join(cmd))67        sys.stdout.flush()68        subprocess.run(cmd, check=True)69def _format_disk(device_path: str) -> None:70    (parted, _) = _parse_script(sys.stdin.read())71    cmds: List[List[str]] = []72    for cmd in parted:73        cmds.append(["/usr/bin/parted", device_path, "-a", "optimal", "-s", *cmd])74    cmds.append(["/usr/bin/partprobe", device_path])75    cmds.append(["/usr/bin/partx", "-vu", device_path])76    _run_commands(cmds)77def _mkfs_disk(device_path: str) -> None:78    (_, filesystems) = _parse_script(sys.stdin.read())79    cmds: List[List[str]] = []80    for mkfs in filesystems:81        cmd: List[str] = []82        if mkfs.fs == "fat32":83            cmd.append("/usr/bin/mkfs.vfat")84            if mkfs.label:85                cmd.extend(["-n", mkfs.label])86        elif mkfs.fs == "ext4":87            cmd.append("/usr/bin/mkfs.ext4")88            if mkfs.label:89                cmd.extend(["-L", mkfs.label])90            if mkfs.reserved:91                cmd.extend(["-m", mkfs.reserved])92        else:93            raise RuntimeError(f"Unsupported filesystem: {mkfs.fs}")94        cmd.append(_make_partition(device_path, mkfs.npart))95        cmds.append(cmd)96    cmds.append(["/usr/bin/partprobe", device_path])97    cmds.append(["/usr/bin/partx", "-vu", device_path])98    _run_commands(cmds)99def _mount_disk(device_path: str, prefix_path: str, mount: bool) -> None:100    (_, filesystems) = _parse_script(sys.stdin.read())101    cmds: List[List[str]] = []102    filesystems = sorted(filesystems, key=(lambda mkfs: len(list(filter(None, mkfs.mount.split("/"))))))103    if not mount:104        filesystems.reverse()105    for mkfs in filesystems:106        if mkfs.mount:107            part_path = _make_partition(device_path, mkfs.npart)108            if mount:109                mount_path = prefix_path + "/" + mkfs.mount110                cmds.append(["/usr/bin/mkdir", "-p", mount_path])111                cmds.append(["/usr/bin/mount", part_path, mount_path])112            else:113                cmds.append(["/usr/bin/umount", part_path])114    _run_commands(cmds)115# =====116def main() -> None:117    if len(sys.argv) <= 2:118        raise SystemExit(f"Usage: {sys.argv[0]} <format|mount|umount> ...")119    if sys.argv[1] == "format":120        if len(sys.argv) != 3:121            raise SystemExit(f"Usage: cat script | {sys.argv[0]} format /dev/path")122        _format_disk(sys.argv[2])123    elif sys.argv[1] == "mkfs":124        if len(sys.argv) != 3:125            raise SystemExit(f"Usage: cat script | {sys.argv[0]} format /dev/path")126        _mkfs_disk(sys.argv[2])127    elif sys.argv[1] == "mount":128        if len(sys.argv) != 4:129            raise SystemExit(f"Usage: {sys.argv[0]} mount /dev/path /prefix/path")130        _mount_disk(sys.argv[2], sys.argv[3], True)131    elif sys.argv[1] == "umount":132        if len(sys.argv) != 3:133            raise SystemExit(f"Usage: {sys.argv[0]} umount /dev/path")134        _mount_disk(sys.argv[2], "/dev/null", False)135    else:136        raise SystemExit(f"Usage: {sys.argv[0]} <format|mount|umount> ...")...

storagesuite.py

Source:storagesuite.py

...14)15from lisa.environment import Environment16from lisa.features import Disk17from lisa.tools import Fdisk, FileSystem, Mdadm, Mkfs, Mount18def _format_disk(19    node: Node,20    disk_list: List[str],21) -> List[str]:22    fdisk = node.tools[Fdisk]23    partition_disks: List[str] = []24    for data_disk in disk_list:25        fdisk.delete_partitions(data_disk)26        partition_disks.append(fdisk.make_partition(data_disk, format=False))27    return partition_disks28def _stop_raid(node: Node) -> None:29    mdadm = node.tools[Mdadm]30    mdadm.stop_raid()31def _make_mount_raid(node: Node, disk_list: List[str]) -> None:32    mdadm = node.tools[Mdadm]33    mdadm.create_raid(disk_list)34    mkfs = node.tools[Mkfs]35    mkfs.format_disk("/dev/md0", FileSystem.ext4)36    mount = node.tools[Mount]37    mount.mount("/dev/md0", "/data", options="nobarrier")38@TestSuiteMetadata(39    area="storage",40    category="functional",41    description="""42    This test suite is to validate storage function in Linux VM.43    """,44)45class StorageTest(TestSuite):46    @TestCaseMetadata(47        description="""48        This test case is to49            1. Make raid0 based on StandardHDDLRS disks.50            2. Mount raid0 with nobarrier options.51        """,52        priority=3,53        requirement=simple_requirement(54            disk=schema.DiskOptionSettings(55                disk_type=schema.DiskType.StandardHDDLRS,56                data_disk_iops=search_space.IntRange(min=500),57                data_disk_count=search_space.IntRange(min=64),58            ),59        ),60    )61    def verify_disk_with_nobarrier(self, node: Node, environment: Environment) -> None:62        disk = node.features[Disk]63        data_disks = disk.get_raw_data_disks()64        disk_count = len(data_disks)65        assert_that(disk_count).described_as(66            "At least 1 data disk for testing."67        ).is_greater_than(0)68        partition_disks = _format_disk(node, data_disks)69        _stop_raid(node)70        _make_mount_raid(node, partition_disks)71    def after_case(self, log: Logger, **kwargs: Any) -> None:72        node: Node = kwargs.pop("node")73        mount = node.tools[Mount]74        mount.umount("/dev/md0", "/data")...

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.