[pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

Author: pre-commit-ci[bot]

virtme/commands/run.py

@@ -20,11 +20,12 @@
 import tempfile
 import termios
 from base64 import b64encode
+from collections.abc import Callable
 from dataclasses import dataclass
 from pathlib import Path
 from shutil import which
 from time import sleep
-from typing import Any, Callable, NoReturn
+from typing import Any, NoReturn
 
 from virtme_ng.utils import (
     CACHE_DIR,
@@ -897,7 +898,10 @@ class DiskArg:
         # I/O driver parameters
         "cache": ("str", "Cache mode (none|writeback|writethrough|directsync|unsafe)"),
         "aio": ("str", "Asynchronous I/O mode (native|threads|io_uring)"),
-        "discard": ("bool", "Pass through TRIM/UNMAP requests (true=unmap, false=ignore)"),
+        "discard": (
+            "bool",
+            "Pass through TRIM/UNMAP requests (true=unmap, false=ignore)",
+        ),
         "detect-zeroes": ("bool", "Detect all-zero writes (true=on/unmap, false=off)"),
         "queues": ("int", "Number of I/O queues"),
         # topology parameters
@@ -918,15 +922,26 @@ def __post_init__(self):
         if self.pop_opt("topology", strtobool, False):
             self.opts = self.topology() | self.opts
 
-    def get_opt(self, name: str, parser: Callable[[str], Any] = str, default: Any = None) -> Any:
+    def get_opt(
+        self, name: str, parser: Callable[[str], Any] = str, default: Any = None
+    ) -> Any:
         opt = self.opts.get(name, None)
         return parser(opt) if opt is not None else default
 
-    def pop_opt(self, name: str, parser: Callable[[str], Any] = str, default: Any = None) -> Any:
+    def pop_opt(
+        self, name: str, parser: Callable[[str], Any] = str, default: Any = None
+    ) -> Any:
         opt = self.opts.pop(name, None)
         return parser(opt) if opt is not None else default
 
-    def pop_opt_qemu(self, name: str, default: Any = None, *, parser: Callable[[str], Any] = str, dest: str | None = None) -> str | None:
+    def pop_opt_qemu(
+        self,
+        name: str,
+        default: Any = None,
+        *,
+        parser: Callable[[str], Any] = str,
+        dest: str | None = None,
+    ) -> str | None:
         opt = self.pop_opt(name, parser, default)
         # return DiskArg.qemu_opt(name=qemu if qemu is not None else name, value=opt)
         if opt is None:
@@ -939,19 +954,18 @@ def topology(self) -> dict[str, str]:
         # Get the real device name (handles symlinks like /dev/mapper -> /dev/dm-X)
         real_path = os.path.realpath(self.path, strict=True)
         dev_name = os.path.basename(real_path)
-        sys_base = Path(f'/sys/block/{dev_name}')
+        sys_base = Path(f"/sys/block/{dev_name}")
 
         attributes = {
             # 'alignment': ('alignment_offset', int),
-            'log-sec': ('queue/logical_block_size', int),
-            'phy-sec': ('queue/physical_block_size', int),
-            'min-io': ('queue/minimum_io_size', int),
-            'opt-io': ('queue/optimal_io_size', int),
-            'rota': ('queue/rotational', bool),
+            "log-sec": ("queue/logical_block_size", int),
+            "phy-sec": ("queue/physical_block_size", int),
+            "min-io": ("queue/minimum_io_size", int),
+            "opt-io": ("queue/optimal_io_size", int),
+            "rota": ("queue/rotational", bool),
             # 'wzeroes': ('queue/write_zeroes_max_bytes', int),
-
             # 'disc-aln': ('discard_alignment', int),
-            'disc-gran': ('queue/discard_granularity', int),
+            "disc-gran": ("queue/discard_granularity", int),
             # 'disc-max': ('queue/discard_max_bytes', int),
             # 'disc-zero': ('queue/discard_zeroes_data', bool),
         }
@@ -973,7 +987,7 @@ def topology(self) -> dict[str, str]:
 
     # Validate name=path arguments from --disk and --blk-disk
     @classmethod
-    def parse(cls, func: str, arg: str) -> 'DiskArg':
+    def parse(cls, func: str, arg: str) -> "DiskArg":
         items = arg.split(",")
 
         namefile = items[0]
@@ -998,13 +1012,17 @@ def parse(cls, func: str, arg: str) -> 'DiskArg':
                 opts[key] = "1"
 
         if "help" in opts:
-            print("\n".join([
-                f"Possible {func} options:",
-            ] + [
-                "{:<20} {}".format(f"{key}=({typ})", value)
-                for key, (typ, value) in
-                DiskArg._OPTS_HELP.items()
-            ]))
+            print(
+                "\n".join(
+                    [
+                        f"Possible {func} options:",
+                    ]
+                    + [
+                        "{:<20} {}".format(f"{key}=({typ})", value)
+                        for key, (typ, value) in DiskArg._OPTS_HELP.items()
+                    ]
+                )
+            )
             sys.exit(0)
 
         return cls(
@@ -1707,56 +1725,69 @@ def do_it() -> int:
 
             # we need those parameters multiple times
             discard = disk.pop_opt("discard", parser=strtobool, default=None)
-            detect_zeroes = disk.pop_opt("detect-zeroes", parser=strtobool, default=None)
+            detect_zeroes = disk.pop_opt(
+                "detect-zeroes", parser=strtobool, default=None
+            )
             # we need this parameter both to transform other parameters and as itself later
             # log_sec = disk.get_opt("log-sec", parser=int, default=512)
 
-            drive_opts.extend([
-                disk.pop_opt_qemu("format", "raw"),
-                disk.pop_opt_qemu("cache", None),
-                disk.pop_opt_qemu("aio", None),
-                f"discard={"unmap" if discard else "ignore"}"
-                    if discard is not None else None,
-                f"detect-zeroes={("unmap" if discard else "on") if detect_zeroes else "off"}"
-                    if detect_zeroes is not None else None,
-            ])
-
-            device_opts.extend([
-                f"discard={"on" if discard else "off"}"
-                    if discard is not None else None,
-                disk.pop_opt_qemu("disc-gran", dest="discard_granularity"),
-                disk.pop_opt_qemu("log-sec", dest="logical_block_size"),
-                disk.pop_opt_qemu("phy-sec", dest="physical_block_size"),
-                # disk.pop_qemu("disc-max", dest="max-discard-sectors", parser=lambda arg: int(arg) / log_sec),
-                # disk.pop_qemu("wzeroes", dest="max-write-zeroes-sectors", parser=lambda arg: int(arg) / log_sec),
-                disk.pop_opt_qemu("min-io", dest="min_io_size"),
-                disk.pop_opt_qemu("opt-io", dest="opt_io_size"),
-                disk.pop_opt_qemu("queues", dest="num-queues"),
-            ])
+            drive_opts.extend(
+                [
+                    disk.pop_opt_qemu("format", "raw"),
+                    disk.pop_opt_qemu("cache", None),
+                    disk.pop_opt_qemu("aio", None),
+                    f"discard={'unmap' if discard else 'ignore'}"
+                    if discard is not None
+                    else None,
+                    f"detect-zeroes={('unmap' if discard else 'on') if detect_zeroes else 'off'}"
+                    if detect_zeroes is not None
+                    else None,
+                ]
+            )
+
+            device_opts.extend(
+                [
+                    f"discard={'on' if discard else 'off'}"
+                    if discard is not None
+                    else None,
+                    disk.pop_opt_qemu("disc-gran", dest="discard_granularity"),
+                    disk.pop_opt_qemu("log-sec", dest="logical_block_size"),
+                    disk.pop_opt_qemu("phy-sec", dest="physical_block_size"),
+                    # disk.pop_qemu("disc-max", dest="max-discard-sectors", parser=lambda arg: int(arg) / log_sec),
+                    # disk.pop_qemu("wzeroes", dest="max-write-zeroes-sectors", parser=lambda arg: int(arg) / log_sec),
+                    disk.pop_opt_qemu("min-io", dest="min_io_size"),
+                    disk.pop_opt_qemu("opt-io", dest="opt_io_size"),
+                    disk.pop_opt_qemu("queues", dest="num-queues"),
+                ]
+            )
             # unused
             disk.opts.pop("rota", None)
 
             if disk.pop_opt("iothread", bool, False):
                 iothreadid = f"iothread{iothread_index}"
                 iothread_index += 1
-                qemuargs.extend([
-                    "-object",
-                    f"iothread,id={iothreadid}",
-                ])
-                device_opts.append(
-                    f"iothread={iothreadid}"
+                qemuargs.extend(
+                    [
+                        "-object",
+                        f"iothread,id={iothreadid}",
+                    ]
                 )
+                device_opts.append(f"iothread={iothreadid}")
 
-            qemuargs.extend([
-                "-drive",
-                ",".join(o for o in drive_opts if o is not None),
-                "-device",
-                ",".join(o for o in device_opts if o is not None),
-            ])
+            qemuargs.extend(
+                [
+                    "-drive",
+                    ",".join(o for o in drive_opts if o is not None),
+                    "-device",
+                    ",".join(o for o in device_opts if o is not None),
+                ]
+            )
 
             # any options that were not consumed are errors
             if disk.opts:
-                raise ValueError(f"invalid --disk parameter: {d!r}\n(keys were not consumed: {disk.opts.keys()})")
+                raise ValueError(
+                    f"invalid --disk parameter: {d!r}\n(keys were not consumed: {disk.opts.keys()})"
+                )
 
     if args.disk:
         for i, d in enumerate(args.disk):
@@ -1784,70 +1815,88 @@ def do_it() -> int:
                 "vendor=virtme",
                 "product=disk",
                 f"serial={disk.name}",
-                f"device_id={device_id}"
-                    if device_id != disk.name else None,
+                f"device_id={device_id}" if device_id != disk.name else None,
             ]
 
             # we need those parameters multiple times
             discard = disk.pop_opt("discard", parser=strtobool, default=None)
-            detect_zeroes = disk.pop_opt("detect-zeroes", parser=strtobool, default=None)
+            detect_zeroes = disk.pop_opt(
+                "detect-zeroes", parser=strtobool, default=None
+            )
             # we need this parameter both to transform other parameters and as itself later
             log_sec = disk.get_opt("log-sec")
 
-            drive_opts.extend([
-                disk.pop_opt_qemu("format", "raw"),
-                disk.pop_opt_qemu("cache", None),
-                disk.pop_opt_qemu("aio", None),
-                f"discard={"unmap" if discard else "ignore"}"
-                    if discard is not None else None,
-                f"detect-zeroes={("unmap" if discard else "on") if detect_zeroes else "off"}"
-                    if detect_zeroes is not None else None,
-            ])
-
-            scsi_opts.extend([
-                disk.pop_opt_qemu("queues", dest="num-queues"),
-            ])
-
-            device_opts.extend([
-                disk.pop_opt_qemu("disc-gran", dest="discard_granularity"),
-                disk.pop_opt_qemu("log-sec", dest="logical_block_size"),
-                # convenience: QEMU does not automatically adjust physical_block_size
-                # to be not less than logical_block_size (it errors out instead), so we do it here
-                disk.pop_opt_qemu("phy-sec", dest="physical_block_size", default=log_sec),
-                # disk.pop_qemu("disc-max", dest="max_unmap_size"),
-                # disk.pop_qemu("wzeroes", dest="???"),
-                disk.pop_opt_qemu("min-io", dest="min_io_size"),
-                disk.pop_opt_qemu("opt-io", dest="opt_io_size"),
-                # sic: set rotation_rate to "1" for non-rotating disks ("1" is a special value
-                # that means "non-rotating medium"), but set to "0" for rotating disks
-                # ("0" means "rotation rate not reported").
-                disk.pop_opt_qemu("rota", dest="rotation_rate",
-                    parser=lambda arg: "0" if strtobool(arg) else "1"),
-            ])
+            drive_opts.extend(
+                [
+                    disk.pop_opt_qemu("format", "raw"),
+                    disk.pop_opt_qemu("cache", None),
+                    disk.pop_opt_qemu("aio", None),
+                    f"discard={'unmap' if discard else 'ignore'}"
+                    if discard is not None
+                    else None,
+                    f"detect-zeroes={('unmap' if discard else 'on') if detect_zeroes else 'off'}"
+                    if detect_zeroes is not None
+                    else None,
+                ]
+            )
+
+            scsi_opts.extend(
+                [
+                    disk.pop_opt_qemu("queues", dest="num-queues"),
+                ]
+            )
+
+            device_opts.extend(
+                [
+                    disk.pop_opt_qemu("disc-gran", dest="discard_granularity"),
+                    disk.pop_opt_qemu("log-sec", dest="logical_block_size"),
+                    # convenience: QEMU does not automatically adjust physical_block_size
+                    # to be not less than logical_block_size (it errors out instead), so we do it here
+                    disk.pop_opt_qemu(
+                        "phy-sec", dest="physical_block_size", default=log_sec
+                    ),
+                    # disk.pop_qemu("disc-max", dest="max_unmap_size"),
+                    # disk.pop_qemu("wzeroes", dest="???"),
+                    disk.pop_opt_qemu("min-io", dest="min_io_size"),
+                    disk.pop_opt_qemu("opt-io", dest="opt_io_size"),
+                    # sic: set rotation_rate to "1" for non-rotating disks ("1" is a special value
+                    # that means "non-rotating medium"), but set to "0" for rotating disks
+                    # ("0" means "rotation rate not reported").
+                    disk.pop_opt_qemu(
+                        "rota",
+                        dest="rotation_rate",
+                        parser=lambda arg: "0" if strtobool(arg) else "1",
+                    ),
+                ]
+            )
 
             if disk.pop_opt("iothread", bool, False):
                 iothreadid = f"iothread{iothread_index}"
                 iothread_index += 1
-                qemuargs.extend([
-                    "-object",
-                    f"iothread,id={iothreadid}",
-                ])
-                scsi_opts.append(
-                    f"iothread={iothreadid}"
+                qemuargs.extend(
+                    [
+                        "-object",
+                        f"iothread,id={iothreadid}",
+                    ]
                 )
+                scsi_opts.append(f"iothread={iothreadid}")
 
-            qemuargs.extend([
-                "-drive",
-                ",".join(o for o in drive_opts if o is not None),
-                "-device",
-                ",".join(o for o in scsi_opts if o is not None),
-                "-device",
-                ",".join(o for o in device_opts if o is not None),
-            ])
+            qemuargs.extend(
+                [
+                    "-drive",
+                    ",".join(o for o in drive_opts if o is not None),
+                    "-device",
+                    ",".join(o for o in scsi_opts if o is not None),
+                    "-device",
+                    ",".join(o for o in device_opts if o is not None),
+                ]
+            )
 
             # any options that were not consumed are errors
             if disk.opts:
-                raise ValueError(f"invalid --disk parameter: {d!r}\n(keys were not consumed: {disk.opts.keys()})")
+                raise ValueError(
+                    f"invalid --disk parameter: {d!r}\n(keys were not consumed: {disk.opts.keys()})"
+                )
 
     ret_path = None
 

virtme_ng/run.py

@@ -1180,8 +1180,8 @@ def ensure_name(dsk: str) -> str:
             """
             items = dsk.split(",")
             first = items[0]
-            if '=' not in first:
-                return f'{first}={dsk}'
+            if "=" not in first:
+                return f"{first}={dsk}"
             return dsk
 
         if args.disk is not None:

virtme_ng/utils.py

@@ -85,6 +85,7 @@ def get_conf(key_path):
             conf = conf[key]
         return conf
 
+
 def strtobool(arg: str) -> bool:
     lower = arg.strip().lower()
     if lower in ("yes", "true", "on", "1"):
@@ -94,6 +95,7 @@ def strtobool(arg: str) -> bool:
     else:
         raise ValueError(f"invalid boolean value: {arg!r}")
 
+
 def scsi_device_id(name: str, max_len: int) -> str:
     """
     Trim a longer string which may or may not be a path to fit within `max_len`
@@ -107,7 +109,7 @@ def scsi_device_id(name: str, max_len: int) -> str:
     name = os.path.normpath(name)
     # Try removing path components from the left first
     while len(name) > max_len:
-        left, sep, right = name.partition('/')
+        left, sep, right = name.partition("/")
         if not sep:
             # still too long, truncate from the left
             return name[-max_len:]