tools/image-info: cosmetics, autopep8

This commit is contained in:
Thomas Lavocat 2023-02-07 10:01:07 +01:00 committed by Achilleas Koutsou
parent 36d890d452
commit 15d80a0cd1

View file

@ -41,7 +41,7 @@ def run_ostree(*args, _input=None, _check=True, **kwargs):
return res return res
def loop_open(devmgr:devices.DeviceManager, name:str, image, size, offset=0): def loop_open(devmgr: devices.DeviceManager, name: str, image, size, offset=0):
""" """
Uses a DeviceManager to open the `name` at `offset` Uses a DeviceManager to open the `name` at `offset`
Retuns a Device object and the path onto wich the image was loopback mounted Retuns a Device object and the path onto wich the image was loopback mounted
@ -64,6 +64,7 @@ def loop_open(devmgr:devices.DeviceManager, name:str, image, size, offset=0):
"path": os.path.join("/dev", reply["path"]) "path": os.path.join("/dev", reply["path"])
} }
@contextlib.contextmanager @contextlib.contextmanager
def convert_image(image, fmt): def convert_image(image, fmt):
with tempfile.TemporaryDirectory(dir="/var/tmp") as tmp: with tempfile.TemporaryDirectory(dir="/var/tmp") as tmp:
@ -232,7 +233,7 @@ def read_partition(device, partition):
else: else:
blkid = {} blkid = {}
partition["label"] = blkid.get("LABEL") # doesn't exist for mbr partition["label"] = blkid.get("LABEL") # doesn't exist for mbr
partition["uuid"] = blkid.get("UUID") partition["uuid"] = blkid.get("UUID")
partition["fstype"] = blkid.get("TYPE") partition["fstype"] = blkid.get("TYPE")
return partition return partition
@ -409,7 +410,7 @@ def rpm_verify(tree):
# verify user and group ownership: # verify user and group ownership:
# https://github.com/rpm-software-management/rpm/issues/882 # https://github.com/rpm-software-management/rpm/issues/882
rpm = subprocess.Popen(["chroot", tree, "rpm", "--verify", "--all"], rpm = subprocess.Popen(["chroot", tree, "rpm", "--verify", "--all"],
stdout=subprocess.PIPE, encoding="utf-8") stdout=subprocess.PIPE, encoding="utf-8")
changed = {} changed = {}
missing = [] missing = []
@ -528,7 +529,8 @@ def read_services(tree, state):
"chrony-wait.service" "chrony-wait.service"
] ]
""" """
services_state = subprocess_check_output(["systemctl", f"--root={tree}", "list-unit-files"], (lambda s: parse_unit_files(s, state))) services_state = subprocess_check_output(
["systemctl", f"--root={tree}", "list-unit-files"], (lambda s: parse_unit_files(s, state)))
# Since systemd v246, some services previously reported as "enabled" / # Since systemd v246, some services previously reported as "enabled" /
# "disabled" are now reported as "alias". There is no systemd command, that # "disabled" are now reported as "alias". There is no systemd command, that
@ -538,7 +540,8 @@ def read_services(tree, state):
# pre/post v246 systemd versions, check all "alias" units and append them # pre/post v246 systemd versions, check all "alias" units and append them
# to the list, if their target is also listed in 'services_state'. # to the list, if their target is also listed in 'services_state'.
if state != "alias": if state != "alias":
services_alias = subprocess_check_output(["systemctl", f"--root={tree}", "list-unit-files"], (lambda s: parse_unit_files(s, "alias"))) services_alias = subprocess_check_output(
["systemctl", f"--root={tree}", "list-unit-files"], (lambda s: parse_unit_files(s, "alias")))
for alias in services_alias: for alias in services_alias:
# The service may be in one of the following places (output of # The service may be in one of the following places (output of
@ -2467,7 +2470,7 @@ def volume_group_for_device(device: str) -> str:
continue continue
if res.returncode != 0: if res.returncode != 0:
raise RuntimeError(res.stderr.strip()) raise RuntimeError(res.stderr.strip())
vg_name = res.stdout.strip() vg_name = res.stdout.strip()
if vg_name: if vg_name:
@ -2483,7 +2486,7 @@ def ensure_device_file(path: str, major: int, minor: int):
os.mknod(path, 0o600 | stat.S_IFBLK, os.makedev(major, minor)) os.mknod(path, 0o600 | stat.S_IFBLK, os.makedev(major, minor))
def discover_lvm(dev:str, parent:devices.Device, devmgr:devices.DeviceManager): def discover_lvm(dev: str, parent: devices.Device, devmgr: devices.DeviceManager):
# find the volume group name for the device file # find the volume group name for the device file
vg_name = volume_group_for_device(dev) vg_name = volume_group_for_device(dev)
@ -2535,7 +2538,7 @@ def discover_lvm(dev:str, parent:devices.Device, devmgr:devices.DeviceManager):
raise RuntimeError("can't find org.osbuild.lvm2.lv") raise RuntimeError("can't find org.osbuild.lvm2.lv")
jsonschema.validate(options, info.get_schema()) jsonschema.validate(options, info.get_schema())
reply = devmgr.open(device) reply = devmgr.open(device)
voldev = reply["path"] # get the path where is mounted the device voldev = reply["path"] # get the path where is mounted the device
minor = reply["node"]["minor"] minor = reply["node"]["minor"]
major = reply["node"]["major"] major = reply["node"]["major"]
@ -2579,11 +2582,11 @@ def append_partitions(report, image):
for part in partitions: for part in partitions:
start, size = part["start"], part["size"] start, size = part["start"], part["size"]
ret = loop_open( ret = loop_open(
devmgr, devmgr,
part["partuuid"], part["partuuid"],
image, image,
size, size,
offset=start) offset=start)
dev = ret["path"] dev = ret["path"]
devices_map[dev] = ret["Device"] devices_map[dev] = ret["Device"]
read_partition(dev, part) read_partition(dev, part)
@ -2685,7 +2688,7 @@ def append_partitions(report, image):
mmgr.mount(mounts.Mount( mmgr.mount(mounts.Mount(
part_device, part_device,
info, info,
devices_map[part_device], # retrieves the associated Device Object devices_map[part_device], # retrieves the associated Device Object
part_mountpoint, part_mountpoint,
options)) options))
@ -2703,11 +2706,11 @@ def analyse_image(image) -> Dict[str, Any]:
size = os.stat(target).st_size size = os.stat(target).st_size
with host.ServiceManager(monitor=monitor.NullMonitor(1)) as mgr: with host.ServiceManager(monitor=monitor.NullMonitor(1)) as mgr:
device = loop_open( device = loop_open(
devices.DeviceManager(mgr, "/dev", os.path.dirname(target)), devices.DeviceManager(mgr, "/dev", os.path.dirname(target)),
os.path.basename(target), os.path.basename(target),
target, target,
size, size,
offset=0)["path"] offset=0)["path"]
report["bootloader"] = read_bootloader_type(device) report["bootloader"] = read_bootloader_type(device)
report.update(read_partition_table(device)) report.update(read_partition_table(device))
if not report["partition-table"]: if not report["partition-table"]:
@ -2836,6 +2839,7 @@ def analyse_compressed(path):
image = os.path.join(tmpdir, files[0]) image = os.path.join(tmpdir, files[0])
return analyse_image(image) return analyse_image(image)
def is_iso(path): def is_iso(path):
return "iso" in pathlib.Path(path).suffix return "iso" in pathlib.Path(path).suffix