Last One Wins with Duplicate Feature Flag Names

.gitignore

@@ -413,3 +413,4 @@ docs/doctrees
 docs/html
 package-lock.json
 package.json
+.env

featuremanagement/_featuremanagerbase.py

@@ -8,7 +8,7 @@
 import hashlib
 import logging
 from abc import ABC
-from typing import List, Optional, Dict, Tuple, Any, Mapping, Callable
+from typing import List, Optional, Dict, Tuple, Any, Mapping, Callable, cast
 from ._models import FeatureFlag, Variant, VariantAssignmentReason, TargetingContext, EvaluationEvent, VariantReference
 
 FEATURE_MANAGEMENT_KEY = "feature_management"
@@ -25,50 +25,6 @@
 logger = logging.getLogger(__name__)
 
 
-def _get_feature_flag(configuration: Mapping[str, Any], feature_flag_name: str) -> Optional[FeatureFlag]:
-    """
-    Gets the FeatureFlag json from the configuration, if it exists it gets converted to a FeatureFlag object.
-
-    :param Mapping configuration: Configuration object.
-    :param str feature_flag_name: Name of the feature flag.
-    :return: FeatureFlag
-    :rtype: FeatureFlag
-    """
-    feature_management = configuration.get(FEATURE_MANAGEMENT_KEY)
-    if not feature_management or not isinstance(feature_management, Mapping):
-        return None
-    feature_flags = feature_management.get(FEATURE_FLAG_KEY)
-    if not feature_flags or not isinstance(feature_flags, list):
-        return None
-
-    for feature_flag in feature_flags:
-        if feature_flag.get("id") == feature_flag_name:
-            return FeatureFlag.convert_from_json(feature_flag)
-
-    return None
-
-
-def _list_feature_flag_names(configuration: Mapping[str, Any]) -> List[str]:
-    """
-    List of all feature flag names.
-
-    :param Mapping configuration: Configuration object.
-    :return: List of feature flag names.
-    """
-    feature_flag_names = []
-    feature_management = configuration.get(FEATURE_MANAGEMENT_KEY)
-    if not feature_management or not isinstance(feature_management, Mapping):
-        return []
-    feature_flags = feature_management.get(FEATURE_FLAG_KEY)
-    if not feature_flags or not isinstance(feature_flags, list):
-        return []
-
-    for feature_flag in feature_flags:
-        feature_flag_names.append(feature_flag.get("id"))
-
-    return feature_flag_names
-
-
 class FeatureManagerBase(ABC):
     """
     Base class for Feature Manager. This class is responsible for all shared logic between the sync and async.
@@ -273,7 +229,7 @@ def _check_feature_base(self, feature_flag_id: str) -> Tuple[EvaluationEvent, bo
             self._copy = self._configuration.get(FEATURE_MANAGEMENT_KEY)
 
         if not self._cache.get(feature_flag_id):
-            feature_flag = _get_feature_flag(self._configuration, feature_flag_id)
+            feature_flag = self._get_feature_flag(feature_flag_id)
             self._cache[feature_flag_id] = feature_flag
         else:
             feature_flag = self._cache.get(feature_flag_id)
@@ -300,4 +256,41 @@ def list_feature_flag_names(self) -> List[str]:
         """
         List of all feature flag names.
         """
-        return _list_feature_flag_names(self._configuration)
+        feature_flag_names: Dict[str, None] = {}
+        for feature_flag in self._get_feature_flags():
+            feature_flag_name = feature_flag.get("id")
+            # Only include entries with a valid string id; duplicates are listed once.
+            if isinstance(feature_flag_name, str):
+                feature_flag_names[feature_flag_name] = None
+
+        return list(feature_flag_names)
+
+    def _get_feature_flag(self, feature_flag_name: str) -> Optional[FeatureFlag]:
+        """
+        Gets the FeatureFlag json from the configuration, if it exists it gets converted to a FeatureFlag object.
+
+        :param str feature_flag_name: Name of the feature flag.
+        :return: FeatureFlag
+        :rtype: FeatureFlag
+        """
+        for feature_flag in reversed(self._get_feature_flags()):
+            # If multiple feature flags share the same id, the last one defined wins.
+            if feature_flag.get("id") == feature_flag_name:
+                return FeatureFlag.convert_from_json(feature_flag)
+
+        return None
+
+    def _get_feature_flags(self) -> List[Any]:
+        """
+        Gets the list of raw feature flag definitions from the configuration.
+
+        :return: List of feature flag definitions, or an empty list if none are configured.
+        :rtype: list
+        """
+        feature_management = self._configuration.get(FEATURE_MANAGEMENT_KEY)
+        if not feature_management or not isinstance(feature_management, Mapping):
+            return []
+        feature_flags = feature_management.get(FEATURE_FLAG_KEY)
+        if not feature_flags or not isinstance(feature_flags, list):
+            return []
+        return cast(List[Any], feature_flags)

tests/test_feature_manager.py

@@ -112,6 +112,36 @@ def test_list_feature_flags(self):
         assert not feature_manager.is_enabled("Beta")
         assert len(feature_manager.list_feature_flag_names()) == 2
 
+    # method: list_feature_flags
+    def test_list_feature_flags_no_feature_flags(self):
+        # feature_management is present but has no valid feature_flags list.
+        feature_manager = FeatureManager({"feature_management": {"feature_flags": None}})
+        assert feature_manager is not None
+        assert not feature_manager.list_feature_flag_names()
+        assert not feature_manager.is_enabled("Alpha")
+
+        # feature_flags is present but is not a list.
+        feature_manager = FeatureManager({"feature_management": {"feature_flags": "not-a-list"}})
+        assert not feature_manager.list_feature_flag_names()
+        assert not feature_manager.is_enabled("Alpha")
+
+    # method: is_enabled
+    def test_duplicate_feature_flag_last_wins(self):
+        feature_flags = {
+            "feature_management": {
+                "feature_flags": [
+                    {"id": "Alpha", "description": "", "enabled": "true", "conditions": {"client_filters": []}},
+                    {"id": "Alpha", "description": "", "enabled": "false", "conditions": {"client_filters": []}},
+                ]
+            }
+        }
+        feature_manager = FeatureManager(feature_flags)
+        assert feature_manager is not None
+        # The last feature flag with the same id should win.
+        assert not feature_manager.is_enabled("Alpha")
+        # Duplicate ids should only be listed once.
+        assert feature_manager.list_feature_flag_names() == ["Alpha"]
+
     # method: is_enabled
     def test_unknown_feature_filter(self):
         feature_flags = {

tests/test_feature_manager_async.py

@@ -119,6 +119,38 @@ async def test_list_feature_flags(self):
         assert not await feature_manager.is_enabled("Beta")
         assert len(feature_manager.list_feature_flag_names()) == 2
 
+    # method: list_feature_flags
+    @pytest.mark.asyncio
+    async def test_list_feature_flags_no_feature_flags(self):
+        # feature_management is present but has no valid feature_flags list.
+        feature_manager = FeatureManager({"feature_management": {"feature_flags": None}})
+        assert feature_manager is not None
+        assert not feature_manager.list_feature_flag_names()
+        assert not await feature_manager.is_enabled("Alpha")
+
+        # feature_flags is present but is not a list.
+        feature_manager = FeatureManager({"feature_management": {"feature_flags": "not-a-list"}})
+        assert not feature_manager.list_feature_flag_names()
+        assert not await feature_manager.is_enabled("Alpha")
+
+    # method: is_enabled
+    @pytest.mark.asyncio
+    async def test_duplicate_feature_flag_last_wins(self):
+        feature_flags = {
+            "feature_management": {
+                "feature_flags": [
+                    {"id": "Alpha", "description": "", "enabled": "true", "conditions": {"client_filters": []}},
+                    {"id": "Alpha", "description": "", "enabled": "false", "conditions": {"client_filters": []}},
+                ]
+            }
+        }
+        feature_manager = FeatureManager(feature_flags)
+        assert feature_manager is not None
+        # The last feature flag with the same id should win.
+        assert not await feature_manager.is_enabled("Alpha")
+        # Duplicate ids should only be listed once.
+        assert feature_manager.list_feature_flag_names() == ["Alpha"]
+
     # method: is_enabled
     @pytest.mark.asyncio
     async def test_unknown_feature_filter(self):