chore: add type annotation to cloudinit.config.cc_growpart

cloudinit/config/cc_growpart.py

@@ -18,7 +18,7 @@
 from abc import ABC, abstractmethod
 from contextlib import suppress
 from pathlib import Path
-from typing import Optional, Tuple
+from typing import List, Optional, Tuple
 
 from cloudinit import lifecycle, performance, subp, temp_utils, util
 from cloudinit.cloud import Cloud
@@ -63,14 +63,16 @@ def __init__(self, distro: Distro):
         self._distro = distro
 
     @abstractmethod
-    def available(self, devices: list) -> bool: ...
+    def available(self, devices: List[str]) -> bool: ...
 
     @abstractmethod
-    def resize(self, diskdev, partnum, partdev, fs): ...
+    def resize(
+        self, diskdev: str, partnum: str, partdev: str, fs: Optional[str]
+    ) -> Tuple[Optional[int], Optional[int]]: ...
 
 
 class ResizeGrowPart(Resizer):
-    def available(self, devices: list):
+    def available(self, devices: list[str]) -> bool:
         try:
             out = subp.subp(
                 ["growpart", "--help"], update_env={"LANG": "C"}
@@ -82,7 +84,9 @@ def available(self, devices: list):
             pass
         return False
 
-    def resize(self, diskdev, partnum, partdev, fs):
+    def resize(
+        self, diskdev: str, partnum: str, partdev: str, fs: Optional[str]
+    ) -> Tuple[Optional[int], Optional[int]]:
         before = get_size(partdev, fs)
 
         # growpart uses tmp dir to store intermediate states
@@ -131,11 +135,13 @@ class ResizeGrowFS(Resizer):
     doesn't need to be enabled in ``rc.conf``.
     """
 
-    def available(self, devices: list):
+    def available(self, devices: list[str]) -> bool:
         """growfs only works on the root partition"""
         return os.path.isfile("/etc/rc.d/growfs") and devices == ["/"]
 
-    def resize(self, diskdev, partnum, partdev, fs):
+    def resize(
+        self, diskdev: str, partnum: str, partdev: str, fs: Optional[str]
+    ) -> Tuple[Optional[int], Optional[int]]:
         before = get_size(partdev, fs)
         try:
             self._distro.manage_service(action="onestart", service="growfs")
@@ -147,7 +153,7 @@ def resize(self, diskdev, partnum, partdev, fs):
 
 
 class ResizeGpart(Resizer):
-    def available(self, devices: list):
+    def available(self, devices: list[str]) -> bool:
         try:
             err = subp.subp(
                 ["gpart", "help"], update_env={"LANG": "C"}, rcs=[0, 1]
@@ -159,7 +165,9 @@ def available(self, devices: list):
             pass
         return False
 
-    def resize(self, diskdev, partnum, partdev, fs):
+    def resize(
+        self, diskdev: str, partnum: str, partdev: str, fs: Optional[str]
+    ) -> Tuple[Optional[int], Optional[int]]:
         """
         GPT disks store metadata at the beginning (primary) and at the
         end (secondary) of the disk. When launching an image with a
@@ -184,7 +192,7 @@ def resize(self, diskdev, partnum, partdev, fs):
         return (before, get_size(partdev, fs))
 
 
-def resizer_factory(mode: str, distro: Distro, devices: list) -> Resizer:
+def resizer_factory(mode: str, distro: Distro, devices: List[str]) -> Resizer:
     resize_class = None
     if mode == "auto":
         for _name, resizer in RESIZERS:
@@ -214,7 +222,7 @@ def resizer_factory(mode: str, distro: Distro, devices: list) -> Resizer:
     return resize_class
 
 
-def get_size(filename, fs) -> Optional[int]:
+def get_size(filename: str, fs: Optional[str]) -> Optional[int]:
     fd = None
     try:
         fd = os.open(filename, os.O_RDONLY)
@@ -228,7 +236,7 @@ def get_size(filename, fs) -> Optional[int]:
             os.close(fd)
 
 
-def get_zfs_size(dataset) -> Optional[int]:
+def get_zfs_size(dataset: str) -> Optional[int]:
     zpool = dataset.split("/")[0]
     try:
         size, _ = subp.subp(["zpool", "get", "-Hpovalue", "size", zpool])
@@ -238,7 +246,7 @@ def get_zfs_size(dataset) -> Optional[int]:
     return int(size.strip())
 
 
-def devent2dev(devent):
+def devent2dev(devent: str) -> Tuple[str, Optional[str]]:
     if devent.startswith("/dev/"):
         return devent, None
 
@@ -263,7 +271,7 @@ def devent2dev(devent):
     return dev, fs
 
 
-def is_encrypted(blockdev, partition) -> bool:
+def is_encrypted(blockdev: str, partition: str) -> bool:
     """
     Check if a device is an encrypted device. blockdev should have
     a /dev/dm-* path whereas partition is something like /dev/sda1.
@@ -307,7 +315,7 @@ def get_underlying_partition(blockdev: str) -> str:
         ) from e
 
 
-def resize_encrypted(blockdev, partition) -> Tuple[str, str]:
+def resize_encrypted(blockdev: str, partition: str) -> Tuple[str, str]:
     """Use 'cryptsetup resize' to resize LUKS volume.
 
     The loaded keyfile is json formatted with 'key' and 'slot' keys.
@@ -364,8 +372,15 @@ def resize_encrypted(blockdev, partition) -> Tuple[str, str]:
     )
 
 
-def _call_resizer(resizer, devent, disk, ptnum, blockdev, fs):
-    info = []
+def _call_resizer(
+    resizer: Resizer,
+    devent: str,
+    disk: str,
+    ptnum: str,
+    blockdev: str,
+    fs: Optional[str],
+) -> List[Tuple[str, str, str]]:
+    info: List[Tuple[str, str, str]] = []
     try:
         old, new = resizer.resize(disk, ptnum, blockdev, fs)
         if old == new:
@@ -378,7 +393,7 @@ def _call_resizer(resizer, devent, disk, ptnum, blockdev, fs):
             )
         elif new is None or old is None:
             msg = ""
-            if disk is not None and ptnum is None:
+            if disk and not ptnum:
                 msg = "changed (%s, %s) size, new size is unknown" % (
                     disk,
                     ptnum,
@@ -388,7 +403,7 @@ def _call_resizer(resizer, devent, disk, ptnum, blockdev, fs):
             info.append((devent, RESIZE.CHANGED, msg))
         else:
             msg = ""
-            if disk is not None and ptnum is None:
+            if disk and not ptnum:
                 msg = "changed (%s, %s) from %s to %s" % (
                     disk,
                     ptnum,
@@ -410,15 +425,17 @@ def _call_resizer(resizer, devent, disk, ptnum, blockdev, fs):
     return info
 
 
-def resize_devices(resizer: Resizer, devices, distro: Distro):
+def resize_devices(
+    resizer: Resizer, devices: List[str], distro: Distro
+) -> List[Tuple[str, str, str]]:
     # returns a tuple of tuples containing (entry-in-devices, action, message)
     devices = copy.copy(devices)
-    info = []
+    info: List[Tuple[str, str, str]] = []
 
     while devices:
         devent = devices.pop(0)
-        disk = None
-        ptnum = None
+        disk = ""
+        ptnum = ""
 
         try:
             blockdev, fs = devent2dev(devent)