improve upgrade-snaps action to better recover in error cases

charms/kubernetes_snaps.py

@@ -24,6 +24,10 @@ class ExternalCloud(Protocol):
     name: Optional[str]
 
 
+class SnapInstallError(Exception):
+    """Raised when a snap install fails for a detectable reason."""
+
+
 log = logging.getLogger(__name__)
 service_account_key_path = Path("/root/cdk/serviceaccount.key")
 tls_ciphers_intermediate = [
@@ -649,12 +653,14 @@ def install(channel, control_plane=False, upgrade=False):
             channel,
             ",".join(sorted(missing)),
         )
-        status.add(BlockedStatus(f"Not all snaps are available on channel={channel}"))
-        return
+        msg = f"Not all snaps are available on channel={channel}"
+        status.add(BlockedStatus(msg))
+        raise SnapInstallError(msg)
 
     if any(is_channel_swap(snap, channel) for snap in which_snaps) and not upgrade:
-        status.add(BlockedStatus("Needs manual upgrade, run the upgrade action."))
-        return
+        msg = "Needs manual upgrade, run the upgrade action."
+        status.add(BlockedStatus(msg))
+        raise SnapInstallError(msg)
 
     # Refresh with ignore_running=True ONLY for non-daemon apps (i.e. kubectl)
     # https://bugs.launchpad.net/bugs/1987331
@@ -803,24 +809,30 @@ def set_default_cni_conf_file(cni_conf_file):
 
 
 def upgrade_snaps(channel: str, event: ActionEvent, control_plane: bool = False):
-    log.info(f"Starting the upgrade of Kubernetes snaps to '{channel}' channel.")
+    """Upgrade the snaps from an upgrade action event."""
+    log_it = f"Starting the upgrade of Kubernetes snaps to {channel}."
+    event.log(log_it)
+    log.info(log_it)
+    error_message = None
+
     try:
         install(channel=channel, control_plane=control_plane, upgrade=True)
-    except (CalledProcessError, Exception) as e:
-        if isinstance(e, CalledProcessError):
-            error_message = f"Upgrade failed with a process error. stdout: {e.stdout}, stderr: {e.stderr}"
+    except status.ReconcilerError as e:
+        ec = e.__context__
+        if isinstance(ec, CalledProcessError):
+            error_message = f"Upgrade failed with a process error. stdout: {ec.stdout}, stderr: {ec.stderr}"
+        elif isinstance(ec, SnapInstallError):
+            error_message = f"Upgrade failed with a detectable error: {ec}"
         else:
-            error_message = f"An unexpected error occurred during the upgrade: {e}"
-
+            error_message = f"An unexpected error occurred during the upgrade: {ec}"
         log.exception(error_message)
-        status.add("Snap upgrade failed. Check action results for more information.")
+
+    if error_message:
         event.fail(error_message)
     else:
-        result_message = (
-            f"Successfully upgraded Kubernetes snaps to the '{channel}' channel."
-        )
-        log.info(result_message)
-        event.set_results({"result": result_message})
+        log_it = f"Successfully upgraded Kubernetes snaps to the {channel}."
+        log.info(log_it)
+        event.set_results({"result": log_it})
 
 
 def v1_taint_from_string(taint: str):

tests/unit/test_kubernetes_snaps.py

@@ -2,6 +2,7 @@
 import unittest.mock as mock
 
 from charms import kubernetes_snaps
+import charms.contextual_status as status
 
 
 @pytest.fixture(autouse=True)
@@ -22,18 +23,37 @@ def subprocess_call():
 def test_upgrade_action_control_plane(is_channel_available, is_channel_swap, caplog):
     mock_event = mock.MagicMock()
     channel = "1.28/edge"
-    kubernetes_snaps.upgrade_snaps(channel, mock_event, control_plane=True)
+    with status.context(mock_event.model.unit):
+        kubernetes_snaps.upgrade_snaps(channel, mock_event, control_plane=True)
     snaps = kubernetes_snaps.BASIC_SNAPS + kubernetes_snaps.CONTROL_PLANE_SNAPS
     is_channel_available.assert_has_calls([mock.call(s, channel) for s in snaps])
     is_channel_swap.assert_has_calls([mock.call(s, channel) for s in snaps])
+    assert f"Starting the upgrade of Kubernetes snaps to {channel}." in caplog.messages
     assert (
-        f"Starting the upgrade of Kubernetes snaps to '{channel}' channel."
-        in caplog.messages
-    )
-    assert (
-        f"Successfully upgraded Kubernetes snaps to the '{channel}' channel."
-        in caplog.messages
+        f"Successfully upgraded Kubernetes snaps to the {channel}." in caplog.messages
     )
+    mock_event.set_results.assert_called_once()
+    mock_event.fail.assert_not_called()
+
+
+@mock.patch.object(kubernetes_snaps, "is_channel_swap", return_value=False)
+@mock.patch.object(kubernetes_snaps, "is_channel_available", return_value=False)
+@mock.patch.object(kubernetes_snaps, "install_snap", mock.MagicMock())
+def test_upgrade_action_control_plane_fails_available(
+    is_channel_available, is_channel_swap, caplog
+):
+    mock_event = mock.MagicMock()
+    channel = "1.28/edge"
+    with status.context(mock_event.model.unit):
+        kubernetes_snaps.upgrade_snaps(channel, mock_event, control_plane=True)
+    snaps = kubernetes_snaps.BASIC_SNAPS + kubernetes_snaps.CONTROL_PLANE_SNAPS
+    is_channel_available.assert_has_calls([mock.call(s, channel) for s in snaps])
+    is_channel_swap.assert_not_called()
+    assert "Starting the upgrade of Kubernetes snaps to" in caplog.messages[0]
+    assert "The following snaps do not have a revision on channel" in caplog.messages[1]
+    assert "Upgrade failed with a detectable error" in caplog.messages[2]
+    mock_event.fail.assert_called_once()
+    mock_event.set_results.assert_not_called()
 
 
 def test_is_snap_available(subprocess_check_output):

tox.ini

@@ -18,23 +18,20 @@ set_env =
 
 [testenv:format]
 description = Apply coding style standards to code
-deps =
-    black
-    ruff
+deps = ruff
 commands =
-    black {[vars]all_path}
     ruff format {[vars]all_path}
+    ruff check --fix {[vars]all_path}
 
 [testenv:lint]
 description = Check code against coding style standards
 deps =
-    black
     ruff
+    tomli
     codespell
 commands =
     codespell {toxinidir}
     ruff check {[vars]all_path}
-    black --check --diff {[vars]all_path}
 
 [testenv:unit]
 deps =