refactor: Make device info part of distro definition (#5067)

cloudinit/config/cc_growpart.py

@@ -292,49 +292,6 @@ def get_size(filename) -> Optional[int]:
             os.close(fd)
 
 
-def device_part_info(devpath):
-    # convert an entry in /dev/ to parent disk and partition number
-
-    # input of /dev/vdb or /dev/disk/by-label/foo
-    # rpath is hopefully a real-ish path in /dev (vda, sdb..)
-    rpath = os.path.realpath(devpath)
-
-    bname = os.path.basename(rpath)
-    syspath = "/sys/class/block/%s" % bname
-
-    if util.is_BSD():
-        # FreeBSD doesn't know of sysfs so just get everything we need from
-        # the device, like /dev/vtbd0p2.
-        fpart = "/dev/" + util.find_freebsd_part(devpath)
-        # Handle both GPT partitions and MBR slices with partitions
-        m = re.search(
-            r"^(?P<dev>/dev/.+)[sp](?P<part_slice>\d+[a-z]*)$", fpart
-        )
-        if m:
-            return m["dev"], m["part_slice"]
-
-    if not os.path.exists(syspath):
-        raise ValueError("%s had no syspath (%s)" % (devpath, syspath))
-
-    ptpath = os.path.join(syspath, "partition")
-    if not os.path.exists(ptpath):
-        raise TypeError("%s not a partition" % devpath)
-
-    ptnum = util.load_text_file(ptpath).rstrip()
-
-    # for a partition, real syspath is something like:
-    # /sys/devices/pci0000:00/0000:00:04.0/virtio1/block/vda/vda1
-    rsyspath = os.path.realpath(syspath)
-    disksyspath = os.path.dirname(rsyspath)
-
-    diskmajmin = util.load_text_file(os.path.join(disksyspath, "dev")).rstrip()
-    diskdevpath = os.path.realpath("/dev/block/%s" % diskmajmin)
-
-    # diskdevpath has something like 253:0
-    # and udev has put links in /dev/block/253:0 to the device name in /dev/
-    return (diskdevpath, ptnum)
-
-
 def devent2dev(devent):
     if devent.startswith("/dev/"):
         return devent
@@ -544,7 +501,7 @@ def resize_devices(resizer, devices, distro: Distro):
             # though we should probably grow the ability to
             continue
         try:
-            (disk, ptnum) = device_part_info(blockdev)
+            disk, ptnum = distro.device_part_info(blockdev)
         except (TypeError, ValueError) as e:
             info.append(
                 (

cloudinit/distros/__init__.py

@@ -1338,7 +1338,6 @@ def eject_media(device: str) -> None:
             )
         subp.subp(cmd)
 
-
     @staticmethod
     def get_mapped_device(blockdev: str) -> Optional[str]:
         """Returns underlying block device for a mapped device.
@@ -1357,6 +1356,42 @@ def get_mapped_device(blockdev: str) -> Optional[str]:
             return realpath
         return None
 
+    @staticmethod
+    def device_part_info(devpath: str) -> tuple:
+        """convert an entry in /dev/ to parent disk and partition number
+
+        input of /dev/vdb or /dev/disk/by-label/foo
+        rpath is hopefully a real-ish path in /dev (vda, sdb..)
+        """
+        rpath = os.path.realpath(devpath)
+
+        bname = os.path.basename(rpath)
+        syspath = "/sys/class/block/%s" % bname
+
+        if not os.path.exists(syspath):
+            raise ValueError("%s had no syspath (%s)" % (devpath, syspath))
+
+        ptpath = os.path.join(syspath, "partition")
+        if not os.path.exists(ptpath):
+            raise TypeError("%s not a partition" % devpath)
+
+        ptnum = util.load_text_file(ptpath).rstrip()
+
+        # for a partition, real syspath is something like:
+        # /sys/devices/pci0000:00/0000:00:04.0/virtio1/block/vda/vda1
+        rsyspath = os.path.realpath(syspath)
+        disksyspath = os.path.dirname(rsyspath)
+
+        diskmajmin = util.load_text_file(
+            os.path.join(disksyspath, "dev")
+        ).rstrip()
+        diskdevpath = os.path.realpath("/dev/block/%s" % diskmajmin)
+
+        # diskdevpath has something like 253:0
+        # and udev has put links in /dev/block/253:0 to the device
+        # name in /dev/
+        return diskdevpath, ptnum
+
 
 def _apply_hostname_transformations_to_url(url: str, transformations: list):
     """

cloudinit/distros/bsd.py

@@ -1,5 +1,6 @@
 import logging
 import platform
+import re
 from typing import List, Optional
 
 import cloudinit.net.netops.bsd_netops as bsd_netops
@@ -153,3 +154,19 @@ def get_proc_ppid(pid):
     @staticmethod
     def get_mapped_device(blockdev: str) -> Optional[str]:
         return None
+
+    @staticmethod
+    def device_part_info(devpath: str) -> tuple:
+        # FreeBSD doesn't know of sysfs so just get everything we need from
+        # the device, like /dev/vtbd0p2.
+        part = util.find_freebsd_part(devpath)
+        if part:
+            fpart = f"/dev/{part}"
+            # Handle both GPT partitions and MBR slices with partitions
+            m = re.search(
+                r"^(?P<dev>/dev/.+)[sp](?P<part_slice>\d+[a-z]*)$", fpart
+            )
+            if m:
+                return m["dev"], m["part_slice"]
+
+        return distros.Distro.device_part_info(devpath)

tests/unittests/config/test_cc_growpart.py

@@ -21,6 +21,7 @@
     get_schema,
     validate_cloudconfig_schema,
 )
+from cloudinit.distros.bsd import BSD
 from cloudinit.subp import SubpResult
 from tests.unittests.helpers import (
     TestCase,
@@ -365,9 +366,9 @@ def mystat(path):
                 raise e
             return real_stat(path)
 
-        opinfo = cc_growpart.device_part_info
+        opinfo = self.distro.device_part_info
         try:
-            cc_growpart.device_part_info = simple_device_part_info
+            self.distro.device_part_info = simple_device_part_info
             os.stat = mystat
 
             resized = cc_growpart.resize_devices(
@@ -389,11 +390,8 @@ def find(name, res):
             self.assertEqual(
                 cc_growpart.RESIZE.SKIPPED, find(enoent[0], resized)[1]
             )
-            # self.assertEqual(resize_calls,
-            #                 [("/dev/XXda", "1", "/dev/XXda1"),
-            #                  ("/dev/YYda", "2", "/dev/YYda2")])
         finally:
-            cc_growpart.device_part_info = opinfo
+            self.distro.device_part_info = opinfo
             os.stat = real_stat
 
 
@@ -469,10 +467,9 @@ def assert_no_resize_or_cleanup(self):
     def common_mocks(self, mocker):
         # These are all "happy path" mocks which will get overridden
         # when needed
-        mocker.patch(
-            "cloudinit.config.cc_growpart.device_part_info",
-            side_effect=self._device_part_info_side_effect,
-        )
+        self.distro = MockDistro
+        original_device_part_info = self.distro.device_part_info
+        self.distro.device_part_info = self._device_part_info_side_effect
         mocker.patch("os.stat")
         mocker.patch("stat.S_ISBLK")
         mocker.patch("stat.S_ISCHR")
@@ -503,10 +500,10 @@ def common_mocks(self, mocker):
         mocker.patch("pathlib.Path.exists", return_value=True)
         self.m_unlink = mocker.patch("pathlib.Path.unlink", autospec=True)
 
-        self.distro = MockDistro
-
         self.resizer = mock.Mock()
         self.resizer.resize = mock.Mock(return_value=(1024, 1024))
+        yield
+        self.distro.device_part_info = original_device_part_info
 
     def test_resize_when_encrypted(self, common_mocks, caplog):
         info = cc_growpart.resize_devices(
@@ -529,9 +526,7 @@ def test_resize_when_encrypted(self, common_mocks, caplog):
         self.assert_resize_and_cleanup()
 
     def test_resize_when_unencrypted(self, common_mocks):
-        info = cc_growpart.resize_devices(
-            self.resizer, ["/"], self.distro
-        )
+        info = cc_growpart.resize_devices(self.resizer, ["/"], self.distro)
         assert len(info) == 1
         assert info[0][0] == "/"
         assert "encrypted" not in info[0][2]
@@ -672,31 +667,25 @@ def __init__(self, **kwds):
 
 class TestDevicePartInfo:
     @pytest.mark.parametrize(
-        "devpath, is_BSD, expected, raised_exception",
+        "devpath, expected, raised_exception",
         (
             pytest.param(
                 "/dev/vtbd0p2",
-                True,
                 ("/dev/vtbd0", "2"),
                 does_not_raise(),
                 id="gpt_partition",
             ),
             pytest.param(
                 "/dev/vbd0s3a",
-                True,
                 ("/dev/vbd0", "3a"),
                 does_not_raise(),
                 id="bsd_mbr_slice_and_partition",
             ),
         ),
     )
-    @mock.patch("cloudinit.util.is_BSD")
-    def test_device_part_info(
-        self, m_is_BSD, is_BSD, devpath, expected, raised_exception
-    ):
-        m_is_BSD.return_value = is_BSD
+    def test_device_part_info(self, devpath, expected, raised_exception):
         with raised_exception:
-            assert expected == cc_growpart.device_part_info(devpath)
+            assert expected == BSD.device_part_info(devpath)
 
 
 class TestGrowpartSchema: