From: Guillaume Abrioux <gabrioux@ibm.com>
Date: Tue, 7 Jan 2025 09:48:48 +0000 (+0000)
Subject: ceph-volume: add typing hints to lvm.zap
X-Git-Tag: v19.2.3~349^2~20
X-Git-Url: http://git.apps.os.sepia.ceph.com/?a=commitdiff_plain;h=09d9c15140d65a84b0f79901fb1b50df51d37c18;p=ceph.git

ceph-volume: add typing hints to lvm.zap

This adds the python typing hints to lvm.zap

Signed-off-by: Guillaume Abrioux <gabrioux@ibm.com>
(cherry picked from commit 046921e29684dbec25131ada8ef11a4f400c63d9)
---

diff --git a/src/ceph-volume/ceph_volume/devices/lvm/zap.py b/src/ceph-volume/ceph_volume/devices/lvm/zap.py
index 92cc13714f9ae..2b6c1be31e71f 100644
--- a/src/ceph-volume/ceph_volume/devices/lvm/zap.py
+++ b/src/ceph-volume/ceph_volume/devices/lvm/zap.py
@@ -42,7 +42,7 @@ def zap_bluestore(path: str) -> None:
         '--yes-i-really-really-mean-it'
     ])
 
-def wipefs(path):
+def wipefs(path: str) -> None:
     """
     Removes the filesystem from an lv or partition.
 
@@ -74,7 +74,7 @@ def wipefs(path):
     raise RuntimeError("could not complete wipefs on device: %s" % path)
 
 
-def zap_data(path):
+def zap_data(path: str) -> None:
     """
     Clears all data from the given path. Path should be
     an absolute path to an lv or partition.
@@ -121,7 +121,7 @@ class Zap:
 
         return list(raw_devices)
 
-    def find_associated_devices(self) -> List[api.Volume]:
+    def find_associated_devices(self) -> List[Device]:
         """From an ``osd_id`` and/or an ``osd_fsid``, filter out all the Logical Volumes (LVs) in the
         system that match those tag values, further detect if any partitions are
         part of the OSD, and then return the set of LVs and partitions (if any).
@@ -278,7 +278,10 @@ class Zap:
         Device examples: vg-name/lv-name, /dev/vg-name/lv-name
         Requirements: Must be a logical volume (LV)
         """
-        lv: api.Volume = device.lv_api
+        if device.lv_api is not None:
+            lv: api.Volume = device.lv_api
+        else:
+            raise RuntimeError(f"Unexpected error while attempting to zap LV device {device}.")
         self.unmount_lv(lv)
         self.parent_device: str = disk.get_parent_device_from_mapper(lv.lv_path)
         zap_device(device.path)
@@ -297,7 +300,7 @@ class Zap:
                     'db': self.args.replace_db,
                     'wal': self.args.replace_wal
                 }
-                if replacement_args.get(lv.tags.get('ceph.type'), False):
+                if replacement_args.get(lv.tags.get('ceph.type', ''), False):
                     mlogger.info(f'Marking {self.parent_device} as being replaced')
                     self._write_replacement_header(self.parent_device)
             else: