pre-commit reformats

pkgs/ceph/nautilus/rbd-locktool.py

@@ -50,7 +50,9 @@ def _query_locker(self):
         lockers = self._rbd("lock", "--format", "json", "list", self.name)
         lockers = json.loads(lockers.decode())
         if len(lockers) > 1:
-            raise RuntimeError("cannot handle multiple (shared) locks", lockers)
+            raise RuntimeError(
+                "cannot handle multiple (shared) locks", lockers
+            )
         try:
             locker = lockers[0]
         except IndexError:
@@ -109,7 +111,9 @@ def main():
     if args.quiet and args.verbose:
         argp.error("-q and -v don't go together")
     if not (args.lock ^ args.unlock ^ args.info):
-        argp.error('must specify exactly one of "--lock", "--unlock", "--info"')
+        argp.error(
+            'must specify exactly one of "--lock", "--unlock", "--info"'
+        )
     logging.basicConfig(
         stream=sys.stdout,
         format="{}: %(message)s".format(argp.prog),

pkgs/fc/agent/fc/manage/createvm.py

@@ -115,7 +115,9 @@ def partition(self):
             )
         )
         cmd(
-            "sgdisk {} -n 2:2048:+1M -c 2:gptbios -t 2:EF02".format(self.device)
+            "sgdisk {} -n 2:2048:+1M -c 2:gptbios -t 2:EF02".format(
+                self.device
+            )
         )
 
     def apply(self):

pkgs/fc/agent/fc/manage/dhcpd.py

@@ -349,7 +349,9 @@ def main():
     dhcpd.query_directory()
     if options.output:
         conffile = fc.util.configfile.ConfigFile(options.output)
-        conffile.write(dhcpd.render(options.include, options.local_include_dir))
+        conffile.write(
+            dhcpd.render(options.include, options.local_include_dir)
+        )
         changed = conffile.commit()
     else:
         sys.stdout.write(dhcpd.render(options.include))

pkgs/fc/ceph/src/fc/ceph/backup.py

@@ -11,7 +11,11 @@
 class BackupManager:
     @staticmethod
     def create(
-        name: str, vgname: str, disks: list[str], encrypt: bool, mountpoint: str
+        name: str,
+        vgname: str,
+        disks: list[str],
+        encrypt: bool,
+        mountpoint: str,
     ):
         console.print(
             f"Creating new backup volume {vgname}/{name} on disks {', '.join(disks)}…"

pkgs/fc/ceph/src/fc/ceph/keys/nautilus.py

@@ -199,7 +199,9 @@ def compare_key(self):
         return self.entity, self.key.to_string(), self.capabilities
 
     def save_client_keyring(self):
-        f = fc.util.configfile.ConfigFile(self.filename, mode=0o600, diff=False)
+        f = fc.util.configfile.ConfigFile(
+            self.filename, mode=0o600, diff=False
+        )
         f.write(
             f"""\
 [{self.entity}]

pkgs/fc/ceph/src/fc/ceph/luks/manage.py

@@ -23,7 +23,9 @@ class LuksDevice(NamedTuple):
     header: Optional[str] = None
 
     @classmethod
-    def lsblk_to_cryptdevices(cls, lsblk_blockdevs: list) -> list["LuksDevice"]:
+    def lsblk_to_cryptdevices(
+        cls, lsblk_blockdevs: list
+    ) -> list["LuksDevice"]:
         """parses the output of lsblk -Js -o NAME,PATH,TYPE,MOUNTPOINT"""
         return [
             cls(
@@ -97,7 +99,8 @@ def create(self, device):
 
     def destroy(self, overwrite=True):
         console.print(
-            f"Destroying keystore in {self.volume.mountpoint} ...", style="bold"
+            f"Destroying keystore in {self.volume.mountpoint} ...",
+            style="bold",
         )
         base_disk = self.volume.lv.base_disk
         self.volume.purge()
@@ -174,7 +177,9 @@ def _do_rekey(self, slot: str, device: str, header: Optional[str]):
 
         header_arg = ["--header", header] if header else []
 
-        dump = run.cryptsetup("luksDump", *header_arg, device, encoding="ascii")
+        dump = run.cryptsetup(
+            "luksDump", *header_arg, device, encoding="ascii"
+        )
         if f"  {slot_id}: luks2" in dump:
             run.cryptsetup(
                 "luksKillSlot",
@@ -216,7 +221,9 @@ def check_luks(name_glob: str, header: Optional[str]) -> int:
                     "luksDump", "--header", dev.header, dev.base_blockdev
                 )
             else:
-                luks_dump = Cryptsetup.cryptsetup("luksDump", dev.base_blockdev)
+                luks_dump = Cryptsetup.cryptsetup(
+                    "luksDump", dev.base_blockdev
+                )
             dump_lines = luks_dump.decode("utf-8").splitlines()
             for check in all_checks:
                 check_ok = True

pkgs/fc/ceph/src/fc/ceph/lvm.py

@@ -133,7 +133,9 @@ def blockdevice(self):
                     self._blockdevice = partition
                     break
             else:
-                raise RuntimeError(f"Could not find partition 1 on {self.name}")
+                raise RuntimeError(
+                    f"Could not find partition 1 on {self.name}"
+                )
         return self._blockdevice
 
     @classmethod
@@ -269,7 +271,10 @@ def encrypted(self):
         raise NotImplementedError
 
     def _create(
-        self, base_device: Optional[GenericBlockDevice], size: str, vg_name: str
+        self,
+        base_device: Optional[GenericBlockDevice],
+        size: str,
+        vg_name: str,
     ):
         raise NotImplementedError
 
@@ -353,7 +358,10 @@ def activate(self):
         self._vg_name = lv[0]["vg_name"]
 
     def _create(
-        self, base_device: Optional[GenericBlockDevice], size: str, vg_name: str
+        self,
+        base_device: Optional[GenericBlockDevice],
+        size: str,
+        vg_name: str,
     ):
         self.ensure_vg(vg_name, base_device)
 
@@ -399,7 +407,9 @@ def purge(self, lv_only=False):
             return
 
         try:
-            pv = run.json.pvs("-S", f"vg_name=~^{re.escape(self._vg_name)}$")[0]
+            pv = run.json.pvs("-S", f"vg_name=~^{re.escape(self._vg_name)}$")[
+                0
+            ]
         except IndexError:
             pass
         else:
@@ -464,7 +474,10 @@ def expected_mapper_name(self) -> str:
         return self.name
 
     def _create(
-        self, base_device: Optional[GenericBlockDevice], size: str, vg_name: str
+        self,
+        base_device: Optional[GenericBlockDevice],
+        size: str,
+        vg_name: str,
     ):
         self.underlay.create(
             name=self.underlay.name,

pkgs/fc/ceph/src/fc/ceph/main.py

@@ -363,7 +363,8 @@ def ceph(args=sys.argv[1:]):
         "maintenance", help="Perform maintenance tasks."
     )
     maint.set_defaults(
-        subsystem=fc.ceph.maintenance.MaintenanceTasks, action=maint.print_usage
+        subsystem=fc.ceph.maintenance.MaintenanceTasks,
+        action=maint.print_usage,
     )
     maint_sub = maint.add_subparsers()
 
@@ -382,10 +383,14 @@ def ceph(args=sys.argv[1:]):
     )
     parser_clean_deleted_vms.set_defaults(action="clean_deleted_vms")
 
-    parser_enter = maint_sub.add_parser("enter", help="Enter maintenance mode.")
+    parser_enter = maint_sub.add_parser(
+        "enter", help="Enter maintenance mode."
+    )
     parser_enter.set_defaults(action="enter")
 
-    parser_leave = maint_sub.add_parser("leave", help="Leave maintenance mode.")
+    parser_leave = maint_sub.add_parser(
+        "leave", help="Leave maintenance mode."
+    )
     parser_leave.set_defaults(action="leave")
 
     # extract parsed arguments from object into a dict

pkgs/fc/ceph/src/fc/ceph/osd/nautilus.py

@@ -26,7 +26,8 @@ def _eval_health_check(status, checkname):
         """
         try:
             check_unclean = (
-                status["health"]["checks"][checkname]["severity"] != "HEALTH_OK"
+                status["health"]["checks"][checkname]["severity"]
+                != "HEALTH_OK"
             )
         except KeyError:
             # clean checks generally do not appear in output
@@ -220,7 +221,9 @@ def reactivate(self, ids: str):
             osd = OSD(id_)
             wait_for_clean_cluster()
             try:
-                self._deactivate_single(osd, as_systemd_unit=False, flush=False)
+                self._deactivate_single(
+                    osd, as_systemd_unit=False, flush=False
+                )
                 self._activate_single(osd, as_systemd_unit=False)
             except Exception:
                 traceback.print_exc()
@@ -288,7 +291,10 @@ def prepare_ext_vg(device: str):
         try:
             partition_table = run.json.sfdisk(device)["partitiontable"]
         except CalledProcessError as e:
-            if b"does not contain a recognized partition table" not in e.stderr:
+            if (
+                b"does not contain a recognized partition table"
+                not in e.stderr
+            ):
                 # Not an empty disk, propagate the error
                 raise
         else:
@@ -584,7 +590,9 @@ def purge(self, no_safety_check: bool, strict_safety_check: bool):
         # Delete OSD object
         while True:
             try:
-                run.ceph("osd", "purge", str(self.id), "--yes-i-really-mean-it")
+                run.ceph(
+                    "osd", "purge", str(self.id), "--yes-i-really-mean-it"
+                )
             except CalledProcessError as e:
                 # OSD is still shutting down, keep trying.
                 if e.returncode == errno.EBUSY:
@@ -608,7 +616,9 @@ def _collect_rebuild_information(self):
 
         # what's the crush location (host?)
         crush_location = "host={0}".format(
-            run.json.ceph("osd", "find", str(self.id))["crush_location"]["host"]
+            run.json.ceph("osd", "find", str(self.id))["crush_location"][
+                "host"
+            ]
         )
 
         print(f"--crush-location={crush_location}")

pkgs/fc/ceph/src/fc/ceph/tests/test_luks.py

@@ -659,15 +659,21 @@ def test_luks_fingerprint_verify(
     inputs_mock.seek(0)
 
     # no fingerprint file
-    assert mock_LUKSKeyStoreManager.fingerprint(confirm=False, verify=True) == 1
+    assert (
+        mock_LUKSKeyStoreManager.fingerprint(confirm=False, verify=True) == 1
+    )
 
     # mismatching fingerprint file
     persist_fingerprint(b"notfoo", tmpdir)
-    assert mock_LUKSKeyStoreManager.fingerprint(confirm=False, verify=True) == 1
+    assert (
+        mock_LUKSKeyStoreManager.fingerprint(confirm=False, verify=True) == 1
+    )
 
     # matching fingerprint file
     persist_fingerprint(b"foo", tmpdir)
-    assert mock_LUKSKeyStoreManager.fingerprint(confirm=False, verify=True) == 0
+    assert (
+        mock_LUKSKeyStoreManager.fingerprint(confirm=False, verify=True) == 0
+    )
 
     captured = capsys.readouterr()
 

pkgs/fc/ceph/src/fc/ceph/tests/test_maintenance.py

@@ -63,14 +63,22 @@ def test_successful_maintenance_cycle(
         {
             "nodes": [
                 {"name": "localhost", "type": "host", "children": [14]},
-                {"name": "localhost-ssd", "type": "host", "children": [27, 13]},
+                {
+                    "name": "localhost-ssd",
+                    "type": "host",
+                    "children": [27, 13],
+                },
             ]
         },
         # osd tree
         {
             "nodes": [
                 {"name": "localhost", "type": "host", "children": [14]},
-                {"name": "localhost-ssd", "type": "host", "children": [27, 13]},
+                {
+                    "name": "localhost-ssd",
+                    "type": "host",
+                    "children": [27, 13],
+                },
             ]
         },
     ]
@@ -86,7 +94,9 @@ def test_successful_maintenance_cycle(
 
     ceph_calls.assert_has_calls(
         [
-            mock.call("osd", "set-group", "noup", "localhost", "localhost-ssd"),
+            mock.call(
+                "osd", "set-group", "noup", "localhost", "localhost-ssd"
+            ),
             mock.call("osd", "down", "13", "14", "27"),
             mock.call(
                 "osd", "unset-group", "noup", "localhost", "localhost-ssd"

pkgs/fc/ceph/src/fc/ceph/util/directory.py

@@ -55,7 +55,9 @@ def wrapper(*args):
         # recognizable value.
         wrapper.__qualname__ = "DirectoryAPI." + name
 
-        retry = stamina.retry(on=RETRY_EXCEPTIONS, wait_exp_base=10, attempts=2)
+        retry = stamina.retry(
+            on=RETRY_EXCEPTIONS, wait_exp_base=10, attempts=2
+        )
         return retry(wrapper)
 
     def __repr__(self):

pkgs/fc/check-ceph/nautilus/tests/test_check_snapshot_restore.py

@@ -191,7 +191,9 @@ def test_parse_pools(
     assert parsed_pools[1].root.used == 65970697666560
 
 
-def test_parse_pools_no_stats_available(example_thresholds, default_pool_roots):
+def test_parse_pools_no_stats_available(
+    example_thresholds, default_pool_roots
+):
     (parse_status, parsed_pools) = snapcheck.parse_pools(
         iter([]), default_pool_roots, example_thresholds
     )

pkgs/fc/check-link-redundancy/check_link_redundancy.py

@@ -37,10 +37,14 @@ def main():
             switches_unique.update(data.keys())
 
     if len(switches_all) != len(args.interfaces):
-        print("CRITICAL - interfaces are missing visible peer devices in LLDP ")
+        print(
+            "CRITICAL - interfaces are missing visible peer devices in LLDP "
+        )
         sys.exit(2)
     elif len(switches_all) != len(switches_unique):
-        print("CRITICAL - multiple interfaces are connected to the same switch")
+        print(
+            "CRITICAL - multiple interfaces are connected to the same switch"
+        )
         sys.exit(2)
     else:
         print("OK - interfaces are connected to different switches")

pkgs/fc/check-rib-integrity/check_rib_integrity.py

@@ -124,7 +124,9 @@ def vtysh_load_evpn_rib():
 
     return {
         int(outer_vni): {
-            mac: hops for inner_vni, mac, hops in macs if inner_vni == outer_vni
+            mac: hops
+            for inner_vni, mac, hops in macs
+            if inner_vni == outer_vni
         }
         for outer_vni in {mac[0] for mac in macs}
     }
@@ -155,7 +157,9 @@ def check_unicast_rib(args):
         if len(set(dests)) != len(dests):
             critical(
                 "duplicate addresses in kernel routing table",
-                "duplicate-kernel-address {}".format(" ".join(fmtaddrs(dests))),
+                "duplicate-kernel-address {}".format(
+                    " ".join(fmtaddrs(dests))
+                ),
             )
 
         for entry in data:

pkgs/fc/ping-on-tap/ping-on-tap.py

@@ -33,7 +33,9 @@ def print_reply(self, req, reply):
         if self.arp_am.is_request(req):
             self.arp_am.print_reply(req, reply)
         else:
-            print("Replying %s to %s" % (reply.getlayer(scapy.IP).dst, req.dst))
+            print(
+                "Replying %s to %s" % (reply.getlayer(scapy.IP).dst, req.dst)
+            )
 
     def make_reply(self, req):
         if self.arp_am.is_request(req):