fix(sunbeam-python): Use Deployment Client to Determine LoadbalancerFeature Requirements

sunbeam-python/sunbeam/features/interface/v1/base.py

@@ -595,6 +595,15 @@ def __init__(self) -> None:
         """Constructor for feature interface."""
         self.user_manifest: Path | None = None
 
+    def get_requirements(self, deployment: Deployment) -> set[FeatureRequirement]:
+        """Return feature requirements for the deployment.
+
+        Deployment is not used in the base implementation, but is provided as an
+        argument to extend functionality for subclasses that may need access to
+        the deployment model and client to determine requirements.
+        """
+        return self.requires
+
     def is_enabled(self, client: Client) -> bool:
         """Feature is enabled or disabled.
 
@@ -704,7 +713,7 @@ def check_enablement_requirements(
             feature = klass()
             if not feature.is_enabled(deployment.get_client()):
                 continue
-            for requirement in feature.requires:
+            for requirement in feature.get_requirements(deployment):
                 if requirement.name != self.name:
                     continue
                 if state == "disable":
@@ -723,7 +732,7 @@ def check_enablement_requirements(
 
     def enable_requirements(self, deployment: Deployment, show_hints: bool):
         """Iterate through requirements, enable features if possible."""
-        for requirement in self.requires:
+        for requirement in self.get_requirements(deployment):
             if not issubclass(requirement.klass, EnableDisableFeature):
                 LOG.debug(
                     "Skipping %s as it is not of type EnableDisableFeature",

sunbeam-python/sunbeam/features/loadbalancer/feature.py

@@ -15,7 +15,6 @@
 from rich.console import Console
 from rich.table import Table
 
-from sunbeam.clusterd.client import Client
 from sunbeam.clusterd.service import ConfigItemNotFoundException
 from sunbeam.commands.configure import retrieve_admin_credentials
 from sunbeam.core import questions
@@ -1503,8 +1502,7 @@ class LoadbalancerFeature(OpenStackControlPlaneFeature):
     name = "loadbalancer"
     tf_plan_location = TerraformPlanLocation.SUNBEAM_TERRAFORM_REPO
 
-    @property
-    def requires(self) -> set[FeatureRequirement]:  # type: ignore[override]
+    def get_requirements(self, deployment: Deployment) -> set[FeatureRequirement]:
         """Require Barbican (secrets) only when Amphora is actually configured.
 
         Checks both the snap feature gate (coarse guard) and the persisted
@@ -1513,7 +1511,9 @@ def requires(self) -> set[FeatureRequirement]:  # type: ignore[override]
         if not is_feature_gate_enabled("feature.loadbalancer-amphora"):
             return set()
         try:
-            saved = questions.load_answers(Client.from_socket(), AMPHORA_CONFIG_SECTION)
+            saved = questions.load_answers(
+                deployment.get_client(), AMPHORA_CONFIG_SECTION
+            )
             if not saved.get(_AMPHORA_ENABLED_KEY, False):
                 return set()
         except Exception:

sunbeam-python/tests/unit/sunbeam/features/test_loadbalancer.py

@@ -878,34 +878,35 @@ def test_juju_wait_octavia_timeout_returns_failed(self):
 
 
 class TestLoadbalancerFeatureRequires:
-    """Test the dynamic ``requires`` property on LoadbalancerFeature."""
+    """Test dynamic requirements on LoadbalancerFeature."""
 
     def _make_feature(self):
         return LoadbalancerFeature()
 
-    def test_requires_empty_when_gate_disabled(self):
+    def test_requires_empty_when_gate_disabled(self, deployment):
         """No FeatureRequirement when loadbalancer-amphora gate is off."""
         feature = self._make_feature()
         with patch(
             "sunbeam.features.loadbalancer.feature.is_feature_gate_enabled",
             return_value=False,
         ):
-            assert feature.requires == set()
+            assert feature.get_requirements(deployment) == set()
 
-    def test_requires_secrets_when_gate_enabled(self):
+    def test_requires_secrets_when_gate_enabled(self, deployment):
         """FeatureRequirement('secrets') returned when gate is on."""
         feature = self._make_feature()
         with (
             patch(
                 "sunbeam.features.loadbalancer.feature.is_feature_gate_enabled",
                 return_value=True,
             ),
-            patch(
-                "sunbeam.features.loadbalancer.feature.Client.from_socket",
-                side_effect=Exception("not a snap"),
+            patch.object(
+                deployment, "get_client", side_effect=Exception("unavailable")
             ),
         ):
-            assert feature.requires == {FeatureRequirement("secrets")}
+            assert feature.get_requirements(deployment) == {
+                FeatureRequirement("secrets")
+            }
 
 
 class TestLoadbalancerFeatureEnabledCommands:
@@ -1683,100 +1684,70 @@ def capture_run_plan(plan, *args, **kwargs):
 
 
 class TestLoadbalancerFeatureRequiresClusterd:
-    """Verify requires reads amphora_enabled from clusterd via Client.from_socket."""
+    """Verify requirements read amphora_enabled via the deployment client."""
 
     def _make_feature(self):
         return LoadbalancerFeature()
 
-    def _gate_on_socket(self, feature, load_answers_return):
-        """Helper: patch gate=True and Client.from_socket + load_answers."""
-        return (
-            patch(
-                "sunbeam.features.loadbalancer.feature.is_feature_gate_enabled",
-                return_value=True,
-            ),
-            patch(
-                "sunbeam.features.loadbalancer.feature.Client.from_socket",
-                return_value=Mock(),
-            ),
-            patch(
-                "sunbeam.features.loadbalancer.feature.questions.load_answers",
-                return_value=load_answers_return,
-            ),
-        )
-
-    def test_requires_secrets_when_amphora_enabled_in_clusterd(self):
+    def test_requires_secrets_when_amphora_enabled_in_clusterd(self, deployment):
         """Requires secrets when clusterd says amphora_enabled=True."""
         feature = self._make_feature()
         with (
             patch(
                 "sunbeam.features.loadbalancer.feature.is_feature_gate_enabled",
                 return_value=True,
             ),
-            patch(
-                "sunbeam.features.loadbalancer.feature.Client.from_socket",
-                return_value=Mock(),
-            ),
             patch(
                 "sunbeam.features.loadbalancer.feature.questions.load_answers",
                 return_value={_AMPHORA_ENABLED_KEY: True},
             ),
         ):
-            reqs = feature.requires
+            reqs = feature.get_requirements(deployment)
         assert len(reqs) == 1
         assert next(iter(reqs)).name == "secrets"
 
-    def test_requires_empty_when_amphora_disabled_in_clusterd(self):
+    def test_requires_empty_when_amphora_disabled_in_clusterd(self, deployment):
         """No requirements when clusterd says amphora_enabled=False."""
         feature = self._make_feature()
         with (
             patch(
                 "sunbeam.features.loadbalancer.feature.is_feature_gate_enabled",
                 return_value=True,
             ),
-            patch(
-                "sunbeam.features.loadbalancer.feature.Client.from_socket",
-                return_value=Mock(),
-            ),
             patch(
                 "sunbeam.features.loadbalancer.feature.questions.load_answers",
                 return_value={_AMPHORA_ENABLED_KEY: False},
             ),
         ):
-            assert feature.requires == set()
+            assert feature.get_requirements(deployment) == set()
 
-    def test_requires_empty_when_clusterd_key_absent(self):
+    def test_requires_empty_when_clusterd_key_absent(self, deployment):
         """No requirements when key is absent (e.g. after post_disable deleted it)."""
         feature = self._make_feature()
         with (
             patch(
                 "sunbeam.features.loadbalancer.feature.is_feature_gate_enabled",
                 return_value=True,
             ),
-            patch(
-                "sunbeam.features.loadbalancer.feature.Client.from_socket",
-                return_value=Mock(),
-            ),
             patch(
                 "sunbeam.features.loadbalancer.feature.questions.load_answers",
                 return_value={},
             ),
         ):
-            assert feature.requires == set()
+            assert feature.get_requirements(deployment) == set()
 
-    def test_requires_secrets_when_socket_unavailable(self):
-        """Falls back to requiring secrets when clusterd socket is unreachable."""
+    def test_requires_secrets_when_client_unavailable(self, deployment):
+        """Fall back to requiring secrets when the deployment client is unavailable."""
         feature = self._make_feature()
         with (
             patch(
                 "sunbeam.features.loadbalancer.feature.is_feature_gate_enabled",
                 return_value=True,
             ),
-            patch(
-                "sunbeam.features.loadbalancer.feature.Client.from_socket",
-                side_effect=Exception("socket not available"),
+            patch.object(
+                deployment, "get_client", side_effect=Exception("unavailable")
             ),
         ):
-            reqs = feature.requires
+            reqs = feature.get_requirements(deployment)
         assert len(reqs) == 1
         assert next(iter(reqs)).name == "secrets"