diff --git a/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_azureappconfigurationprovider.py b/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_azureappconfigurationprovider.py index b5190d9b26f1..5bccd20bec99 100644 --- a/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_azureappconfigurationprovider.py +++ b/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_azureappconfigurationprovider.py @@ -27,6 +27,7 @@ from ._constants import ( FEATURE_MANAGEMENT_KEY, FEATURE_FLAG_KEY, + SNAPSHOT_REF_CONTENT_TYPE, ) from ._azureappconfigurationproviderbase import ( AzureAppConfigurationProviderBase, @@ -299,7 +300,7 @@ def _attempt_refresh(self, client: ConfigurationClient, replica_count: int, is_f if settings_refreshed: # Configuration Settings have been refreshed - processed_settings = self._process_configurations(configuration_settings) + processed_settings = self._process_configurations(configuration_settings, client) processed_settings = self._process_feature_flags(processed_settings, processed_feature_flags, feature_flags) self._dict = processed_settings @@ -383,7 +384,7 @@ def _load_all(self, **kwargs: Any) -> None: try: configuration_settings = client.load_configuration_settings(self._selects, headers=headers, **kwargs) watched_settings = self._update_watched_settings(configuration_settings) - processed_settings = self._process_configurations(configuration_settings) + processed_settings = self._process_configurations(configuration_settings, client) if self._feature_flag_enabled: feature_flags: List[FeatureFlagConfigurationSetting] = client.load_feature_flags( @@ -422,7 +423,9 @@ def _load_all(self, **kwargs: Any) -> None: is_failover_request = True raise exception - def _process_configurations(self, configuration_settings: List[ConfigurationSetting]) -> Dict[str, Any]: + def _process_configurations( + self, configuration_settings: List[ConfigurationSetting], client: ConfigurationClient + ) -> Dict[str, Any]: # configuration_settings can contain duplicate keys, but they are in priority order, i.e. later settings take # precedence. Only process the settings with the highest priority (i.e. the last one in the list). unique_settings = self._deduplicate_settings(configuration_settings) @@ -440,6 +443,31 @@ def _process_configurations(self, configuration_settings: List[ConfigurationSett if self._feature_flag_refresh_enabled: self._watched_feature_flags[(settings.key, settings.label)] = settings.etag + elif settings.content_type and SNAPSHOT_REF_CONTENT_TYPE in settings.content_type: + # Check if this is a snapshot reference + + # Track snapshot reference usage for telemetry + self._tracing_context.uses_snapshot_reference = True + try: + # Resolve the snapshot reference to actual settings + snapshot_settings = client.resolve_snapshot_reference(settings) + + # Recursively process the resolved snapshot settings to handle feature flags, + # configuration mapping + snapshot_configuration_list = list(snapshot_settings.values()) + resolved_settings = self._process_configurations(snapshot_configuration_list, client) + + # Merge the resolved settings into our configuration + configuration_settings_processed.update(resolved_settings) + except (ValueError, AzureError) as e: + logger.warning( + "Failed to resolve snapshot reference for key '%s' (label: '%s'): %s", + settings.key, + settings.label, + str(e), + ) + # Continue processing other settings even if snapshot resolution fails + else: key = self._process_key_name(settings) value = self._process_key_value(settings) diff --git a/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_azureappconfigurationproviderbase.py b/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_azureappconfigurationproviderbase.py index d5ff6b58d5cc..36f0943581b6 100644 --- a/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_azureappconfigurationproviderbase.py +++ b/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_azureappconfigurationproviderbase.py @@ -39,6 +39,7 @@ ALLOCATION_ID_KEY, APP_CONFIG_AI_MIME_PROFILE, APP_CONFIG_AICC_MIME_PROFILE, + SNAPSHOT_REF_CONTENT_TYPE, FEATURE_MANAGEMENT_KEY, FEATURE_FLAG_KEY, ) @@ -471,6 +472,8 @@ def _process_key_value_base(self, config: ConfigurationSetting) -> Union[str, Di self._tracing_context.uses_ai_configuration = True if APP_CONFIG_AICC_MIME_PROFILE in config.content_type: self._tracing_context.uses_aicc_configuration = True + if SNAPSHOT_REF_CONTENT_TYPE in config.content_type: + self._tracing_context.uses_snapshot_reference = True return json.loads(config.value) except json.JSONDecodeError: try: diff --git a/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_client_manager.py b/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_client_manager.py index 80b1f47051c9..128dff9bdf53 100644 --- a/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_client_manager.py +++ b/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_client_manager.py @@ -11,7 +11,7 @@ from typing_extensions import Self from azure.core import MatchConditions from azure.core.tracing.decorator import distributed_trace -from azure.core.exceptions import HttpResponseError +from azure.core.exceptions import HttpResponseError, AzureError from azure.core.credentials import TokenCredential from azure.appconfiguration import ( # type:ignore # pylint:disable=no-name-in-module ConfigurationSetting, @@ -28,6 +28,8 @@ from ._models import SettingSelector from ._constants import FEATURE_FLAG_PREFIX from ._discovery import find_auto_failover_endpoints +from ._snapshot_reference_parser import SnapshotReferenceParser +from ._constants import SNAPSHOT_REF_CONTENT_TYPE @dataclass @@ -261,6 +263,43 @@ def __enter__(self): def __exit__(self, *args): self._client.__exit__(*args) + def resolve_snapshot_reference(self, setting: ConfigurationSetting, **kwargs) -> Dict[str, ConfigurationSetting]: + """ + Resolve a snapshot reference configuration setting to the actual snapshot data. + + :param ConfigurationSetting setting: The snapshot reference configuration setting + :return: A dictionary of resolved configuration settings from the snapshot + :rtype: Dict[str, ConfigurationSetting] + :raises ValueError: When the setting is not a valid snapshot reference + """ + if not setting.content_type or not SNAPSHOT_REF_CONTENT_TYPE in setting.content_type: + raise ValueError("Setting is not a snapshot reference") + + try: + # Parse the snapshot reference + snapshot_name = SnapshotReferenceParser.parse(setting) + + # Create a selector for the snapshot + snapshot_selector = SettingSelector(snapshot_name=snapshot_name) + + # Use existing load_configuration_settings to load from snapshot + configurations = self.load_configuration_settings([snapshot_selector], **kwargs) + + # Build a dictionary keyed by configuration key + snapshot_settings = {} + for config in configurations: + # Last wins for duplicate keys during iteration + snapshot_settings[config.key] = config + + return snapshot_settings + + except AzureError as e: + # Wrap Azure errors with more context + raise ValueError( + f"Failed to resolve snapshot reference for key '{setting.key}' " + f"(label: '{setting.label}'). Azure service error occurred." + ) from e + class ConfigurationClientManager(ConfigurationClientManagerBase): # pylint:disable=too-many-instance-attributes def __init__( diff --git a/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_constants.py b/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_constants.py index c683b43ecfba..58981c2a3fcc 100644 --- a/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_constants.py +++ b/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_constants.py @@ -27,3 +27,10 @@ # Mime profiles APP_CONFIG_AI_MIME_PROFILE = "https://azconfig.io/mime-profiles/ai/" APP_CONFIG_AICC_MIME_PROFILE = "https://azconfig.io/mime-profiles/ai/chat-completion" + +# Snapshot reference tracing key +SNAPSHOT_REFERENCE_TAG = "UsesSnapshotReference" + +# Snapshot reference constants +SNAPSHOT_REF_CONTENT_TYPE = 'application/json; profile="https://azconfig.io/mime-profiles/snapshot-ref"; charset=utf-8' +SNAPSHOT_NAME_FIELD = "snapshot_name" diff --git a/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_request_tracing_context.py b/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_request_tracing_context.py index 8b5fa3d94bc9..b3994e6cc95c 100644 --- a/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_request_tracing_context.py +++ b/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_request_tracing_context.py @@ -15,6 +15,7 @@ KubernetesEnvironmentVariable, APP_CONFIG_AI_MIME_PROFILE, APP_CONFIG_AICC_MIME_PROFILE, + SNAPSHOT_REFERENCE_TAG, ) # Feature flag filter names @@ -86,6 +87,7 @@ def __init__(self, load_balancing_enabled: bool = False) -> None: self.uses_load_balancing = load_balancing_enabled self.uses_ai_configuration = False self.uses_aicc_configuration = False # AI Chat Completion + self.uses_snapshot_reference = False self.uses_telemetry = False self.uses_seed = False self.max_variants: Optional[int] = None @@ -218,6 +220,9 @@ def update_correlation_context_header( if self.is_failover_request: tags.append(FAILOVER_TAG) + if self.uses_snapshot_reference: + tags.append(SNAPSHOT_REFERENCE_TAG) + # Build the correlation context string context_parts: List[str] = [] diff --git a/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_snapshot_reference_parser.py b/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_snapshot_reference_parser.py new file mode 100644 index 000000000000..fe037770c022 --- /dev/null +++ b/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/_snapshot_reference_parser.py @@ -0,0 +1,77 @@ +# ------------------------------------------------------------------------ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See License.txt in the project root for +# license information. +# ------------------------------------------------------------------------- + +import json +from azure.appconfiguration import ConfigurationSetting # type: ignore +from ._constants import SNAPSHOT_NAME_FIELD + + +class SnapshotReferenceParser: + """ + Parser for snapshot reference configuration settings. + """ + + @staticmethod + def parse(setting: ConfigurationSetting) -> str: + """ + Parse a snapshot reference from a configuration setting containing snapshot reference JSON. + + :param ConfigurationSetting setting: The configuration setting containing the snapshot reference JSON + :return: The snapshot name extracted from the reference + :rtype: str + :raises ValueError: When the setting is None + :raises ValueError: When the setting contains invalid JSON, invalid snapshot reference format, + or empty/whitespace snapshot name + """ + if setting is None: + raise ValueError("Setting cannot be None") + + if not setting.value or setting.value.strip() == "": + raise ValueError( + f"Invalid snapshot reference format for key '{setting.key}' " + f"(label: '{setting.label}'). Value cannot be empty." + ) + + try: + # Parse the JSON content + json_content = json.loads(setting.value) + + if not isinstance(json_content, dict): + raise ValueError( + f"Invalid snapshot reference format for key '{setting.key}' " + f"(label: '{setting.label}'). Expected JSON object." + ) + + # Extract the snapshot name + snapshot_name = json_content.get(SNAPSHOT_NAME_FIELD) + + if snapshot_name is None: + raise ValueError( + f"Invalid snapshot reference format for key '{setting.key}' " + f"(label: '{setting.label}'). The '{SNAPSHOT_NAME_FIELD}' " + f"property is required." + ) + + if not isinstance(snapshot_name, str): + raise ValueError( + f"Invalid snapshot reference format for key '{setting.key}' " + f"(label: '{setting.label}'). The '{SNAPSHOT_NAME_FIELD}' " + f"property must be a string value, but found {type(snapshot_name).__name__}." + ) + + if not snapshot_name.strip(): + raise ValueError( + f"Invalid snapshot reference format for key '{setting.key}' " + f"(label: '{setting.label}'). Snapshot name cannot be empty or whitespace." + ) + + return snapshot_name.strip() + + except json.JSONDecodeError as json_ex: + raise ValueError( + f"Invalid snapshot reference format for key '{setting.key}' " + f"(label: '{setting.label}'). Invalid JSON format." + ) from json_ex diff --git a/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/aio/_async_client_manager.py b/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/aio/_async_client_manager.py index 98457e3f1913..f1af17255a22 100644 --- a/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/aio/_async_client_manager.py +++ b/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/aio/_async_client_manager.py @@ -11,7 +11,7 @@ from typing_extensions import Self from azure.core import MatchConditions from azure.core.tracing.decorator import distributed_trace -from azure.core.exceptions import HttpResponseError +from azure.core.exceptions import HttpResponseError, AzureError from azure.appconfiguration import ( # type:ignore # pylint:disable=no-name-in-module ConfigurationSetting, FeatureFlagConfigurationSetting, @@ -26,6 +26,8 @@ ) from .._models import SettingSelector from .._constants import FEATURE_FLAG_PREFIX +from .._snapshot_reference_parser import SnapshotReferenceParser +from .._constants import SNAPSHOT_REF_CONTENT_TYPE from ._async_discovery import find_auto_failover_endpoints if TYPE_CHECKING: @@ -265,6 +267,45 @@ async def __aenter__(self): async def __aexit__(self, *args): await self._client.__aexit__(*args) + async def resolve_snapshot_reference( + self, setting: ConfigurationSetting, **kwargs + ) -> Dict[str, ConfigurationSetting]: + """ + Resolve a snapshot reference configuration setting to the actual snapshot data. + + :param ConfigurationSetting setting: The snapshot reference configuration setting + :return: A dictionary of resolved configuration settings from the snapshot + :rtype: Dict[str, ConfigurationSetting] + :raises ValueError: When the setting is not a valid snapshot reference + """ + if not setting.content_type or not SNAPSHOT_REF_CONTENT_TYPE in setting.content_type: + raise ValueError("Setting is not a snapshot reference") + + try: + # Parse the snapshot reference + snapshot_name = SnapshotReferenceParser.parse(setting) + + # Create a selector for the snapshot + snapshot_selector = SettingSelector(snapshot_name=snapshot_name) + + # Use existing load_configuration_settings to load from snapshot + configurations = await self.load_configuration_settings([snapshot_selector], **kwargs) + + # Build a dictionary keyed by configuration key + snapshot_settings = {} + for config in configurations: + # Last wins for duplicate keys during iteration + snapshot_settings[config.key] = config + + return snapshot_settings + + except AzureError as e: + # Wrap Azure errors with more context + raise ValueError( + f"Failed to resolve snapshot reference for key '{setting.key}' " + f"(label: '{setting.label}'). Azure service error occurred." + ) from e + class AsyncConfigurationClientManager(ConfigurationClientManagerBase): # pylint:disable=too-many-instance-attributes def __init__( diff --git a/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/aio/_azureappconfigurationproviderasync.py b/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/aio/_azureappconfigurationproviderasync.py index 565fb6c5bd03..2ffb02b0762d 100644 --- a/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/aio/_azureappconfigurationproviderasync.py +++ b/sdk/appconfiguration/azure-appconfiguration-provider/azure/appconfiguration/provider/aio/_azureappconfigurationproviderasync.py @@ -29,6 +29,7 @@ from .._constants import ( FEATURE_MANAGEMENT_KEY, FEATURE_FLAG_KEY, + SNAPSHOT_REF_CONTENT_TYPE, ) from .._azureappconfigurationproviderbase import ( AzureAppConfigurationProviderBase, @@ -311,7 +312,7 @@ async def _attempt_refresh( if settings_refreshed: # Configuration Settings have been refreshed - processed_settings = await self._process_configurations(configuration_settings) + processed_settings = await self._process_configurations(configuration_settings, client) processed_settings = self._process_feature_flags(processed_settings, processed_feature_flags, feature_flags) self._dict = processed_settings @@ -397,7 +398,7 @@ async def _load_all(self, **kwargs: Any) -> None: self._selects, headers=headers, **kwargs ) watched_settings = self._update_watched_settings(configuration_settings) - processed_settings = await self._process_configurations(configuration_settings) + processed_settings = await self._process_configurations(configuration_settings, client) if self._feature_flag_enabled: feature_flags: List[FeatureFlagConfigurationSetting] = await client.load_feature_flags( @@ -438,7 +439,9 @@ async def _load_all(self, **kwargs: Any) -> None: is_failover_request = True raise exception - async def _process_configurations(self, configuration_settings: List[ConfigurationSetting]) -> Dict[str, Any]: + async def _process_configurations( + self, configuration_settings: List[ConfigurationSetting], client: ConfigurationClient + ) -> Dict[str, Any]: # configuration_settings can contain duplicate keys, but they are in priority order, i.e. later settings take # precedence. Only process the settings with the highest priority (i.e. the last one in the list). unique_settings = self._deduplicate_settings(configuration_settings) @@ -455,6 +458,30 @@ async def _process_configurations(self, configuration_settings: List[Configurati if self._feature_flag_refresh_enabled: self._watched_feature_flags[(settings.key, settings.label)] = settings.etag + elif settings.content_type and SNAPSHOT_REF_CONTENT_TYPE in settings.content_type: + # Check if this is a snapshot reference + + # Track snapshot reference usage for telemetry + self._tracing_context.uses_snapshot_reference = True + try: + # Resolve the snapshot reference to actual settings + snapshot_settings = await client.resolve_snapshot_reference(settings) + + # Recursively process the resolved snapshot settings to handle feature flags, + # configuration mapping + snapshot_configuration_list = list(snapshot_settings.values()) + resolved_settings = await self._process_configurations(snapshot_configuration_list, client) + # Merge the resolved settings into our configuration + configuration_settings_processed.update(resolved_settings) + except (ValueError, AzureError) as e: + logger.warning( + "Failed to resolve snapshot reference for key '%s' (label: '%s'): %s", + settings.key, + settings.label, + str(e), + ) + # Continue processing other settings even if snapshot resolution fails + else: key = self._process_key_name(settings) value = await self._process_key_value(settings) diff --git a/sdk/appconfiguration/azure-appconfiguration-provider/samples/refresh_sample.py b/sdk/appconfiguration/azure-appconfiguration-provider/samples/refresh_sample.py index 085f9a2f5f82..8a23bc2906a4 100644 --- a/sdk/appconfiguration/azure-appconfiguration-provider/samples/refresh_sample.py +++ b/sdk/appconfiguration/azure-appconfiguration-provider/samples/refresh_sample.py @@ -11,6 +11,7 @@ from sample_utilities import get_client_modifications import os import time +import random kwargs = get_client_modifications() connection_string = os.environ.get("APPCONFIGURATION_CONNECTION_STRING") @@ -27,10 +28,13 @@ def my_callback_on_fail(error): print("Refresh failed!") +rand = random.random() +watch_key = WatchKey("message" + str(rand)) + # Connecting to Azure App Configuration using connection string, and refreshing when the configuration setting message changes config = load( connection_string=connection_string, - refresh_on=[WatchKey("message")], + refresh_on=[watch_key], refresh_interval=1, on_refresh_error=my_callback_on_fail, **kwargs, @@ -42,6 +46,10 @@ def my_callback_on_fail(error): # Updating the configuration setting configuration_setting.value = "Hello World Updated!" +configuration_setting2 = ConfigurationSetting(key="message" + str(rand), value="2") + +client.set_configuration_setting(configuration_setting=configuration_setting2) + client.set_configuration_setting(configuration_setting=configuration_setting) # Waiting for the refresh interval to pass diff --git a/sdk/appconfiguration/azure-appconfiguration-provider/tests/aio/test_async_snapshot_references_integration.py b/sdk/appconfiguration/azure-appconfiguration-provider/tests/aio/test_async_snapshot_references_integration.py new file mode 100644 index 000000000000..8b58055c2b24 --- /dev/null +++ b/sdk/appconfiguration/azure-appconfiguration-provider/tests/aio/test_async_snapshot_references_integration.py @@ -0,0 +1,321 @@ +# ------------------------------------------------------------------------- +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See License.txt in the project root for +# license information. +# -------------------------------------------------------------------------- + +import pytest +from unittest.mock import Mock, patch, AsyncMock +from azure.appconfiguration import ConfigurationSetting +from azure.appconfiguration.provider import SettingSelector +from azure.appconfiguration.provider.aio._async_client_manager import AsyncConfigurationClientManager +from azure.appconfiguration.provider._constants import SNAPSHOT_REF_CONTENT_TYPE +from azure.core.exceptions import ResourceNotFoundError, AzureError +from azure.identity.aio import DefaultAzureCredential + + +class TestAsyncClientManagerSnapshotReferences: + """Integration tests for snapshot references in the async client manager.""" + + @pytest.fixture + def mock_client(self): + """Create a mock Azure App Configuration async client.""" + client = AsyncMock() + + async def mock_list_settings(*args, **kwargs): + # Return an async iterator + for item in []: + yield item + + client.list_configuration_settings = mock_list_settings + return client + + @pytest.fixture + def client_manager(self, mock_client): + """Create an async client manager with mocked client.""" + with patch("azure.appconfiguration.aio.AzureAppConfigurationClient", return_value=mock_client): + return AsyncConfigurationClientManager( + connection_string=None, + endpoint="https://test.azconfig.io", + credential=DefaultAzureCredential(), + user_agent="test-user-agent", + retry_total=3, + retry_backoff_max=60, + replica_discovery_enabled=False, + min_backoff_sec=30, + max_backoff_sec=600, + load_balancing_enabled=False, + ) + + @pytest.mark.asyncio + async def test_resolve_snapshot_reference_success(self, client_manager, mock_client): + """Test successfully resolving a snapshot reference.""" + # Setup snapshot reference setting + snapshot_ref = ConfigurationSetting( + key="SnapshotRef1", + value='{"snapshot_name": "test-snapshot"}', + content_type=SNAPSHOT_REF_CONTENT_TYPE, + ) + + # Setup snapshot data + snapshot_settings = [ + ConfigurationSetting(key="App:Setting1", value="SnapshotValue1"), + ConfigurationSetting(key="App:Setting2", value="SnapshotValue2"), + ] + + # Mock the load_configuration_settings method to return snapshot data + with patch.object( + client_manager._original_client, "load_configuration_settings", new_callable=AsyncMock + ) as mock_load: + mock_load.return_value = snapshot_settings + + # Test resolving the snapshot reference + result = await client_manager._original_client.resolve_snapshot_reference(snapshot_ref) + + assert result == {"App:Setting1": snapshot_settings[0], "App:Setting2": snapshot_settings[1]} + mock_load.assert_called_once() + + @pytest.mark.asyncio + async def test_resolve_snapshot_reference_snapshot_not_found(self, client_manager, mock_client): + """Test resolving a snapshot reference when snapshot doesn't exist.""" + # Setup snapshot reference setting + snapshot_ref = ConfigurationSetting( + key="SnapshotRef1", + value='{"snapshot_name": "nonexistent-snapshot"}', + content_type=SNAPSHOT_REF_CONTENT_TYPE, + ) + + with patch.object( + client_manager._original_client, "load_configuration_settings", new_callable=AsyncMock + ) as mock_load: + mock_load.side_effect = AzureError("Snapshot not found") + + # Test resolving the snapshot reference + with pytest.raises(ValueError, match="Failed to resolve snapshot reference.*Azure service error occurred"): + await client_manager._original_client.resolve_snapshot_reference(snapshot_ref) + + @pytest.mark.asyncio + async def test_load_snapshot_data_success(self, client_manager, mock_client): + """Test successfully loading snapshot data.""" + # Setup snapshot data + snapshot_settings = [ + ConfigurationSetting(key="App:Setting1", value="SnapshotValue1"), + ConfigurationSetting(key="App:Setting2", value="SnapshotValue2"), + ] + + # Setup mock responses + async def mock_list_settings(*args, **kwargs): + for setting in snapshot_settings: + yield setting + + mock_client.list_configuration_settings = mock_list_settings + mock_client.get_snapshot = AsyncMock(return_value=Mock(composition_type="key")) + + # Test loading snapshot data using load_configuration_settings + selector = SettingSelector(snapshot_name="test-snapshot") + result = await client_manager._original_client.load_configuration_settings([selector]) + + assert result == snapshot_settings + + +class TestAzureAppConfigurationProviderAsyncSnapshotReferences: + """Integration tests for snapshot references in the async provider.""" + + @pytest.fixture + def mock_client(self): + """Create a mock Azure App Configuration async client.""" + client = AsyncMock() + return client + + @pytest.mark.asyncio + async def test_provider_processes_snapshot_reference(self, mock_client): + """Test that the async provider correctly processes snapshot references during configuration loading.""" + # Setup configuration settings including a snapshot reference + regular_setting = ConfigurationSetting(key="App:RegularSetting", value="RegularValue") + snapshot_ref = ConfigurationSetting( + key="SnapshotRef1", + value='{"snapshot_name": "test-snapshot"}', + content_type=SNAPSHOT_REF_CONTENT_TYPE, + ) + + # Setup snapshot data + snapshot_settings = [ + ConfigurationSetting(key="App:SnapshotSetting1", value="SnapshotValue1"), + ConfigurationSetting(key="App:SnapshotSetting2", value="SnapshotValue2"), + ] + + # Configure mock to return different data for different calls + async def mock_list_settings(*args, **kwargs): + if "snapshot_name" in kwargs: + for setting in snapshot_settings: + yield setting + else: + for setting in [regular_setting, snapshot_ref]: + yield setting + + mock_client.list_configuration_settings = mock_list_settings + + # Test with the provider + with patch("azure.appconfiguration.aio.AzureAppConfigurationClient", return_value=mock_client): + from azure.appconfiguration.provider.aio import load + + provider = await load( + endpoint="https://test.azconfig.io", + credential=DefaultAzureCredential(), + ) + + # Verify that all settings are present in the provider + assert "App:RegularSetting" in provider + assert "App:SnapshotSetting1" in provider + assert "App:SnapshotSetting2" in provider + assert provider["App:RegularSetting"] == "RegularValue" + assert provider["App:SnapshotSetting1"] == "SnapshotValue1" + assert provider["App:SnapshotSetting2"] == "SnapshotValue2" + + # The snapshot reference itself should not be in the configuration + assert "SnapshotRef1" not in provider + + await provider.close() + + @pytest.mark.asyncio + async def test_provider_handles_snapshot_reference_error(self, mock_client): + """Test that the async provider handles snapshot reference errors gracefully.""" + # Setup configuration with a snapshot reference that will fail + snapshot_ref = ConfigurationSetting( + key="SnapshotRef1", + value='{"snapshot_name": "nonexistent-snapshot"}', + content_type=SNAPSHOT_REF_CONTENT_TYPE, + ) + + async def mock_list_settings(*args, **kwargs): + if "snapshot_name" in kwargs: + raise ResourceNotFoundError("Snapshot not found") + else: + for setting in [snapshot_ref]: + yield setting + + mock_client.list_configuration_settings = mock_list_settings + + # Test that provider creation fails with appropriate error + with patch("azure.appconfiguration.aio.AzureAppConfigurationClient", return_value=mock_client): + from azure.appconfiguration.provider.aio import load + + with pytest.raises(ValueError, match="Failed to resolve snapshot reference.*Azure service error occurred"): + provider = await load( + endpoint="https://test.azconfig.io", + credential=DefaultAzureCredential(), + ) + + @pytest.mark.asyncio + async def test_provider_with_selectors_and_snapshot_references(self, mock_client): + """Test async provider behavior with selectors when processing snapshot references.""" + # Setup configuration settings + regular_setting = ConfigurationSetting(key="App:RegularSetting", value="RegularValue") + other_setting = ConfigurationSetting(key="Other:Setting", value="OtherValue") + snapshot_ref = ConfigurationSetting( + key="SnapshotRef1", + value='{"snapshot_name": "test-snapshot"}', + content_type=SNAPSHOT_REF_CONTENT_TYPE, + ) + + # Setup snapshot data (contains both App:* and Other:* keys) + snapshot_settings = [ + ConfigurationSetting(key="App:SnapshotSetting1", value="SnapshotValue1"), + ConfigurationSetting(key="Other:SnapshotSetting1", value="OtherSnapshotValue1"), + ] + + async def mock_list_settings(*args, **kwargs): + if "snapshot_name" in kwargs: + for setting in snapshot_settings: + yield setting + else: + for setting in [regular_setting, other_setting, snapshot_ref]: + yield setting + + mock_client.list_configuration_settings = mock_list_settings + + # Test with selectors that only include App:* keys + with patch("azure.appconfiguration.aio.AzureAppConfigurationClient", return_value=mock_client): + from azure.appconfiguration.provider.aio import load + + provider = await load( + endpoint="https://test.azconfig.io", + credential=DefaultAzureCredential(), + selects=[SettingSelector(key_filter="App:*")], + ) + + # Should include regular App:* setting and snapshot App:* setting + assert "App:RegularSetting" in provider + assert "App:SnapshotSetting1" in provider + + # Should not include Other:* settings (filtered out by selector) + assert "Other:Setting" not in provider + assert "Other:SnapshotSetting1" not in provider + + # Snapshot reference itself should not be present + assert "SnapshotRef1" not in provider + + await provider.close() + + @pytest.mark.asyncio + async def test_provider_recursive_snapshot_processing(self, mock_client): + """Test that async provider correctly handles recursive snapshot reference processing.""" + # Setup configuration settings with feature flags + feature_flag = ConfigurationSetting( + key=".appconfig.featureflag/TestFlag", + value='{"enabled": true, "conditions": {"client_filters": []}}', + content_type="application/vnd.microsoft.appconfig.ff+json;charset=utf-8", + ) + + snapshot_ref = ConfigurationSetting( + key="SnapshotRef1", + value='{"snapshot_name": "test-snapshot"}', + content_type=SNAPSHOT_REF_CONTENT_TYPE, + ) + + # Setup snapshot data that includes both regular settings and feature flags + snapshot_settings = [ + ConfigurationSetting(key="App:SnapshotSetting1", value="SnapshotValue1"), + ConfigurationSetting( + key=".appconfig.featureflag/SnapshotFlag", + value='{"enabled": false, "conditions": {"client_filters": []}}', + content_type="application/vnd.microsoft.appconfig.ff+json;charset=utf-8", + ), + ] + + async def mock_list_settings(*args, **kwargs): + if "snapshot_name" in kwargs: + for setting in snapshot_settings: + yield setting + else: + # Filter based on key_filter + key_filter = kwargs.get("key_filter", "") + if key_filter.startswith(".appconfig.featureflag/"): + yield feature_flag + else: + yield snapshot_ref + + mock_client.list_configuration_settings = mock_list_settings + + # Test with feature flags enabled + with patch("azure.appconfiguration.aio.AzureAppConfigurationClient", return_value=mock_client): + from azure.appconfiguration.provider.aio import load + + provider = await load( + endpoint="https://test.azconfig.io", + credential=DefaultAzureCredential(), + feature_flag_enabled=True, + ) + + # Should include regular snapshot setting + assert "App:SnapshotSetting1" in provider + assert provider["App:SnapshotSetting1"] == "SnapshotValue1" + + # Should include feature flags from both regular config and snapshot + assert "feature_management" in provider + feature_flags = provider["feature_management"]["feature_flags"] + flag_names = [flag["id"] for flag in feature_flags] + assert "TestFlag" in flag_names + assert "SnapshotFlag" in flag_names + + await provider.close() diff --git a/sdk/appconfiguration/azure-appconfiguration-provider/tests/test_request_tracing_context.py b/sdk/appconfiguration/azure-appconfiguration-provider/tests/test_request_tracing_context.py index 1c2537c00086..dfdcbb47df2c 100644 --- a/sdk/appconfiguration/azure-appconfiguration-provider/tests/test_request_tracing_context.py +++ b/sdk/appconfiguration/azure-appconfiguration-provider/tests/test_request_tracing_context.py @@ -6,6 +6,7 @@ import unittest import os from unittest.mock import patch, Mock +from azure.appconfiguration import ConfigurationSetting from azure.appconfiguration.provider._request_tracing_context import ( _RequestTracingContext, @@ -30,6 +31,11 @@ KubernetesEnvironmentVariable, APP_CONFIG_AI_MIME_PROFILE, APP_CONFIG_AICC_MIME_PROFILE, + SNAPSHOT_REFERENCE_TAG, + SNAPSHOT_REF_CONTENT_TYPE, +) +from azure.appconfiguration.provider._azureappconfigurationproviderbase import ( + AzureAppConfigurationProviderBase, ) @@ -244,110 +250,99 @@ def test_reset_feature_filter_usage(self): self.assertEqual(self.context.feature_filter_usage, {}) -class TestCreateCorrelationContextHeader(unittest.TestCase): +class TestUpdateCorrelationContextHeader(unittest.TestCase): """Test the update_correlation_context_header method.""" def setUp(self): """Set up test environment.""" self.context = _RequestTracingContext() + self.headers = {} - def test_disabled_tracing_returns_empty_string(self): + def test_disabled_tracing(self): """Test that disabled tracing returns empty string.""" with patch.dict(os.environ, {REQUEST_TRACING_DISABLED_ENVIRONMENT_VARIABLE: "true"}): - headers = {} - result = self.context.update_correlation_context_header(headers, RequestType.STARTUP, 0, False, False) - self.assertEqual(result, headers) # Should return headers unchanged + result = self.context.update_correlation_context_header(self.headers, RequestType.STARTUP, 0, False, False) + self.assertEqual(result, self.headers) self.assertNotIn("Correlation-Context", result) def test_basic_correlation_context(self): """Test basic correlation context generation.""" - headers = {} - result = self.context.update_correlation_context_header(headers, RequestType.STARTUP, 0, False, False) + result = self.context.update_correlation_context_header(self.headers, RequestType.STARTUP, 0, False, False) self.assertIn("Correlation-Context", result) self.assertIn("RequestType=Startup", result["Correlation-Context"]) - def test_correlation_context_with_replica_count(self): + def test_with_replica_count(self): """Test correlation context with replica count.""" - headers = {} - result = self.context.update_correlation_context_header(headers, RequestType.WATCH, 3, False, False) + result = self.context.update_correlation_context_header(self.headers, RequestType.WATCH, 3, False, False) self.assertIn("Correlation-Context", result) self.assertIn("RequestType=Watch", result["Correlation-Context"]) self.assertIn("ReplicaCount=3", result["Correlation-Context"]) - def test_correlation_context_with_host_type(self): + def test_with_host_type(self): """Test correlation context with host type detection.""" - with patch.object(_RequestTracingContext, "get_host_type", return_value=HostType.AZURE_FUNCTION): - # Update host_type since it's not automatically set - self.context.host_type = HostType.AZURE_FUNCTION - headers = {} - result = self.context.update_correlation_context_header(headers, RequestType.STARTUP, 0, False, False) - self.assertIn("Correlation-Context", result) - self.assertIn("Host=AzureFunction", result["Correlation-Context"]) - - def test_correlation_context_with_feature_filters(self): + self.context.host_type = HostType.AZURE_FUNCTION + result = self.context.update_correlation_context_header(self.headers, RequestType.STARTUP, 0, False, False) + self.assertIn("Correlation-Context", result) + self.assertIn("Host=AzureFunction", result["Correlation-Context"]) + + def test_with_feature_filters(self): """Test correlation context with feature filters.""" self.context.feature_filter_usage = { CUSTOM_FILTER_KEY: True, PERCENTAGE_FILTER_KEY: True, } - headers = {} - result = self.context.update_correlation_context_header(headers, RequestType.STARTUP, 0, False, False) + result = self.context.update_correlation_context_header(self.headers, RequestType.STARTUP, 0, False, False) self.assertIn("Correlation-Context", result) self.assertIn("Filter=", result["Correlation-Context"]) self.assertIn(CUSTOM_FILTER_KEY, result["Correlation-Context"]) self.assertIn(PERCENTAGE_FILTER_KEY, result["Correlation-Context"]) - def test_correlation_context_with_max_variants(self): + def test_with_max_variants(self): """Test correlation context with max variants.""" self.context.max_variants = 5 - headers = {} - result = self.context.update_correlation_context_header(headers, RequestType.STARTUP, 0, False, False) + result = self.context.update_correlation_context_header(self.headers, RequestType.STARTUP, 0, False, False) self.assertIn("Correlation-Context", result) self.assertIn("MaxVariants=5", result["Correlation-Context"]) - def test_correlation_context_with_ff_features(self): + def test_with_ff_features(self): """Test correlation context with feature flag features.""" self.context.uses_seed = True self.context.uses_telemetry = True - headers = {} - result = self.context.update_correlation_context_header(headers, RequestType.STARTUP, 0, False, False) + result = self.context.update_correlation_context_header(self.headers, RequestType.STARTUP, 0, False, False) self.assertIn("Correlation-Context", result) self.assertIn("FFFeatures=", result["Correlation-Context"]) self.assertIn(FEATURE_FLAG_USES_SEED_TAG, result["Correlation-Context"]) self.assertIn(FEATURE_FLAG_USES_TELEMETRY_TAG, result["Correlation-Context"]) @patch("azure.appconfiguration.provider._request_tracing_context.version") - def test_correlation_context_with_version(self, mock_version): + def test_with_feature_management_version(self, mock_version): """Test correlation context with feature management version.""" mock_version.return_value = "1.0.0" self.context.feature_management_version = "1.0.0" - headers = {} - result = self.context.update_correlation_context_header(headers, RequestType.STARTUP, 0, False, False) + result = self.context.update_correlation_context_header( + self.headers, RequestType.STARTUP, 0, False, True, False + ) self.assertIn("Correlation-Context", result) self.assertIn("FMPyVer=1.0.0", result["Correlation-Context"]) - def test_correlation_context_with_general_features(self): + def test_with_general_features(self): """Test correlation context with general features.""" self.context.uses_load_balancing = True self.context.uses_ai_configuration = True - headers = {} - result = self.context.update_correlation_context_header(headers, RequestType.STARTUP, 0, False, False) + result = self.context.update_correlation_context_header(self.headers, RequestType.STARTUP, 0, False, False) self.assertIn("Correlation-Context", result) self.assertIn("Features=LB+AI", result["Correlation-Context"]) - def test_correlation_context_with_tags(self): - """Test correlation context with various tags.""" - self.context.is_key_vault_configured = True - self.context.is_failover_request = True - self.context.is_push_refresh_used = True - - headers = {} - result = self.context.update_correlation_context_header(headers, RequestType.STARTUP, 0, True, False, True) + def test_with_key_vault_and_failover(self): + """Test correlation context with key vault and failover tags.""" + result = self.context.update_correlation_context_header(self.headers, RequestType.STARTUP, 2, True, False, True) self.assertIn("Correlation-Context", result) + self.assertIn("RequestType=Startup", result["Correlation-Context"]) + self.assertIn("ReplicaCount=2", result["Correlation-Context"]) self.assertIn("UsesKeyVault", result["Correlation-Context"]) self.assertIn("Failover", result["Correlation-Context"]) - def test_correlation_context_comprehensive(self): + def test_comprehensive_correlation_context(self): """Test correlation context with all features enabled.""" # Set up all possible features self.context.host_type = HostType.AZURE_WEB_APP @@ -359,8 +354,7 @@ def test_correlation_context_comprehensive(self): self.context.is_key_vault_configured = True self.context.is_failover_request = True - headers = {} - result = self.context.update_correlation_context_header(headers, RequestType.WATCH, 2, True, True, True) + result = self.context.update_correlation_context_header(self.headers, RequestType.WATCH, 2, True, True, True) # Verify all components are present self.assertIn("Correlation-Context", result) @@ -377,71 +371,145 @@ def test_correlation_context_comprehensive(self): self.assertIn("Failover", correlation_context) -class TestUpdateCorrelationContextHeader(unittest.TestCase): - """Test the update_correlation_context_header method.""" +class TestSnapshotReferenceTracking(unittest.TestCase): + """Test snapshot reference tracking in request tracing context.""" - def setUp(self): - """Set up test environment.""" - self.context = _RequestTracingContext() - self.headers = {} + def test_snapshot_reference_tag_constant(self): + """Test that the snapshot reference tag constant has the expected value.""" + self.assertEqual(SNAPSHOT_REFERENCE_TAG, "UsesSnapshotReference=true") - def test_update_correlation_context_header_basic(self): - """Test basic correlation context header update.""" - result = self.context.update_correlation_context_header( - self.headers, - RequestType.STARTUP, - 2, # replica_count - True, # uses_key_vault - True, # feature_flag_enabled - False, # is_failover_request + def test_initialization(self): + """Test that request tracing context initializes snapshot reference tracking.""" + context = _RequestTracingContext() + self.assertFalse(context.uses_snapshot_reference) + + def test_set_snapshot_reference_usage(self): + """Test setting snapshot reference usage in tracing context.""" + context = _RequestTracingContext() + + # Initially false + self.assertFalse(context.uses_snapshot_reference) + + # Set to true + context.uses_snapshot_reference = True + self.assertTrue(context.uses_snapshot_reference) + + # Set back to false + context.uses_snapshot_reference = False + self.assertFalse(context.uses_snapshot_reference) + + def test_correlation_context_without_snapshot_reference(self): + """Test correlation context header when not using snapshot references.""" + context = _RequestTracingContext() + context.uses_snapshot_reference = False + + headers = {} + updated_headers = context.update_correlation_context_header( + headers=headers, + request_type="Startup", + replica_count=0, + uses_key_vault=False, + feature_flag_enabled=False, + is_failover_request=False, ) - self.assertIn("Correlation-Context", result) - self.assertIn("RequestType=Startup", result["Correlation-Context"]) - self.assertIn("ReplicaCount=2", result["Correlation-Context"]) - self.assertIn("UsesKeyVault", result["Correlation-Context"]) + correlation_header = updated_headers.get("Correlation-Context", "") + self.assertIn("RequestType=Startup", correlation_header) + self.assertNotIn(SNAPSHOT_REFERENCE_TAG, correlation_header) - @patch("azure.appconfiguration.provider._request_tracing_context.version") - def test_update_correlation_context_header_with_version(self, mock_version): - """Test correlation context header update with feature management version.""" - mock_version.return_value = "1.0.0" + def test_correlation_context_with_snapshot_reference(self): + """Test correlation context header when using snapshot references.""" + context = _RequestTracingContext() + context.uses_snapshot_reference = True - result = self.context.update_correlation_context_header( - self.headers, - RequestType.STARTUP, - 0, # replica_count - False, # uses_key_vault - True, # feature_flag_enabled - False, # is_failover_request + headers = {} + updated_headers = context.update_correlation_context_header( + headers=headers, + request_type="Startup", + replica_count=0, + uses_key_vault=False, + feature_flag_enabled=False, + is_failover_request=False, ) - self.assertIn("Correlation-Context", result) - self.assertIn("FMPyVer=1.0.0", result["Correlation-Context"]) + correlation_header = updated_headers.get("Correlation-Context", "") + self.assertIn("RequestType=Startup", correlation_header) + self.assertIn(SNAPSHOT_REFERENCE_TAG, correlation_header) + + def test_snapshot_reference_detection_in_provider_base(self): + """Test that snapshot reference detection works in provider base.""" + # Test with snapshot reference content type + snapshot_ref_setting = ConfigurationSetting( + key="SnapshotRef1", + value='{"snapshot_name": "test-snapshot"}', + content_type=SNAPSHOT_REF_CONTENT_TYPE, + ) - def test_update_correlation_context_header_failover(self): - """Test correlation context header update with failover request.""" - result = self.context.update_correlation_context_header( - self.headers, - RequestType.WATCH, - 1, # replica_count - False, # uses_key_vault - False, # feature_flag_enabled - True, # is_failover_request + self.assertTrue( + snapshot_ref_setting.content_type and SNAPSHOT_REF_CONTENT_TYPE in snapshot_ref_setting.content_type ) - self.assertIn("Correlation-Context", result) - self.assertIn("Failover", result["Correlation-Context"]) + # Test with regular content type + regular_setting = ConfigurationSetting( + key="RegularKey", + value="RegularValue", + content_type="application/vnd.microsoft.appconfig.kv+json", + ) - def test_update_correlation_context_header_disabled_tracing(self): - """Test correlation context header update with disabled tracing.""" - with patch.dict(os.environ, {REQUEST_TRACING_DISABLED_ENVIRONMENT_VARIABLE: "true"}): - result = self.context.update_correlation_context_header( - self.headers, - RequestType.STARTUP, - 1, # replica_count - False, # uses_key_vault - False, # feature_flag_enabled - False, # is_failover_request - ) + self.assertFalse(regular_setting.content_type and SNAPSHOT_REF_CONTENT_TYPE in regular_setting.content_type) - self.assertNotIn("Correlation-Context", result) + def test_tracing_context_updated_during_processing(self): + """Test that tracing context is updated when processing configuration with snapshot references.""" + provider = Mock(spec=AzureAppConfigurationProviderBase) + provider._tracing_context = _RequestTracingContext() + + setting = ConfigurationSetting( + key="SnapshotRef1", + value='{"snapshot_name": "test-snapshot"}', + content_type=SNAPSHOT_REF_CONTENT_TYPE, + ) + + # Simulate checking for snapshot reference content type + if setting.content_type and SNAPSHOT_REF_CONTENT_TYPE in setting.content_type: + provider._tracing_context.uses_snapshot_reference = True + + self.assertTrue(provider._tracing_context.uses_snapshot_reference) + + def test_tracing_context_not_updated_for_regular_settings(self): + """Test that tracing context is not updated for regular settings.""" + provider = Mock(spec=AzureAppConfigurationProviderBase) + provider._tracing_context = _RequestTracingContext() + + setting = ConfigurationSetting( + key="RegularKey", + value="RegularValue", + content_type="application/vnd.microsoft.appconfig.kv+json", + ) + + # Simulate checking for snapshot reference content type (should not match) + if setting.content_type and SNAPSHOT_REF_CONTENT_TYPE in setting.content_type: + provider._tracing_context.uses_snapshot_reference = True + + self.assertFalse(provider._tracing_context.uses_snapshot_reference) + + def test_correlation_context_with_multiple_tags(self): + """Test correlation context header format when multiple tags are present.""" + context = _RequestTracingContext() + context.uses_snapshot_reference = True + + headers = {} + updated_headers = context.update_correlation_context_header( + headers=headers, + request_type="Startup", + replica_count=2, + uses_key_vault=True, + feature_flag_enabled=False, + is_failover_request=True, + ) + + correlation_header = updated_headers.get("Correlation-Context", "") + self.assertIn("RequestType=Startup", correlation_header) + self.assertIn("ReplicaCount=2", correlation_header) + self.assertIn("UsesKeyVault", correlation_header) + self.assertIn("Failover", correlation_header) + self.assertIn(SNAPSHOT_REFERENCE_TAG, correlation_header) diff --git a/sdk/appconfiguration/azure-appconfiguration-provider/tests/test_snapshot_references.py b/sdk/appconfiguration/azure-appconfiguration-provider/tests/test_snapshot_references.py new file mode 100644 index 000000000000..e6a5d885ea19 --- /dev/null +++ b/sdk/appconfiguration/azure-appconfiguration-provider/tests/test_snapshot_references.py @@ -0,0 +1,136 @@ +# ------------------------------------------------------------------------- +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See License.txt in the project root for +# license information. +# -------------------------------------------------------------------------- + +import pytest +from unittest.mock import Mock, patch +from azure.appconfiguration import ConfigurationSetting +from azure.appconfiguration.provider._snapshot_reference_parser import SnapshotReferenceParser +from azure.appconfiguration.provider._constants import ( + SNAPSHOT_REF_CONTENT_TYPE, +) + + +class TestSnapshotReferenceParser: + """Tests for the SnapshotReferenceParser class.""" + + def test_parse_valid_snapshot_reference(self): + """Test parsing a valid snapshot reference.""" + setting = ConfigurationSetting( + key="SnapshotRef1", + value='{"snapshot_name": "test-snapshot"}', + content_type=SNAPSHOT_REF_CONTENT_TYPE, + ) + + result = SnapshotReferenceParser.parse(setting) + + assert isinstance(result, str) + assert result == "test-snapshot" + + def test_parse_snapshot_reference_with_whitespace(self): + """Test parsing a snapshot reference with whitespace in snapshot name.""" + setting = ConfigurationSetting( + key="SnapshotRef1", + value='{"snapshot_name": " test-snapshot "}', + content_type=SNAPSHOT_REF_CONTENT_TYPE, + ) + + result = SnapshotReferenceParser.parse(setting) + + assert isinstance(result, str) + assert result == "test-snapshot" # Should be trimmed + + def test_parse_none_setting_raises_error(self): + """Test that parsing None setting raises ValueError.""" + with pytest.raises(ValueError, match="Setting cannot be None"): + SnapshotReferenceParser.parse(None) + + def test_parse_empty_value_raises_error(self): + """Test that parsing empty value raises ValueError.""" + setting = ConfigurationSetting( + key="SnapshotRef1", + value="{}", + content_type=SNAPSHOT_REF_CONTENT_TYPE, + ) + + with pytest.raises(ValueError, match="Invalid snapshot reference format.*snapshot_name.*is required"): + SnapshotReferenceParser.parse(setting) + + def test_parse_whitespace_value_raises_error(self): + """Test that parsing whitespace-only value raises ValueError.""" + setting = ConfigurationSetting( + key="SnapshotRef1", + value=" ", + content_type=SNAPSHOT_REF_CONTENT_TYPE, + ) + + with pytest.raises(ValueError, match="Invalid snapshot reference format.*Value cannot be empty"): + SnapshotReferenceParser.parse(setting) + + def test_parse_invalid_json_raises_error(self): + """Test that parsing invalid JSON raises ValueError.""" + setting = ConfigurationSetting( + key="SnapshotRef1", + value="{invalid json, missing quotes}", + content_type=SNAPSHOT_REF_CONTENT_TYPE, + ) + + with pytest.raises(ValueError, match="Invalid snapshot reference format.*Invalid JSON format"): + SnapshotReferenceParser.parse(setting) + + def test_parse_non_object_json_raises_error(self): + """Test that parsing non-object JSON raises ValueError.""" + setting = ConfigurationSetting( + key="SnapshotRef1", + value='"just a string"', + content_type=SNAPSHOT_REF_CONTENT_TYPE, + ) + + with pytest.raises(ValueError, match="Invalid snapshot reference format.*Expected JSON object"): + SnapshotReferenceParser.parse(setting) + + def test_parse_missing_snapshot_name_property_raises_error(self): + """Test that missing snapshot_name property raises ValueError.""" + setting = ConfigurationSetting( + key="SnapshotRef1", + value='{"other_property": "value"}', + content_type=SNAPSHOT_REF_CONTENT_TYPE, + ) + + with pytest.raises(ValueError, match="Invalid snapshot reference format.*property is required"): + SnapshotReferenceParser.parse(setting) + + def test_parse_non_string_snapshot_name_raises_error(self): + """Test that non-string snapshot_name property raises ValueError.""" + setting = ConfigurationSetting( + key="SnapshotRef1", + value='{"snapshot_name": 123}', + content_type=SNAPSHOT_REF_CONTENT_TYPE, + ) + + with pytest.raises(ValueError, match="Invalid snapshot reference format.*must be a string value"): + SnapshotReferenceParser.parse(setting) + + def test_parse_empty_snapshot_name_raises_error(self): + """Test that empty snapshot_name raises ValueError.""" + setting = ConfigurationSetting( + key="SnapshotRef1", + value='{"snapshot_name": ""}', + content_type=SNAPSHOT_REF_CONTENT_TYPE, + ) + + with pytest.raises(ValueError, match="Invalid snapshot reference format.*cannot be empty or whitespace"): + SnapshotReferenceParser.parse(setting) + + def test_parse_whitespace_snapshot_name_raises_error(self): + """Test that whitespace-only snapshot_name raises ValueError.""" + setting = ConfigurationSetting( + key="SnapshotRef1", + value='{"snapshot_name": " "}', + content_type=SNAPSHOT_REF_CONTENT_TYPE, + ) + + with pytest.raises(ValueError, match="Invalid snapshot reference format.*cannot be empty or whitespace"): + SnapshotReferenceParser.parse(setting) diff --git a/sdk/appconfiguration/azure-appconfiguration-provider/tests/test_snapshot_references_integration.py b/sdk/appconfiguration/azure-appconfiguration-provider/tests/test_snapshot_references_integration.py new file mode 100644 index 000000000000..d0a372373e8c --- /dev/null +++ b/sdk/appconfiguration/azure-appconfiguration-provider/tests/test_snapshot_references_integration.py @@ -0,0 +1,193 @@ +# ------------------------------------------------------------------------- +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See License.txt in the project root for +# license information. +# -------------------------------------------------------------------------- + +import pytest +from azure.appconfiguration import ConfigurationSetting +from azure.appconfiguration.provider._snapshot_reference_parser import SnapshotReferenceParser +from azure.appconfiguration.provider._constants import SNAPSHOT_REF_CONTENT_TYPE +from azure.appconfiguration.provider._request_tracing_context import _RequestTracingContext + + +class TestSnapshotReferenceParser: + """Integration tests for snapshot reference parser.""" + + def test_parse_valid_snapshot_reference(self): + """Test parsing a valid snapshot reference JSON.""" + setting = ConfigurationSetting( + key="SnapshotRef1", + value='{"snapshot_name": "test-snapshot"}', + content_type=SNAPSHOT_REF_CONTENT_TYPE, + ) + + result = SnapshotReferenceParser.parse(setting) + assert result == "test-snapshot" + + def test_parse_snapshot_reference_with_whitespace(self): + """Test parsing snapshot reference with whitespace in snapshot name.""" + setting = ConfigurationSetting( + key="SnapshotRef1", + value='{"snapshot_name": " test-snapshot "}', + content_type=SNAPSHOT_REF_CONTENT_TYPE, + ) + + result = SnapshotReferenceParser.parse(setting) + assert result == "test-snapshot" # Should be trimmed + + def test_parse_invalid_json(self): + """Test parsing invalid JSON raises ValueError.""" + setting = ConfigurationSetting( + key="SnapshotRef1", + value='{"invalid": json}', # Invalid JSON + content_type=SNAPSHOT_REF_CONTENT_TYPE, + ) + + with pytest.raises(ValueError, match="Invalid snapshot reference format.*Invalid JSON format"): + SnapshotReferenceParser.parse(setting) + + def test_parse_missing_snapshot_name_field(self): + """Test parsing JSON without snapshot_name field raises ValueError.""" + setting = ConfigurationSetting( + key="SnapshotRef1", + value='{"other_field": "value"}', + content_type=SNAPSHOT_REF_CONTENT_TYPE, + ) + + with pytest.raises(ValueError, match="Invalid snapshot reference format.*'snapshot_name' property is required"): + SnapshotReferenceParser.parse(setting) + + def test_parse_empty_snapshot_name(self): + """Test parsing JSON with empty snapshot name raises ValueError.""" + setting = ConfigurationSetting( + key="SnapshotRef1", + value='{"snapshot_name": ""}', + content_type=SNAPSHOT_REF_CONTENT_TYPE, + ) + + with pytest.raises(ValueError, match="Invalid snapshot reference format.*Snapshot name cannot be empty"): + SnapshotReferenceParser.parse(setting) + + def test_parse_whitespace_only_snapshot_name(self): + """Test parsing JSON with whitespace-only snapshot name raises ValueError.""" + setting = ConfigurationSetting( + key="SnapshotRef1", + value='{"snapshot_name": " "}', + content_type=SNAPSHOT_REF_CONTENT_TYPE, + ) + + with pytest.raises(ValueError, match="Invalid snapshot reference format.*Snapshot name cannot be empty"): + SnapshotReferenceParser.parse(setting) + + def test_parse_non_string_snapshot_name(self): + """Test parsing JSON with non-string snapshot name raises ValueError.""" + setting = ConfigurationSetting( + key="SnapshotRef1", + value='{"snapshot_name": 123}', # Number instead of string + content_type=SNAPSHOT_REF_CONTENT_TYPE, + ) + + with pytest.raises(ValueError, match="Invalid snapshot reference format.*must be a string value"): + SnapshotReferenceParser.parse(setting) + + def test_parse_none_setting(self): + """Test parsing None setting raises ValueError.""" + with pytest.raises(ValueError, match="Setting cannot be None"): + SnapshotReferenceParser.parse(None) + + def test_parse_empty_value(self): + """Test parsing setting with empty value raises ValueError.""" + setting = ConfigurationSetting( + key="SnapshotRef1", + value="", + content_type=SNAPSHOT_REF_CONTENT_TYPE, + ) + + with pytest.raises(ValueError, match="Invalid snapshot reference format.*Value cannot be empty"): + SnapshotReferenceParser.parse(setting) + + +class TestSnapshotReferenceContentTypeDetection: + """Integration tests for snapshot reference content type detection.""" + + def test_snapshot_reference_content_type_detection(self): + """Test that snapshot reference content type is correctly detected.""" + # Test with snapshot reference content type + snapshot_ref_setting = ConfigurationSetting( + key="SnapshotRef1", + value='{"snapshot_name": "test-snapshot"}', + content_type=SNAPSHOT_REF_CONTENT_TYPE, + ) + + # Should detect as snapshot reference + assert snapshot_ref_setting.content_type and SNAPSHOT_REF_CONTENT_TYPE in snapshot_ref_setting.content_type + + def test_regular_setting_content_type_detection(self): + """Test that regular settings are not detected as snapshot references.""" + # Test with regular content type + regular_setting = ConfigurationSetting( + key="RegularKey", + value="RegularValue", + content_type="application/vnd.microsoft.appconfig.kv+json", + ) + + # Should not detect as snapshot reference + assert not (regular_setting.content_type and SNAPSHOT_REF_CONTENT_TYPE in regular_setting.content_type) + + def test_feature_flag_content_type_detection(self): + """Test that feature flags are not detected as snapshot references.""" + # Test with feature flag content type + feature_flag_setting = ConfigurationSetting( + key=".appconfig.featureflag/TestFlag", + value='{"enabled": true}', + content_type="application/vnd.microsoft.appconfig.ff+json;charset=utf-8", + ) + + # Should not detect as snapshot reference + assert not ( + feature_flag_setting.content_type and SNAPSHOT_REF_CONTENT_TYPE in feature_flag_setting.content_type + ) + + def test_key_vault_reference_content_type_detection(self): + """Test that Key Vault references are not detected as snapshot references.""" + # Test with Key Vault reference content type + kv_ref_setting = ConfigurationSetting( + key="DatabaseConnectionString", + value='{"uri": "https://vault.vault.azure.net/secrets/connectionstring"}', + content_type="application/vnd.microsoft.appconfig.keyvaultref+json;charset=utf-8", + ) + + # Should not detect as snapshot reference + assert not (kv_ref_setting.content_type and SNAPSHOT_REF_CONTENT_TYPE in kv_ref_setting.content_type) + + +class TestRequestTracingContextIntegration: + """Integration tests for request tracing context with snapshot references.""" + + def test_request_tracing_context_snapshot_reference_tracking(self): + """Test that request tracing context properly tracks snapshot reference usage.""" + context = _RequestTracingContext() + + # Initially should not use snapshot references + assert not context.uses_snapshot_reference + + # After enabling, should report usage + context.uses_snapshot_reference = True + assert context.uses_snapshot_reference + + def test_request_tracing_context_correlation_header(self): + """Test correlation context header generation with snapshot reference usage.""" + context = _RequestTracingContext() + context.uses_snapshot_reference = True + + # Test correlation context header update + headers = {"existing": "header"} + context.update_correlation_context_header(headers) + + # Should contain correlation context header + assert "Correlation-Context" in headers + correlation_header = headers["Correlation-Context"] + + # Should contain snapshot reference tag + assert "UsesSnapshotReference" in correlation_header