Merge "Adjust get issuer url code and unit tests"

python3-k8sapp-openstack/k8sapp_openstack/k8sapp_openstack/helm/keystone.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2019-2024 Wind River Systems, Inc.
+# Copyright (c) 2019-2025 Wind River Systems, Inc.
 #
 # SPDX-License-Identifier: Apache-2.0
 #
@@ -8,13 +8,13 @@ import os
 
 from oslo_log import log as logging
 from six.moves import configparser
-from sysinv.common import constants
 from sysinv.common import exception
 from sysinv.db import api as dbapi
 from sysinv.helm import common
 
 from k8sapp_openstack.common import constants as app_constants
 from k8sapp_openstack.helm import openstack
+from k8sapp_openstack.utils import get_dex_issuer_url
 from k8sapp_openstack.utils import is_dex_enabled
 
 LOG = logging.getLogger(__name__)
@@ -323,27 +323,28 @@ class KeystoneHelm(openstack.OpenstackBaseHelm):
         }
 
     def _get_oidc_overrides(self):
+        """
+        Generate OIDC override values for Dex integration.
+
+        This function builds the OIDC override dictionary containing the
+        `provider_remote_id`, which is derived from the system's Dex issuer URL.
+        The value is added even if the OIDC application itself is not applied,
+        since it is only used when `dex_idp.enabled` is set to True.
+
+        Returns:
+            dict: A dictionary with the Dex OIDC override in the format:
+                {
+                    'dex_idp': {
+                        'provider_remote_id': <issuer_url or empty string>
+                    }
+                }
+        """
         db = dbapi.get_instance()
         dex_enabled = is_dex_enabled()
-        # since this will only be used if dex_idp.enabled is true, it can be ammended to the
+        # Because this will only be used if dex_idp.enabled is true, it can be ammended to the
         # overrides even if oidc is not applied
         return {
             'dex_idp': {
-                'provider_remote_id': self.get_dex_issuer_url(db, dex_enabled)
+                'provider_remote_id': get_dex_issuer_url(db, dex_enabled)
             }
         }
-
-    def get_dex_issuer_url(self, db, dex_enabled):
-
-        try:
-            oidc_issuer_url = db.service_parameter_get_one(
-                service=constants.SERVICE_TYPE_KUBERNETES,
-                section=constants.SERVICE_PARAM_SECTION_KUBERNETES_APISERVER,
-                name=constants.SERVICE_PARAM_NAME_OIDC_ISSUER_URL)
-            return oidc_issuer_url.value
-        except Exception as e:
-            if dex_enabled:
-                LOG.error(f"Failed to retrieve OIDC issuer URL: {e}")
-                raise exception.NotFound("Failed to retrieve OIDC issuer URL")
-            else:
-                return ""

python3-k8sapp-openstack/k8sapp_openstack/k8sapp_openstack/tests/utils/test_utils.py

@@ -1218,27 +1218,52 @@ class UtilsTest(dbbase.ControllerHostTestCase):
         assert result == ""
 
     @mock.patch("k8sapp_openstack.utils._get_value_from_application")
-    def test_returns_true_when_enabled_true(self, mock_get_value):
+    def test_is_dex_enabled_returns_true(self, mock_get_value):
-        mock_get_value.return_value = "true"
+        mock_get_value.return_value = True
 
         result = app_utils.is_dex_enabled()
         self.assertTrue(result)
         mock_get_value.assert_called_once_with(
-            default_value="false",
+            default_value=False,
             chart_name=app_constants.HELM_CHART_KEYSTONE,
             override_name="conf.federation.dex_idp.enabled",
         )
 
     @mock.patch("k8sapp_openstack.utils._get_value_from_application")
-    def test_returns_false_when_enabled_false(self, mock_get_value):
+    def test_is_dex_enabled_returns_false(self, mock_get_value):
-        mock_get_value.return_value = "false"
+        """ test is_dex_enabled for when dex_ipd.enabled equals false
+        """
+        mock_get_value.return_value = False
 
         result = app_utils.is_dex_enabled()
         self.assertFalse(result)
 
-    @mock.patch("k8sapp_openstack.utils._get_value_from_application")
+    def test_get_dex_issuer_url_enabled_success(self):
-    def test_returns_false_when_enabled_other(self, mock_get_value):
+        """ Test get_dex_issuer_url with successfully retrieving parameter
-        mock_get_value.return_value = "anything_else"
+        """
+        db_mock = mock.Mock()
+        db_mock.service_parameter_get_one.return_value.value = "https://dex.example.com"
 
-        result = app_utils.is_dex_enabled()
+        result = app_utils.get_dex_issuer_url(db_mock, dex_enabled=True)
-        self.assertFalse(result)
+        assert result == "https://dex.example.com"
+
+    def test_get_dex_issuer_url_enabled_not_found(self):
+        """ Test get_dex_issuer_url with dex enabled but not configured
+        """
+        db_mock = mock.Mock()
+        db_mock.service_parameter_get_one.side_effect = Exception("DB error")
+
+        self.assertRaises(
+            exception.NotFound,
+            app_utils.get_dex_issuer_url,
+            db_mock,
+            dex_enabled=True)
+
+    def test_get_dex_issuer_url_disabled_not_found(self):
+        """ Test get_dex_issuer_url with dex disabled
+        """
+        db_mock = mock.Mock()
+        db_mock.service_parameter_get_one.side_effect = Exception("DB error")
+
+        result = app_utils.get_dex_issuer_url(db_mock, dex_enabled=False)
+        assert result == ""

python3-k8sapp-openstack/k8sapp_openstack/k8sapp_openstack/utils.py

@@ -1536,14 +1536,44 @@ def get_server_list() -> str:
 
 
 def is_dex_enabled() -> bool:
-    """ Retrieves if DEX integration has been enabled by user
+    """
+    Determine whether DEX integration is enabled in Keystone overrides.
 
     Returns:
-        bool: Whether user has enabled or not DEX integration.
+        bool: True if DEX integration is enabled, False otherwise.
     """
     enabled = _get_value_from_application(
-        default_value="false",
+        default_value=False,
         chart_name=app_constants.HELM_CHART_KEYSTONE,
-        override_name="conf.federation.dex_idp.enabled").lower()
+        override_name="conf.federation.dex_idp.enabled")
 
-    return enabled == 'true'
+    return enabled
+
+
+def get_dex_issuer_url(db, dex_enabled) -> str:
+    """
+    Retrieve the OIDC issuer URL from system parameters.
+
+    Args:
+        db: The system database instance.
+        dex_enabled (bool): Indicates if Dex is enabled via user overrides.
+
+    Returns:
+        str: The OIDC issuer URL if it exists. Returns an empty string if Dex is disabled
+            and the parameter is not found.
+
+    Raises:
+        NotFound: If Dex is enabled but the OIDC issuer URL cannot be retrieved.
+    """
+    try:
+        oidc_issuer_url = db.service_parameter_get_one(
+            service=constants.SERVICE_TYPE_KUBERNETES,
+            section=constants.SERVICE_PARAM_SECTION_KUBERNETES_APISERVER,
+            name=constants.SERVICE_PARAM_NAME_OIDC_ISSUER_URL)
+        return oidc_issuer_url.value
+    except Exception as e:
+        if dex_enabled:
+            LOG.error(f"Failed to retrieve OIDC issuer URL: {e}")
+            raise exception.NotFound("Failed to retrieve OIDC issuer URL")
+        else:
+            return ""