fix(upgrades): grant k8s cluster trust before refreshing octavia-k8s

sunbeam-python/sunbeam/core/juju.py

@@ -1569,6 +1569,27 @@ def _wait_until_status(status: "jubilant.statustypes.Status"):
                 timeout=timeout,
             )
 
+    def is_k8s_model(self, model: str) -> bool:
+        """Return True if the model is a k8s (CAAS) model.
+
+        :param model: Name of the model
+        """
+        return self.get_model(model).get("model-type") == "caas"
+
+    def charm_trust(self, application_name: str, model: str) -> None:
+        """Grant cluster-scoped trust to a k8s charm application.
+
+        On k8s models, ``juju refresh --trust`` does not create a
+        ClusterRoleBinding. Only ``juju trust <app> --scope=cluster``
+        creates the binding required for hooks that access the k8s API
+        (e.g. patching StatefulSets).
+
+        :param application_name: Name of application
+        :param model: Model containing the application
+        """
+        with self._model(model) as juju:
+            juju.trust(application_name, scope="cluster")
+
     def charm_refresh(
         self,
         application_name: str,
@@ -1585,9 +1606,12 @@ def charm_refresh(
         :param channel: Channel to refresh to, if None uses current channel
         :param revision: Revision to refresh to, if None uses latest revision
         :param base: Select a different base than is currently running
-        :param trust: If true, allows charm to run hooks that require access to
-            cloud credentials
+        :param trust: If true, grants cluster-scoped k8s RBAC trust before
+            refresh so that upgrade-charm hooks can access the k8s API.
+            On non-k8s models, trust is passed directly to juju refresh.
         """
+        if trust and self.is_k8s_model(model):
+            self.charm_trust(application_name, model)
         with self._model(model) as juju:
             juju.refresh(
                 application_name,

sunbeam-python/sunbeam/steps/upgrades/intra_channel.py

@@ -51,6 +51,11 @@
 
 INFRA_APPS = ["mysql-k8s", "vault-k8s", "k8s"]
 
+# Charms that must be refreshed with trust=True so that their upgrade-charm
+# hook has the necessary k8s RBAC permissions (e.g. get/patch StatefulSets).
+# octavia-k8s needs trust to remove legacy containers during upgrade.
+CHARMS_REQUIRING_TRUST = {"octavia-k8s"}
+
 # Snap-based charm applications that expose a refresh-snap action.
 # These need to be refreshed explicitly after the charm refresh because
 # their snaps are held to prevent spontaneous snapd auto-refreshes.
@@ -214,10 +219,11 @@ def refresh_apps(
             if charm in INFRA_APPS:
                 continue
             manifest_charm = self.manifest.find_charm(charm)
+            trust = charm in CHARMS_REQUIRING_TRUST
 
             if not manifest_charm:
                 LOG.debug(f"Running refresh for app {app_name} (no manifest entry)")
-                self.jhelper.charm_refresh(app_name, model)
+                self.jhelper.charm_refresh(app_name, model, trust=trust)
                 refreshed_apps.append(app_name)
             else:
                 LOG.debug(f"Running refresh for app {app_name} with manifest config")
@@ -226,6 +232,7 @@ def refresh_apps(
                     model,
                     channel=manifest_charm.channel,
                     revision=manifest_charm.revision,
+                    trust=trust,
                 )
                 refreshed_apps.append(app_name)
 

sunbeam-python/tests/unit/sunbeam/core/test_juju.py

@@ -289,13 +289,50 @@ def test_charm_refresh_with_base(jhelper, juju):
     )
 
 
-def test_charm_refresh_with_trust(jhelper, juju):
-    jhelper.charm_refresh("app", "test-model", trust=True)
+def test_charm_refresh_with_trust_on_k8s_model(jhelper, juju):
+    """On a k8s model, trust=True triggers juju.trust(scope=cluster) before refresh."""
+    with patch.object(jhelper, "is_k8s_model", return_value=True):
+        jhelper.charm_refresh("app", "test-model", trust=True)
+    juju.trust.assert_called_once_with("app", scope="cluster")
     juju.refresh.assert_called_with(
         "app", channel=None, revision=None, base=None, trust=True
     )
 
 
+def test_charm_refresh_with_trust_on_machine_model(jhelper, juju):
+    """On a machine model with trust=True, juju.trust must NOT be called."""
+    with patch.object(jhelper, "is_k8s_model", return_value=False):
+        jhelper.charm_refresh("app", "test-model", trust=True)
+    juju.trust.assert_not_called()
+    juju.refresh.assert_called_with(
+        "app", channel=None, revision=None, base=None, trust=True
+    )
+
+
+def test_charm_refresh_without_trust_does_not_call_juju_trust(jhelper, juju):
+    """When trust=False (default), juju.trust must not be called."""
+    jhelper.charm_refresh("app", "test-model")
+    juju.trust.assert_not_called()
+
+
+def test_charm_trust(jhelper, juju):
+    """charm_trust calls juju.trust with scope=cluster."""
+    jhelper.charm_trust("app", "test-model")
+    juju.trust.assert_called_once_with("app", scope="cluster")
+
+
+def test_is_k8s_model_caas(jhelper):
+    """is_k8s_model returns True for caas model-type."""
+    with patch.object(jhelper, "get_model", return_value={"model-type": "caas"}):
+        assert jhelper.is_k8s_model("openstack") is True
+
+
+def test_is_k8s_model_iaas(jhelper):
+    """is_k8s_model returns False for iaas model-type."""
+    with patch.object(jhelper, "get_model", return_value={"model-type": "iaas"}):
+        assert jhelper.is_k8s_model("openstack-machines") is False
+
+
 def test_get_spaces(jhelper):
     juju_mock = MagicMock()
     juju_mock.cli = MagicMock(return_value=json.dumps({"spaces": [{"name": "space1"}]}))

sunbeam-python/tests/unit/sunbeam/steps/upgrades/test_intra_channel.py

@@ -55,8 +55,8 @@ def test_refresh_apps_no_manifest_entry(self):
         # Execute
         result = self.upgrader.refresh_apps(apps, model)
 
-        # Verify charm_refresh was called without channel/revision
-        self.jhelper.charm_refresh.assert_called_once_with("nova", model)
+        # Verify charm_refresh was called without channel/revision, trust=False
+        self.jhelper.charm_refresh.assert_called_once_with("nova", model, trust=False)
         assert result.result_type == ResultType.COMPLETED
 
     def test_refresh_apps_with_manifest_channel_and_revision(self):
@@ -79,12 +79,13 @@ def test_refresh_apps_with_manifest_channel_and_revision(self):
         # Execute
         result = self.upgrader.refresh_apps(apps, model)
 
-        # Verify charm_refresh was called with channel and revision
+        # Verify charm_refresh was called with channel and revision, trust=False
         self.jhelper.charm_refresh.assert_called_once_with(
             "nova",
             model,
             channel="2024.1/stable",
             revision=150,
+            trust=False,
         )
         assert result.result_type == ResultType.COMPLETED
 
@@ -108,12 +109,13 @@ def test_refresh_apps_with_manifest_channel_only(self):
         # Execute
         result = self.upgrader.refresh_apps(apps, model)
 
-        # Verify charm_refresh was called with channel but None revision
+        # Verify charm_refresh was called with channel but None revision, trust=False
         self.jhelper.charm_refresh.assert_called_once_with(
             "nova",
             model,
             channel="2024.1/stable",
             revision=None,
+            trust=False,
         )
         assert result.result_type == ResultType.COMPLETED
 
@@ -145,16 +147,72 @@ def test_refresh_apps_multiple_apps_mixed_manifest(self):
         # Verify charm_refresh was called for all apps
         assert self.jhelper.charm_refresh.call_count == 3
 
-        # Check calls
+        # Check calls — all non-octavia charms get trust=False
         calls = self.jhelper.charm_refresh.call_args_list
+        assert (
+            call("nova", model, channel="2024.1/candidate", revision=200, trust=False)
+            in calls
+        )
+        assert call("neutron", model, trust=False) in calls
+        assert call("cinder", model, trust=False) in calls
+
+        assert result.result_type == ResultType.COMPLETED
+
+    def test_refresh_apps_octavia_passes_trust_true(self):
+        """octavia-k8s is in CHARMS_REQUIRING_TRUST so trust=True must be passed."""
+        apps = {
+            "octavia": ("octavia-k8s", "2024.1/stable", 157),
+        }
+        model = "openstack"
+
+        self.jhelper.wait_until_active = Mock()
+
+        result = self.upgrader.refresh_apps(apps, model)
+
+        self.jhelper.charm_refresh.assert_called_once_with("octavia", model, trust=True)
+        assert result.result_type == ResultType.COMPLETED
+
+    def test_refresh_apps_octavia_with_manifest_passes_trust_true(self):
+        """octavia-k8s with manifest entry still gets trust=True."""
+        apps = {
+            "octavia": ("octavia-k8s", "2024.1/stable", 157),
+        }
+        model = "openstack"
+
+        manifest_charm = Mock()
+        manifest_charm.channel = "2024.1/beta"
+        manifest_charm.revision = 204
+        self.manifest.find_charm.return_value = manifest_charm
+
+        self.jhelper.wait_until_active = Mock()
+
+        result = self.upgrader.refresh_apps(apps, model)
+
+        self.jhelper.charm_refresh.assert_called_once_with(
+            "octavia",
+            model,
+            channel="2024.1/beta",
+            revision=204,
+            trust=True,
+        )
+        assert result.result_type == ResultType.COMPLETED
 
-        # Nova should be called with manifest config
-        assert call("nova", model, channel="2024.1/candidate", revision=200) in calls
+    def test_refresh_apps_mixed_trust_and_non_trust_charms(self):
+        """Octavia gets trust=True; other charms get trust=False in the same batch."""
+        apps = {
+            "nova": ("nova-k8s", "2024.1/stable", 123),
+            "octavia": ("octavia-k8s", "2024.1/stable", 157),
+        }
+        model = "openstack"
+
+        self.manifest.find_charm.return_value = None
+        self.jhelper.wait_until_active = Mock()
 
-        # Neutron and Cinder should be called without channel/revision
-        assert call("neutron", model) in calls
-        assert call("cinder", model) in calls
+        result = self.upgrader.refresh_apps(apps, model)
 
+        calls = self.jhelper.charm_refresh.call_args_list
+        assert call("nova", model, trust=False) in calls
+        assert call("octavia", model, trust=True) in calls
         assert result.result_type == ResultType.COMPLETED
 
     def test_refresh_apps_machine_model(self):
@@ -171,8 +229,10 @@ def test_refresh_apps_machine_model(self):
         # Execute
         result = self.upgrader.refresh_apps(apps, model)
 
-        # Verify charm_refresh was called
-        self.jhelper.charm_refresh.assert_called_once_with("nova-compute", model)
+        # Verify charm_refresh was called with trust=False
+        self.jhelper.charm_refresh.assert_called_once_with(
+            "nova-compute", model, trust=False
+        )
 
         # Verify wait_application_ready was called (not wait_until_active)
         self.jhelper.wait_application_ready.assert_called_once()
@@ -259,7 +319,7 @@ def test_refresh_apps_status_snapshot_exception_graceful(self):
             result = self.upgrader.refresh_apps(apps, model)
 
         # charm_refresh still called, result still COMPLETED
-        self.jhelper.charm_refresh.assert_called_once_with("nova", model)
+        self.jhelper.charm_refresh.assert_called_once_with("nova", model, trust=False)
         assert result.result_type == ResultType.COMPLETED