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")...
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!!