diff --git a/sunbeam-python/sunbeam/features/interface/v1/base.py b/sunbeam-python/sunbeam/features/interface/v1/base.py index 6e194e570..6ab568661 100644 --- a/sunbeam-python/sunbeam/features/interface/v1/base.py +++ b/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", diff --git a/sunbeam-python/sunbeam/features/loadbalancer/feature.py b/sunbeam-python/sunbeam/features/loadbalancer/feature.py index da559bf9a..a554af8ba 100644 --- a/sunbeam-python/sunbeam/features/loadbalancer/feature.py +++ b/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: diff --git a/sunbeam-python/tests/unit/sunbeam/features/test_loadbalancer.py b/sunbeam-python/tests/unit/sunbeam/features/test_loadbalancer.py index 6c9f3880f..e6917b7ad 100644 --- a/sunbeam-python/tests/unit/sunbeam/features/test_loadbalancer.py +++ b/sunbeam-python/tests/unit/sunbeam/features/test_loadbalancer.py @@ -878,21 +878,21 @@ 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 ( @@ -900,12 +900,13 @@ def test_requires_secrets_when_gate_enabled(self): "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,29 +1684,12 @@ 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 ( @@ -1713,20 +1697,16 @@ def test_requires_secrets_when_amphora_enabled_in_clusterd(self): "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 ( @@ -1734,18 +1714,14 @@ def test_requires_empty_when_amphora_disabled_in_clusterd(self): "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 ( @@ -1753,30 +1729,25 @@ def test_requires_empty_when_clusterd_key_absent(self): "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"