curtin-dev team mailing list archive

[Olivier Gayot <mp+494780@xxxxxxxxxxxxxxxxxx>]

Olivier Gayot has proposed merging ~ogayot/curtin:curthooks-add-unittest into curtin:master.

Commit message:
do not squash

Requested reviews:
  curtin developers (curtin-dev)

For more details, see:
https://code.launchpad.net/~ogayot/curtin/+git/curtin/+merge/494780

Improve test coverage for curthooks. There is some refactoring involved.

This was originally part of the curtin apt-download vs apt-install split.
-- 
Your team curtin developers is requested to review the proposed merge of ~ogayot/curtin:curthooks-add-unittest into curtin:master.

diff --git a/curtin/commands/curthooks.py b/curtin/commands/curthooks.py
index a31ef41..6045988 100644
--- a/curtin/commands/curthooks.py
+++ b/curtin/commands/curthooks.py
@@ -1981,6 +1981,78 @@ def uses_grub(machine):
     return True
 
 
+def curthook_zfs_dkms(target: str):
+    # LP: #1742560 prevent zfs-dkms from being installed (Xenial)
+    if distro.lsb_release(target=target)['codename'] == 'xenial':
+        distro.apt_update(target=target)
+        with util.ChrootableTarget(target) as in_chroot:
+            in_chroot.subp(['apt-mark', 'hold', 'zfs-dkms'])
+
+
+def curthook_install_kernel(cfg: dict, target: str) -> None:
+    setup_zipl(cfg, target)
+    setup_kernel_img_conf(target)
+    install_kernel(cfg, target)
+    restore_dist_interfaces(cfg, target)
+    chzdev_persist_active_online(cfg, target)
+
+
+def curthook_cloudinit(cfg: dict, target: str, osfamily,
+                       *, stack_prefix) -> None:
+    # set cloud-init maas datasource
+    if cfg.get('cloudconfig'):
+        handle_cloudconfig(
+            cfg['cloudconfig'],
+            base_dir=paths.target_path(target,
+                                       'etc/cloud/cloud.cfg.d'))
+
+    if osfamily == DISTROS.redhat:
+        # For vmtests to force execute redhat_upgrade_cloud_init, uncomment
+        # the value in examples/tests/centos_defaults.yaml
+        if cfg.get('_ammend_centos_curthooks'):
+            with events.ReportEventStack(
+                    name=stack_prefix + '/upgrading cloud-init',
+                    reporting_enabled=True, level="INFO",
+                    description="Upgrading cloud-init in target"):
+                redhat_upgrade_cloud_init(cfg.get('network', {}), target)
+
+
+def curthook_zpool_cache(target: str) -> None:
+    # check for the zpool cache file and copy to target if present
+    zpool_cache = '/etc/zfs/zpool.cache'
+    if os.path.exists(zpool_cache):
+        copy_zpool_cache(zpool_cache, target)
+
+
+def curthook_zkey(target: str, osfamily, state) -> None:
+    zkey_repository = '/etc/zkey/repository'
+    zkey_used = os.path.join(os.path.split(state['fstab'])[0], "zkey_used")
+    if all(map(os.path.exists, [zkey_repository, zkey_used])):
+        distro.install_packages(['s390-tools-zkey'], target=target,
+                                osfamily=osfamily)
+        copy_zkey_repository(zkey_repository, target)
+
+
+def curthook_crypttab(target: str, state) -> None:
+    # If a crypttab file was created by block_meta than it needs to be
+    # copied onto the target system, and update_initramfs() needs to be
+    # run, so that the cryptsetup hooks are properly configured on the
+    # installed system and it will be able to open encrypted volumes
+    # at boot.
+    crypttab_location = os.path.join(os.path.split(state['fstab'])[0],
+                                     "crypttab")
+    if os.path.exists(crypttab_location):
+        copy_crypttab(crypttab_location, target)
+        update_initramfs(target)
+
+
+def curthook_dmname_rules(target: str, state) -> None:
+    # If udev dname rules were created, copy them to target
+    udev_rules_d = os.path.join(state['scratch'], "rules.d")
+    if os.path.isdir(udev_rules_d):
+        copy_dname_rules(udev_rules_d, target)
+
+
 def builtin_curthooks(cfg: dict, target: str, state: dict):
     """Run the built-in curthooks
 
@@ -2009,11 +2081,7 @@ def builtin_curthooks(cfg: dict, target: str, state: dict):
             disable_overlayroot(cfg, target)
             disable_update_initramfs(cfg, target, machine)
 
-        # LP: #1742560 prevent zfs-dkms from being installed (Xenial)
-        if distro.lsb_release(target=target)['codename'] == 'xenial':
-            distro.apt_update(target=target)
-            with util.ChrootableTarget(target) as in_chroot:
-                in_chroot.subp(['apt-mark', 'hold', 'zfs-dkms'])
+        curthook_zfs_dkms(target)
 
     # packages may be needed prior to installing kernel
     with events.ReportEventStack(
@@ -2045,11 +2113,7 @@ def builtin_curthooks(cfg: dict, target: str, state: dict):
                 name=stack_prefix + '/installing-kernel',
                 reporting_enabled=True, level="INFO",
                 description="installing kernel"):
-            setup_zipl(cfg, target)
-            setup_kernel_img_conf(target)
-            install_kernel(cfg, target)
-            restore_dist_interfaces(cfg, target)
-            chzdev_persist_active_online(cfg, target)
+            curthook_install_kernel(cfg, target)
 
     with events.ReportEventStack(
             name=stack_prefix + '/setting-up-swap',
@@ -2058,22 +2122,7 @@ def builtin_curthooks(cfg: dict, target: str, state: dict):
         add_swap(cfg, target, state.get('fstab'))
 
     if osfamily in {DISTROS.debian, DISTROS.suse, DISTROS.redhat}:
-        # set cloud-init maas datasource
-        if cfg.get('cloudconfig'):
-            handle_cloudconfig(
-                cfg['cloudconfig'],
-                base_dir=paths.target_path(target,
-                                           'etc/cloud/cloud.cfg.d'))
-
-        if osfamily == DISTROS.redhat:
-            # For vmtests to force execute redhat_upgrade_cloud_init, uncomment
-            # the value in examples/tests/centos_defaults.yaml
-            if cfg.get('_ammend_centos_curthooks'):
-                with events.ReportEventStack(
-                        name=stack_prefix + '/upgrading cloud-init',
-                        reporting_enabled=True, level="INFO",
-                        description="Upgrading cloud-init in target"):
-                    redhat_upgrade_cloud_init(cfg.get('network', {}), target)
+        curthook_cloudinit(cfg, target, osfamily, stack_prefix=stack_prefix)
 
     with events.ReportEventStack(
             name=stack_prefix + '/apply-networking-config',
@@ -2113,33 +2162,11 @@ def builtin_curthooks(cfg: dict, target: str, state: dict):
         handle_pollinate_user_agent(cfg, target)
 
     if osfamily == DISTROS.debian:
-        # check for the zpool cache file and copy to target if present
-        zpool_cache = '/etc/zfs/zpool.cache'
-        if os.path.exists(zpool_cache):
-            copy_zpool_cache(zpool_cache, target)
-
-        zkey_repository = '/etc/zkey/repository'
-        zkey_used = os.path.join(os.path.split(state['fstab'])[0], "zkey_used")
-        if all(map(os.path.exists, [zkey_repository, zkey_used])):
-            distro.install_packages(['s390-tools-zkey'], target=target,
-                                    osfamily=osfamily)
-            copy_zkey_repository(zkey_repository, target)
-
-        # If a crypttab file was created by block_meta than it needs to be
-        # copied onto the target system, and update_initramfs() needs to be
-        # run, so that the cryptsetup hooks are properly configured on the
-        # installed system and it will be able to open encrypted volumes
-        # at boot.
-        crypttab_location = os.path.join(os.path.split(state['fstab'])[0],
-                                         "crypttab")
-        if os.path.exists(crypttab_location):
-            copy_crypttab(crypttab_location, target)
-            update_initramfs(target)
+        curthook_zpool_cache(target)
+        curthook_zkey(target, osfamily, state)
+        curthook_crypttab(target, state)
 
-    # If udev dname rules were created, copy them to target
-    udev_rules_d = os.path.join(state['scratch'], "rules.d")
-    if os.path.isdir(udev_rules_d):
-        copy_dname_rules(udev_rules_d, target)
+    curthook_dmname_rules(target, state)
 
     # Setup kernel crash dumps
     with events.ReportEventStack(
diff --git a/tests/unittests/test_curthooks.py b/tests/unittests/test_curthooks.py
index a775573..b6a46de 100644
--- a/tests/unittests/test_curthooks.py
+++ b/tests/unittests/test_curthooks.py
@@ -1,8 +1,10 @@
 # This file is part of curtin. See LICENSE file for copyright and license info.
 
+import argparse
+import contextlib
 import os
 from pathlib import Path
-from unittest.mock import call, Mock, patch
+from unittest.mock import ANY, call, Mock, patch
 import textwrap
 from typing import Optional
 
@@ -2881,4 +2883,184 @@ class TestDpkgReconfigure(CiTestCase):
             env=fk_env,
         )
 
+
+@patch("curtin.commands.curthooks.setup_zipl")
+@patch("curtin.commands.curthooks.setup_kernel_img_conf")
+@patch("curtin.commands.curthooks.install_kernel")
+@patch("curtin.commands.curthooks.restore_dist_interfaces")
+@patch("curtin.commands.curthooks.chzdev_persist_active_online")
+class TestCurthookInstallKernel(CiTestCase):
+    def test_simple(
+            self, m_chzdev_persist_active_online, m_restore_dist_interfaces,
+            m_install_kernel, m_setup_kernel_img_conf, m_setup_zipl):
+        cfg = {}
+        target = "/tgt"
+
+        curthooks.curthook_install_kernel(cfg, target)
+
+        m_setup_zipl.assert_called_once_with({}, "/tgt")
+        m_setup_kernel_img_conf.assert_called_once_with("/tgt")
+        m_install_kernel.assert_called_once_with(cfg, "/tgt")
+        m_restore_dist_interfaces.assert_called_once_with(cfg, "/tgt")
+        m_chzdev_persist_active_online.assert_called_once_with(cfg, "/tgt")
+
+
+@patch("curtin.commands.curthooks.redhat_upgrade_cloud_init")
+@patch("curtin.commands.curthooks.handle_cloudconfig")
+class TestCurthookCloudinit(CiTestCase):
+    def test_no_cloudconfig(self, m_handle_cloudconfig, m_redhat_upgrade):
+        curthooks.curthook_cloudinit(
+                {}, "/tgt", distro.DISTROS.debian, stack_prefix="")
+        m_handle_cloudconfig.assert_not_called()
+        m_redhat_upgrade.assert_not_called()
+
+    def test_with_cloudconfig(self, m_handle_cloudconfig, m_redhat_upgrade):
+        curthooks.curthook_cloudinit(
+                {"cloudconfig": "<CLOUDCONFIG>"},
+                "/tgt", distro.DISTROS.debian, stack_prefix="")
+        m_handle_cloudconfig.assert_called_once_with(
+                "<CLOUDCONFIG>", base_dir="/tgt/etc/cloud/cloud.cfg.d")
+        m_redhat_upgrade.assert_not_called()
+
+    def test_redhat_no_ammend_centos(
+            self, m_handle_cloudconfig, m_redhat_upgrade):
+        curthooks.curthook_cloudinit(
+                {}, "/tgt", distro.DISTROS.redhat, stack_prefix="")
+        m_handle_cloudconfig.assert_not_called()
+        m_redhat_upgrade.assert_not_called()
+
+    def test_redhat_with_ammend_centos(
+            self, m_handle_cloudconfig, m_redhat_upgrade):
+        curthooks.curthook_cloudinit(
+                {"network": "<NETCONFIG>",
+                 "_ammend_centos_curthooks": True},
+                "/tgt", distro.DISTROS.redhat, stack_prefix="")
+        m_handle_cloudconfig.assert_not_called()
+        m_redhat_upgrade.assert_called_once_with("<NETCONFIG>", "/tgt")
+
+
+@patch("curtin.commands.curthooks.platform.machine",
+       Mock(return_value="amd64"))
+@patch("curtin.commands.curthooks.events.ReportEventStack",
+       Mock(return_value=contextlib.nullcontext()))
+@patch("curtin.commands.curthooks.distro.get_distroinfo")
+class TestBuiltinCurthooks(CiTestCase):
+    def test_ubuntu_noble(self, m_distroinfo):
+        prefix = "curtin.commands.curthooks."
+
+        cfg = {}
+        state = {"fstab": "/etc/fstab"}
+
+        expected = [
+            ("do_apt_config", call(cfg, "/tgt")),
+            ("disable_overlayroot", call(cfg, "/tgt")),
+            ("disable_update_initramfs", call(cfg, "/tgt", "amd64")),
+            ("curthook_zfs_dkms", call("/tgt")),
+            ("install_missing_packages",
+             call(cfg, "/tgt", osfamily=distro.DISTROS.debian)),
+            ("configure_iscsi",
+             call(cfg, "/etc", "/tgt", osfamily=distro.DISTROS.debian)),
+            ("configure_mdadm",
+             call(cfg, "/etc", "/tgt", osfamily=distro.DISTROS.debian)),
+            ("configure_nvme_over_tcp", call(cfg, Path("/tgt"))),
+            ("curthook_install_kernel", call(cfg, "/tgt")),
+            ("add_swap", call(cfg, "/tgt", "/etc/fstab")),
+            ("curthook_cloudinit",
+             call(cfg, "/tgt", distro.DISTROS.debian, stack_prefix="")),
+            ("apply_networking", call("/tgt", state)),
+            ("copy_fstab", call("/etc/fstab", "/tgt")),
+            ("detect_and_handle_multipath",
+             call(cfg, "/tgt", osfamily=distro.DISTROS.debian)),
+            ("system_upgrade",
+             call(cfg, "/tgt", osfamily=distro.DISTROS.debian)),
+            ("redhat_apply_selinux_autorelabel", None),
+            ("handle_pollinate_user_agent", call(cfg, "/tgt")),
+            ("curthook_zpool_cache", call("/tgt")),
+            ("curthook_zkey", call("/tgt", distro.DISTROS.debian, state)),
+            ("curthook_crypttab", call("/tgt", state)),
+            ("curthook_dmname_rules", call("/tgt", state)),
+            ("configure_kernel_crash_dumps", call(cfg, Path("/tgt"))),
+            ("enable_update_initramfs", call(cfg, "/tgt", "amd64")),
+            ("reconfigure_kernel", call("/tgt")),
+            ("redhat_update_initramfs", None),
+            ("setup_boot",
+             call(cfg, "/tgt", "amd64", "",
+                  osfamily=distro.DISTROS.debian, variant="ubuntu")),
+            ("copy_cdrom", call("/cdrom", "/tgt")),
+        ]
+        m_distroinfo.return_value = distro.DistroInfo(
+                "ubuntu", distro.DISTROS.debian)
+
+        with contextlib.ExitStack() as stack:
+            cms = []
+            for name, m_call in expected:
+                symbol = f"{prefix}{name}"
+                cms.append((
+                    stack.enter_context(patch(symbol, autospec=True)),
+                    m_call))
+
+            curthooks.builtin_curthooks(cfg, "/tgt", state)
+
+        for m, m_call in cms:
+            if m_call is None:
+                m.assert_not_called()
+            else:
+                self.assertEqual(m.mock_calls, [m_call])
+
+
+@patch("curtin.commands.curthooks.util.run_hook_if_exists",
+       return_value=False)
+@patch("curtin.commands.curthooks.distro.is_ubuntu_core",
+       return_value=False)
+@patch("curtin.commands.curthooks.ubuntu_core_curthooks")
+@patch("curtin.commands.curthooks.builtin_curthooks")
+@patch("curtin.commands.curthooks.util.EFIVarFSBug", Mock())
+@patch("curtin.commands.curthooks.events.ReportEventStack",
+       Mock(return_value=contextlib.nullcontext()))
+class TestCurthooks(CiTestCase):
+    def test_ubuntu(
+            self, m_builtin_curthooks, m_uc_curthooks, m_is_uc, m_run_hookk):
+        with self.assertRaises(SystemExit) as m_exit:
+            curthooks.curthooks(
+                    argparse.Namespace(target="/target", config={}))
+        self.assertEqual(0, m_exit.exception.code)
+        m_builtin_curthooks.assert_called_once_with({}, "/target", ANY)
+        m_uc_curthooks.assert_not_called()
+
+    def test_ubuntu_core(
+            self, m_builtin_curthooks, m_uc_curthooks, m_is_uc, m_run_hook):
+        m_is_uc.return_value = True
+        with self.assertRaises(SystemExit) as m_exit:
+            curthooks.curthooks(
+                    argparse.Namespace(target="/target", config={}))
+        self.assertEqual(0, m_exit.exception.code)
+        m_builtin_curthooks.assert_not_called()
+        m_uc_curthooks.assert_called_once_with({}, "/target")
+
+    def test_no_target(
+            self, m_builtin_curthooks, m_uc_curthooks, m_is_uc, m_run_hook):
+        with self.assertRaises(SystemExit) as m_exit:
+            curthooks.curthooks(argparse.Namespace(target=None, config={}))
+        self.assertEqual(2, m_exit.exception.code)
+
+    def test_existing_curtin_hooks(
+            self, m_builtin_curthooks, m_uc_curthooks, m_is_uc, m_run_hook):
+        m_run_hook.return_value = True
+        with self.assertRaises(SystemExit) as m_exit:
+            curthooks.curthooks(
+                    argparse.Namespace(target="/target", config={}))
+        self.assertEqual(0, m_exit.exception.code)
+        m_builtin_curthooks.assert_not_called()
+        m_uc_curthooks.assert_not_called()
+
+        m_run_hook.reset_mock()
+        # If we explicitly ask for builtin curthooks...
+        with self.assertRaises(SystemExit) as m_exit:
+            curthooks.curthooks(argparse.Namespace(
+                target="/target",
+                config={"curthooks": {"mode": "builtin"}}))
+        m_run_hook.assert_not_called()
+        m_builtin_curthooks.assert_called()
+
+
 # vi: ts=4 expandtab syntax=python