Fixing Feature Flag Snapshot

sdk/appconfiguration/azure-appconfiguration-provider/CHANGELOG.md

@@ -8,6 +8,9 @@
 
 ### Bugs Fixed
 
+- Fixes a bug where `feature_flag_selects` could be passed in as `None` which resulted in an exception on load, doing this now results in loading the default feature flags.
+- Fixes a bug where `feature_flag_selects` couldn't load snapshots.
+
 ### Other Changes
 
 ## 2.3.1 (2025-11-13)

sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_azureappconfigurationproviderbase.py

@@ -231,6 +231,8 @@ def __init__(self, **kwargs: Any) -> None:
         self._refresh_timer: _RefreshTimer = _RefreshTimer(**kwargs)
         self._feature_flag_enabled = kwargs.pop("feature_flag_enabled", False)
         self._feature_flag_selectors = kwargs.pop("feature_flag_selectors", [SettingSelector(key_filter="*")])
+        if self._feature_flag_selectors is None:
+            self._feature_flag_selectors = [SettingSelector(key_filter="*")]
         self._watched_feature_flags: Dict[Tuple[str, str], Optional[str]] = {}
         self._feature_flag_refresh_timer: _RefreshTimer = _RefreshTimer(**kwargs)
         self._feature_flag_refresh_enabled = kwargs.pop("feature_flag_refresh_enabled", False)

sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_client_manager.py

@@ -137,6 +137,7 @@ def _check_configuration_setting(
     def load_configuration_settings(self, selects: List[SettingSelector], **kwargs) -> List[ConfigurationSetting]:
         configuration_settings = []
         for select in selects:
+            configurations = []
             if select.snapshot_name is not None:
                 # When loading from a snapshot, ignore key_filter, label_filter, and tag_filters
                 snapshot = self._client.get_snapshot(select.snapshot_name)
@@ -166,18 +167,26 @@ def load_feature_flags(
         # Needs to be removed unknown keyword argument for list_configuration_settings
         kwargs.pop("sentinel_keys", None)
         for select in feature_flag_selectors:
-            # Handle None key_filter by converting to empty string
-            key_filter = select.key_filter if select.key_filter is not None else ""
-            feature_flags = self._client.list_configuration_settings(
-                key_filter=FEATURE_FLAG_PREFIX + key_filter,
-                label_filter=select.label_filter,
-                tags_filter=select.tag_filters,
-                **kwargs,
-            )
+            feature_flags = []
+            if select.snapshot_name is not None:
+                # When loading from a snapshot, ignore key_filter, label_filter, and tag_filters
+                snapshot = self._client.get_snapshot(select.snapshot_name)
+                if snapshot.composition_type != SnapshotComposition.KEY:
+                    raise ValueError(f"Snapshot '{select.snapshot_name}' is not a key snapshot.")
+                feature_flags = self._client.list_configuration_settings(snapshot_name=select.snapshot_name, **kwargs)
+            else:
+                # Handle None key_filter by converting to empty string
+                key_filter = select.key_filter if select.key_filter is not None else ""
+                feature_flags = self._client.list_configuration_settings(
+                    key_filter=FEATURE_FLAG_PREFIX + key_filter,
+                    label_filter=select.label_filter,
+                    tags_filter=select.tag_filters,
+                    **kwargs,
+                )
             for feature_flag in feature_flags:
                 if not isinstance(feature_flag, FeatureFlagConfigurationSetting):
                     # If the feature flag is not a FeatureFlagConfigurationSetting, it means it was selected by
-                    # mistake, so we should ignore it.
+                    # mistake, so we should ignore it, or it was loaded via snapshot.
                     continue
                 loaded_feature_flags.append(feature_flag)
 

sdk/appconfiguration/azure-appconfiguration-provider/samples/aad_sample.py

@@ -26,7 +26,15 @@
 
 # Connection to Azure App Configuration using SettingSelector
 selects = [SettingSelector(key_filter="message*")]
-config = load(endpoint=endpoint, credential=credential, selects=selects, **kwargs)
+config = load(
+    endpoint=endpoint,
+    credential=credential,
+    selects=selects,
+    feature_flag_enabled=True,
+    feature_flag_selectors=None,
+    **kwargs
+)
 
 print("message found: " + str("message" in config))
 print("test.message found: " + str("test.message" in config))
+print("feature_flag_enabled found: " + str(config["feature_management"]))

sdk/appconfiguration/azure-appconfiguration-provider/tests/aio/test_async_snapshots.py

@@ -7,20 +7,22 @@
 import pytest
 import time
 from azure.appconfiguration.provider._models import SettingSelector
-from azure.appconfiguration.provider._constants import NULL_CHAR
+from azure.appconfiguration.provider._constants import NULL_CHAR, FEATURE_MANAGEMENT_KEY, FEATURE_FLAG_KEY
 from azure.appconfiguration.provider.aio import load
 from azure.appconfiguration.provider import WatchKey
 from azure.appconfiguration import (
     ConfigurationSetting,
-    ConfigurationSettingsFilter,
-    SnapshotComposition,
-    SnapshotStatus,
+    FeatureFlagConfigurationSetting,
 )
 from azure.core.exceptions import ResourceNotFoundError
-from devtools_testutils import is_live
 from devtools_testutils.aio import recorded_by_proxy_async
 from async_preparers import app_config_decorator_async
-from asynctestcase import AppConfigTestCase
+from asynctestcase import (
+    AppConfigTestCase,
+    cleanup_test_resources_async,
+    set_test_settings_async,
+    create_snapshot_async,
+)
 
 
 class TestSnapshotSupport:
@@ -189,9 +191,23 @@ async def test_create_snapshot_and_load_provider(self, appconfiguration_connecti
             ConfigurationSetting(key="refresh_test_key", value="original_refresh_value", label=NULL_CHAR),
         ]
 
-        # Set the configuration settings
-        for setting in test_settings:
-            await sdk_client.set_configuration_setting(setting)
+        # Create feature flag settings for the snapshot
+        # Note: Feature flags in snapshots are NOT loaded as feature flags by the provider
+        test_feature_flags = [
+            FeatureFlagConfigurationSetting(
+                feature_id="SnapshotFeature",
+                enabled=True,
+                label=NULL_CHAR,
+            ),
+            FeatureFlagConfigurationSetting(
+                feature_id="SnapshotFeatureDisabled",
+                enabled=False,
+                label=NULL_CHAR,
+            ),
+        ]
+
+        # Set the configuration settings and feature flags
+        await set_test_settings_async(sdk_client, test_settings + test_feature_flags)
 
         variables = kwargs.pop("variables", {})
         dynamic_snapshot_name_postfix = variables.setdefault("dynamic_snapshot_name_postfix", str(int(time.time())))
@@ -200,28 +216,18 @@ async def test_create_snapshot_and_load_provider(self, appconfiguration_connecti
         snapshot_name = f"test-snapshot-{dynamic_snapshot_name_postfix}"
 
         try:
-            # Create the snapshot
-            snapshot_poller = await sdk_client.begin_create_snapshot(
-                name=snapshot_name,
-                filters=[ConfigurationSettingsFilter(key="snapshot_test_*")],  # Include all our test keys
-                composition_type=SnapshotComposition.KEY,
-                retention_period=3600,  # Min valid value is 1 hour
+            # Create the snapshot including both config settings and feature flags
+            await create_snapshot_async(
+                sdk_client,
+                snapshot_name,
+                key_filters=["snapshot_test_*", ".appconfig.featureflag/SnapshotFeature*"],
             )
-            snapshot = await snapshot_poller.result()
-
-            # Verify snapshot was created successfully
-            if is_live():
-                assert snapshot.name == snapshot_name
-            else:
-                assert snapshot.name == "Sanitized"
-            assert snapshot.status == SnapshotStatus.READY
-            assert snapshot.composition_type == SnapshotComposition.KEY
 
-            # Load provider using the snapshot with refresh enabled
+            # Load provider using the snapshot with refresh enabled and feature flags enabled
             async with await self.create_client(
                 connection_string=appconfiguration_connection_string,
                 selects=[
-                    SettingSelector(snapshot_name=snapshot_name),  # Snapshot data
+                    SettingSelector(snapshot_name=snapshot_name),  # Snapshot data (includes feature flags)
                     SettingSelector(key_filter="refresh_test_key"),  # Non-snapshot key for refresh testing
                 ],
                 refresh_on=[WatchKey("refresh_test_key")],  # Watch non-snapshot key for refresh
@@ -238,6 +244,10 @@ async def test_create_snapshot_and_load_provider(self, appconfiguration_connecti
                 snapshot_keys = [key for key in provider.keys() if key.startswith("snapshot_test_")]
                 assert len(snapshot_keys) == 3
 
+                # Verify feature flags from snapshots are NOT loaded as feature flags
+                # (snapshots don't support feature flag loading, only regular selects do)
+                assert FEATURE_MANAGEMENT_KEY not in provider, "Feature flags should not be loaded from snapshots"
+
                 # Test snapshot immutability: modify the original settings
                 modified_settings = [
                     ConfigurationSetting(
@@ -260,12 +270,8 @@ async def test_create_snapshot_and_load_provider(self, appconfiguration_connecti
                 ]
 
                 # Update the original settings with new values
-                for setting in modified_settings:
-                    await sdk_client.set_configuration_setting(setting)
-
-                # Add a completely new key after initial load
                 new_key = ConfigurationSetting(key="new_key_added_after_load", value="new_value", label=NULL_CHAR)
-                await sdk_client.set_configuration_setting(new_key)
+                await set_test_settings_async(sdk_client, modified_settings + [new_key])
 
                 # Wait for refresh interval to pass
                 time.sleep(1)
@@ -296,23 +302,148 @@ async def test_create_snapshot_and_load_provider(self, appconfiguration_connecti
                 assert provider_current["snapshot_test_json"]["nested"] == "MODIFIED_VALUE"  # Modified value
 
         finally:
-            # Clean up: delete the snapshot and test settings
-            try:
-                # Archive the snapshot (delete is not supported, but archive effectively removes it)
-                await sdk_client.archive_snapshot(snapshot_name)
-            except Exception:
-                pass
-
-            # Clean up test settings
-            for setting in test_settings:
-                try:
-                    await sdk_client.delete_configuration_setting(key=setting.key, label=setting.label)
-                except Exception:
-                    pass
-
-            # Clean up additional test keys
-            try:
-                await sdk_client.delete_configuration_setting(key="new_key_added_after_load", label=NULL_CHAR)
-            except Exception:
-                pass
+            # Clean up test resources
+            cleanup_settings = (
+                test_settings
+                + test_feature_flags
+                + [ConfigurationSetting(key="new_key_added_after_load", value="", label=NULL_CHAR)]
+            )
+            await cleanup_test_resources_async(
+                sdk_client,
+                settings=cleanup_settings,
+                snapshot_names=[snapshot_name],
+            )
+        return variables
+
+    @pytest.mark.live_test_only  # Needed to fix an azure core dependency compatibility issue
+    @app_config_decorator_async
+    @recorded_by_proxy_async
+    async def test_create_snapshot_and_load_provider_with_feature_flags(
+        self, appconfiguration_connection_string, **kwargs
+    ):
+        """Test creating a snapshot and loading provider with feature flags from non-snapshot selectors."""
+        # Create SDK client for setup
+        sdk_client = self.create_sdk_client(appconfiguration_connection_string)
+
+        # Create unique test configuration settings for the snapshot
+        test_settings = [
+            ConfigurationSetting(key="ff_snapshot_test_key1", value="ff_snapshot_test_value1", label=NULL_CHAR),
+            ConfigurationSetting(key="ff_snapshot_test_key2", value="ff_snapshot_test_value2", label=NULL_CHAR),
+        ]
+
+        # Create feature flag settings - some for snapshot, some for regular loading
+        # Note: Feature flags in snapshots are NOT loaded as feature flags by the provider
+        snapshot_feature_flags = [
+            FeatureFlagConfigurationSetting(
+                feature_id="SnapshotOnlyFeature",
+                enabled=True,
+                label=NULL_CHAR,
+            ),
+        ]
+
+        # Feature flags loaded via regular selectors (not from snapshot)
+        regular_feature_flags = [
+            FeatureFlagConfigurationSetting(
+                feature_id="RegularFeature",
+                enabled=True,
+                label=NULL_CHAR,
+            ),
+            FeatureFlagConfigurationSetting(
+                feature_id="RegularFeatureDisabled",
+                enabled=False,
+                label=NULL_CHAR,
+            ),
+        ]
+
+        # Set the configuration settings and feature flags
+        await set_test_settings_async(sdk_client, test_settings + snapshot_feature_flags + regular_feature_flags)
+
+        variables = kwargs.pop("variables", {})
+        dynamic_snapshot_name_postfix = variables.setdefault("dynamic_ff_snapshot_name_postfix", str(int(time.time())))
+
+        # Create a unique snapshot name with timestamp to avoid conflicts
+        snapshot_name = f"test-ff-snapshot-{dynamic_snapshot_name_postfix}"
+
+        try:
+            # Create the snapshot including config settings and snapshot-only feature flags
+            await create_snapshot_async(
+                sdk_client,
+                snapshot_name,
+                key_filters=["ff_snapshot_test_*", ".appconfig.featureflag/SnapshotOnlyFeature"],
+            )
+
+            # Load provider using snapshot for config settings and regular selectors for feature flags
+            async with await self.create_client(
+                connection_string=appconfiguration_connection_string,
+                selects=[
+                    SettingSelector(snapshot_name=snapshot_name),  # Snapshot data (config + snapshot FF)
+                ],
+                feature_flag_enabled=True,  # Enable feature flags
+                feature_flag_selectors=[
+                    SettingSelector(key_filter="RegularFeature*"),  # Load only RegularFeature* flags
+                ],
+            ) as provider:
+
+                # Verify snapshot configuration settings are loaded
+                assert provider["ff_snapshot_test_key1"] == "ff_snapshot_test_value1"
+                assert provider["ff_snapshot_test_key2"] == "ff_snapshot_test_value2"
+
+                # Verify feature flags loaded via regular selectors ARE loaded
+                feature_flags = provider.get(FEATURE_MANAGEMENT_KEY, {}).get(FEATURE_FLAG_KEY, [])
+                feature_flag_ids = {ff["id"]: ff["enabled"] for ff in feature_flags}
+
+                # Regular feature flags should be loaded
+                assert "RegularFeature" in feature_flag_ids, "RegularFeature should be loaded via regular selector"
+                assert feature_flag_ids["RegularFeature"] is True
+                assert "RegularFeatureDisabled" in feature_flag_ids, "RegularFeatureDisabled should be loaded"
+                assert feature_flag_ids["RegularFeatureDisabled"] is False
+
+                # Snapshot-only feature flag should NOT be loaded as a feature flag
+                assert (
+                    "SnapshotOnlyFeature" not in feature_flag_ids
+                ), "SnapshotOnlyFeature should not be loaded as FF from snapshot"
+
+                # Verify exactly 2 feature flags are loaded (the regular ones)
+                assert len(feature_flags) == 2, f"Expected 2 feature flags, got {len(feature_flags)}"
+
+            # Modify the regular feature flags
+            modified_feature_flags = [
+                FeatureFlagConfigurationSetting(
+                    feature_id="RegularFeature",
+                    enabled=False,  # Changed from True to False
+                    label=NULL_CHAR,
+                ),
+                FeatureFlagConfigurationSetting(
+                    feature_id="RegularFeatureDisabled",
+                    enabled=True,  # Changed from False to True
+                    label=NULL_CHAR,
+                ),
+            ]
+
+            await set_test_settings_async(sdk_client, modified_feature_flags)
+
+            # Load a fresh provider without snapshot to verify current feature flag values
+            async with await self.create_client(
+                connection_string=appconfiguration_connection_string,
+                selects=[SettingSelector(key_filter="ff_snapshot_test_*")],
+                feature_flag_enabled=True,
+                feature_flag_selectors=[
+                    SettingSelector(key_filter="RegularFeature*"),
+                ],
+            ) as provider_current:
+
+                # Current feature flag values should be the modified ones
+                current_feature_flags = provider_current.get(FEATURE_MANAGEMENT_KEY, {}).get(FEATURE_FLAG_KEY, [])
+                current_ff_ids = {ff["id"]: ff["enabled"] for ff in current_feature_flags}
+                assert current_ff_ids.get("RegularFeature") is False  # Modified value
+                assert current_ff_ids.get("RegularFeatureDisabled") is True  # Modified value
+
+        finally:
+            # Clean up test resources
+            cleanup_settings = test_settings + snapshot_feature_flags + regular_feature_flags
+            await cleanup_test_resources_async(
+                sdk_client,
+                settings=cleanup_settings,
+                snapshot_names=[snapshot_name],
+            )
         return variables