Implement "last feature flag wins" behavior for duplicate feature flag IDs

.github/workflows/validate.yml

@@ -32,7 +32,7 @@ jobs:
       run: |
         pip install -r tests/requirements.txt
         pytest tests --doctest-modules --cov-report=xml --cov-report=html
-    - name: Analysing the samples with pylint
+    - name: Analysing the samples/tests with pylint
       run: |
         pip install -r samples/requirements.txt
         pylint --disable=missing-function-docstring,missing-class-docstring samples tests

featuremanagement/_featuremanagerbase.py

@@ -27,6 +27,7 @@
 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.
+    If multiple feature flags have the same id, the last one wins.
 
     :param Mapping configuration: Configuration object.
     :param str feature_flag_name: Name of the feature flag.
@@ -40,32 +41,32 @@ def _get_feature_flag(configuration: Mapping[str, Any], feature_flag_name: str)
     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)
+    index = len(feature_flags) - 1
+
+    while index >= 0:
+        if feature_flags[index].get("id") == feature_flag_name:
+            return FeatureFlag.convert_from_json(feature_flags[index])
+        index -= 1
 
     return None
 
 
 def _list_feature_flag_names(configuration: Mapping[str, Any]) -> List[str]:
     """
-    List of all feature flag names.
+    List of feature flag names, with duplicates removed.
 
     :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
+    flag_ids = [feature_flag.get("id") for feature_flag in feature_flags if feature_flag.get("id")]
+    return list(dict.fromkeys(flag_ids))
 
 
 class FeatureManagerBase(ABC):

tests/test_feature_manager.py

@@ -150,6 +150,132 @@ def fake_telemetry_callback(self, evaluation_event):
         assert evaluation_event
         self.called_telemetry = True
 
+    # method: duplicate_feature_flag_handling
+    def test_duplicate_feature_flags_last_wins(self):
+        """Test that when multiple feature flags have the same ID, the last one wins."""
+        feature_flags = {
+            "feature_management": {
+                "feature_flags": [
+                    {
+                        "id": "DuplicateFlag",
+                        "description": "First",
+                        "enabled": "true",
+                        "conditions": {"client_filters": []},
+                    },
+                    {
+                        "id": "DuplicateFlag",
+                        "description": "Second",
+                        "enabled": "false",
+                        "conditions": {"client_filters": []},
+                    },
+                    {
+                        "id": "DuplicateFlag",
+                        "description": "Third",
+                        "enabled": "true",
+                        "conditions": {"client_filters": []},
+                    },
+                ]
+            }
+        }
+        feature_manager = FeatureManager(feature_flags)
+
+        # The last flag should win (enabled: true)
+        assert feature_manager.is_enabled("DuplicateFlag") is True
+
+        # Should only list unique names
+        flag_names = feature_manager.list_feature_flag_names()
+        assert "DuplicateFlag" in flag_names
+        # Count how many times DuplicateFlag appears in the list
+        duplicate_count = flag_names.count("DuplicateFlag")
+        assert duplicate_count == 1, f"Expected DuplicateFlag to appear once, but appeared {duplicate_count} times"
+
+    def test_duplicate_feature_flags_last_wins_disabled(self):
+        """Test that when multiple feature flags have the same ID, the last one wins even if disabled."""
+        feature_flags = {
+            "feature_management": {
+                "feature_flags": [
+                    {
+                        "id": "DuplicateFlag",
+                        "description": "First",
+                        "enabled": "true",
+                        "conditions": {"client_filters": []},
+                    },
+                    {
+                        "id": "DuplicateFlag",
+                        "description": "Second",
+                        "enabled": "true",
+                        "conditions": {"client_filters": []},
+                    },
+                    {
+                        "id": "DuplicateFlag",
+                        "description": "Third",
+                        "enabled": "false",
+                        "conditions": {"client_filters": []},
+                    },
+                ]
+            }
+        }
+        feature_manager = FeatureManager(feature_flags)
+
+        # The last flag should win (enabled: false)
+        assert feature_manager.is_enabled("DuplicateFlag") is False
+
+    def test_duplicate_feature_flags_mixed_with_unique(self):
+        """Test behavior with a mix of duplicate and unique feature flags."""
+        feature_flags = {
+            "feature_management": {
+                "feature_flags": [
+                    {
+                        "id": "UniqueFlag1",
+                        "description": "First unique",
+                        "enabled": "true",
+                        "conditions": {"client_filters": []},
+                    },
+                    {
+                        "id": "DuplicateFlag",
+                        "description": "First duplicate",
+                        "enabled": "false",
+                        "conditions": {"client_filters": []},
+                    },
+                    {
+                        "id": "UniqueFlag2",
+                        "description": "Second unique",
+                        "enabled": "false",
+                        "conditions": {"client_filters": []},
+                    },
+                    {
+                        "id": "DuplicateFlag",
+                        "description": "Second duplicate",
+                        "enabled": "true",
+                        "conditions": {"client_filters": []},
+                    },
+                    {
+                        "id": "UniqueFlag3",
+                        "description": "Third unique",
+                        "enabled": "true",
+                        "conditions": {"client_filters": []},
+                    },
+                ]
+            }
+        }
+        feature_manager = FeatureManager(feature_flags)
+
+        # Test unique flags work as expected
+        assert feature_manager.is_enabled("UniqueFlag1") is True
+        assert feature_manager.is_enabled("UniqueFlag2") is False
+        assert feature_manager.is_enabled("UniqueFlag3") is True
+
+        # Test duplicate flag - last should win (enabled: true)
+        assert feature_manager.is_enabled("DuplicateFlag") is True
+
+        # Test list includes all unique names
+        flag_names = feature_manager.list_feature_flag_names()
+        expected_names = ["UniqueFlag1", "DuplicateFlag", "UniqueFlag2", "UniqueFlag3"]
+        assert set(flag_names) == set(expected_names)
+        # Ensure each name appears only once
+        for name in expected_names:
+            assert flag_names.count(name) == 1
+
 
 class AlwaysOn(FeatureFilter):
     def evaluate(self, context, **kwargs):

tests/test_feature_manager_async.py

@@ -179,6 +179,78 @@ async def fake_telemetry_callback_async(self, evaluation_event):
         assert evaluation_event
         self.called_telemetry = True
 
+    # method: duplicate_feature_flag_handling
+    @pytest.mark.asyncio
+    async def test_duplicate_feature_flags_last_wins_async(self):
+        """Test that when multiple feature flags have the same ID, the last one wins."""
+        feature_flags = {
+            "feature_management": {
+                "feature_flags": [
+                    {
+                        "id": "DuplicateFlag",
+                        "description": "First",
+                        "enabled": "true",
+                        "conditions": {"client_filters": []},
+                    },
+                    {
+                        "id": "DuplicateFlag",
+                        "description": "Second",
+                        "enabled": "false",
+                        "conditions": {"client_filters": []},
+                    },
+                    {
+                        "id": "DuplicateFlag",
+                        "description": "Third",
+                        "enabled": "true",
+                        "conditions": {"client_filters": []},
+                    },
+                ]
+            }
+        }
+        feature_manager = FeatureManager(feature_flags)
+
+        # The last flag should win (enabled: true)
+        assert await feature_manager.is_enabled("DuplicateFlag") is True
+
+        # Should only list unique names
+        flag_names = feature_manager.list_feature_flag_names()
+        assert "DuplicateFlag" in flag_names
+        # Count how many times DuplicateFlag appears in the list
+        duplicate_count = flag_names.count("DuplicateFlag")
+        assert duplicate_count == 1, f"Expected DuplicateFlag to appear once, but appeared {duplicate_count} times"
+
+    @pytest.mark.asyncio
+    async def test_duplicate_feature_flags_last_wins_disabled_async(self):
+        """Test that when multiple feature flags have the same ID, the last one wins even if disabled."""
+        feature_flags = {
+            "feature_management": {
+                "feature_flags": [
+                    {
+                        "id": "DuplicateFlag",
+                        "description": "First",
+                        "enabled": "true",
+                        "conditions": {"client_filters": []},
+                    },
+                    {
+                        "id": "DuplicateFlag",
+                        "description": "Second",
+                        "enabled": "true",
+                        "conditions": {"client_filters": []},
+                    },
+                    {
+                        "id": "DuplicateFlag",
+                        "description": "Third",
+                        "enabled": "false",
+                        "conditions": {"client_filters": []},
+                    },
+                ]
+            }
+        }
+        feature_manager = FeatureManager(feature_flags)
+
+        # The last flag should win (enabled: false)
+        assert await feature_manager.is_enabled("DuplicateFlag") is False
+
 
 class AlwaysOn(FeatureFilter):
     async def evaluate(self, context, **kwargs):