feat(snap): avoid refresh on package_upgrade: true and refresh.hold

cloudinit/distros/package_management/snap.py

@@ -35,4 +35,23 @@ def install_packages(self, pkglist: Iterable) -> UninstalledPackages:
 
     @staticmethod
     def upgrade_packages():
-        subp.subp(["snap", "refresh"])
+        command = ["snap", "get", "system", "-d"]
+        snap_hold = None
+        try:
+            result = subp.subp(command)
+            snap_hold = (
+                util.load_json(result.stdout).get("refresh", {}).get("hold")
+            )
+        except subp.ProcessExecutionError as e:
+            LOG.info(
+                "Continuing to snap refresh. Unable to run command: %s: %s",
+                command,
+                e,
+            )
+        if snap_hold == "forever":
+            LOG.info(
+                "Skipping snap refresh because refresh.hold is set to '%s'",
+                snap_hold,
+            )
+        else:
+            subp.subp(["snap", "refresh"])

doc/module-docs/cc_package_update_upgrade_install/data.yaml

@@ -1,13 +1,16 @@
 cc_package_update_upgrade_install:
   description: |
     This module allows packages to be updated, upgraded or installed during
-    boot. If any packages are to be installed or an upgrade is to be performed
-    then the package cache will be updated first. If a package installation or
-    upgrade requires a reboot, then a reboot can be performed if
-    ``package_reboot_if_required`` is specified.
+    boot using any available package manager present on a system such as apt,
+    pkg, snap, yum or zypper. If any packages are to be installed or an upgrade
+    is to be performed then the package cache will be updated first. If a
+    package installation or upgrade requires a reboot, then a reboot can be
+    performed if ``package_reboot_if_required`` is specified.
   examples:
   - comment: |
       Example 1:
     file: cc_package_update_upgrade_install/example1.yaml
+  - comment: "By default, ``package_upgrade: true`` performs upgrades on any installed package manager. To avoid calling ``snap refresh`` in images with snap installed, set snap refresh.hold to ``forever`` will prevent cloud-init's snap interaction during any boot"
+    file: cc_package_update_upgrade_install/example2.yaml
   name: Package Update Upgrade Install
   title: Update, upgrade, and install packages

doc/module-docs/cc_package_update_upgrade_install/example2.yaml

@@ -0,0 +1,7 @@
+#cloud-config
+package_update: true
+package_upgrade: true
+snap:
+  commands:
+    00: snap refresh --hold=forever
+package_reboot_if_required: true

tests/integration_tests/modules/test_package_update_upgrade_install.py

@@ -86,6 +86,40 @@ def test_snap_packages_are_installed(self, class_client):
         assert "curl" in output
         assert "postman" in output
 
+    def test_snap_refresh_not_called_when_refresh_hold_forever(
+        self, class_client
+    ):
+        """Assert snap refresh is not called when snap refresh --hold is set.
+
+        Certain network-limited or secure environments may opt to avoid
+        contacting snap API endpoints. In those scenarios, it is expected
+        that automated snap refresh is held for all snaps. Typically, this is
+        done with snap refresh --hold in those environments.
+
+        Assert cloud-init does not attempt to call snap refresh when
+        refresh.hold is forever.
+        """
+        assert class_client.execute(
+            [
+                "grep",
+                r"Running command \['snap', 'refresh'",
+                "/var/log/cloud-init.log",
+            ]
+        ).ok
+        assert class_client.execute("snap refresh --hold").ok
+        class_client.instance.clean()
+        class_client.restart()
+        assert class_client.execute(
+            [
+                "grep",
+                r"Running command \['snap', 'refresh']",
+                "/var/log/cloud-init.log",
+            ]
+        ).failed
+        assert class_client.execute(
+            "grep 'Skipping snap refresh' /var/log/cloud-init.log"
+        ).ok
+
 
 HELLO_VERSIONS_BY_RELEASE = {
     "oracular": "2.10-3build2",

tests/unittests/config/test_cc_package_update_upgrade_install.py

@@ -122,7 +122,7 @@ def _isfile(filename: str):
 
         caplog.set_level(logging.WARNING)
         with mock.patch(
-            "cloudinit.subp.subp", return_value=("fakeout", "fakeerr")
+            "cloudinit.subp.subp", return_value=SubpResult("{}", "fakeerr")
         ) as m_subp:
             with mock.patch("os.path.isfile", side_effect=_isfile):
                 with mock.patch(M_PATH + "time.sleep") as m_sleep:

tests/unittests/distros/test_ubuntu.py

@@ -1,7 +1,10 @@
 # This file is part of cloud-init. See LICENSE file for license information.
+import logging
+
 import pytest
 
 from cloudinit.distros import fetch
+from cloudinit.subp import SubpResult
 
 
 class TestPackageCommand:
@@ -14,7 +17,7 @@ def test_package_command_only_refresh_snap_when_available(
             "cloudinit.distros.ubuntu.Snap.available",
             return_value=snap_available,
         )
-        m_snap_upgrade_packges = mocker.patch(
+        m_snap_upgrade_packages = mocker.patch(
             "cloudinit.distros.ubuntu.Snap.upgrade_packages",
             return_value=snap_available,
         )
@@ -27,6 +30,81 @@ def test_package_command_only_refresh_snap_when_available(
         m_apt_run_package_command.assert_called_once_with("upgrade")
         m_snap_available.assert_called_once()
         if snap_available:
-            m_snap_upgrade_packges.assert_called_once()
+            m_snap_upgrade_packages.assert_called_once()
+        else:
+            m_snap_upgrade_packages.assert_not_called()
+
+    @pytest.mark.parametrize(
+        "subp_side_effect,expected_log",
+        (
+            pytest.param(
+                [
+                    SubpResult(
+                        stdout='{"refresh": {"hold": "forever"}}', stderr=None
+                    )
+                ],
+                "Skipping snap refresh because refresh.hold is set to"
+                " 'forever'",
+                id="skip_snap_refresh_due_to_global_hold_forever",
+            ),
+            pytest.param(
+                [
+                    SubpResult(
+                        stdout=(
+                            '{"refresh": {"hold":'
+                            ' "2024-07-08T15:38:20-06:00"}}'
+                        ),
+                        stderr=None,
+                    ),
+                    SubpResult(stdout="All snaps up to date.", stderr=""),
+                ],
+                "",
+                id="perform_snap_refresh_due_to_temporary_global_hold",
+            ),
+            pytest.param(
+                [
+                    SubpResult(
+                        stdout="{}",
+                        stderr=(
+                            'error: snap "core" has no "refresh.hold" '
+                            "configuration option"
+                        ),
+                    ),
+                    SubpResult(stdout="All snaps up to date.", stderr=""),
+                ],
+                "",
+                id="snap_refresh_performed_when_no_global_hold_is_set",
+            ),
+        ),
+    )
+    def test_package_command_avoids_snap_refresh_when_refresh_hold_is_forever(
+        self, subp_side_effect, expected_log, caplog, mocker
+    ):
+        """Do not call snap refresh when snap refresh.hold is forever.
+
+        This indicates an environment where snaps refreshes are not preferred
+        for whatever reason.
+        """
+        m_snap_available = mocker.patch(
+            "cloudinit.distros.ubuntu.Snap.available",
+            return_value=True,
+        )
+        m_subp = mocker.patch(
+            "cloudinit.subp.subp",
+            side_effect=subp_side_effect,
+        )
+        m_apt_run_package_command = mocker.patch(
+            "cloudinit.distros.package_management.apt.Apt.run_package_command",
+        )
+        cls = fetch("ubuntu")
+        distro = cls("ubuntu", {}, None)
+        with caplog.at_level(logging.INFO):
+            distro.package_command("upgrade")
+        m_apt_run_package_command.assert_called_once_with("upgrade")
+        m_snap_available.assert_called_once()
+        expected_calls = [mocker.call(["snap", "get", "system", "-d"])]
+        if expected_log:
+            assert expected_log in caplog.text
         else:
-            m_snap_upgrade_packges.assert_not_called()
+            expected_calls.append(mocker.call(["snap", "refresh"]))
+        assert m_subp.call_args_list == expected_calls