diff --git a/src/azure-cli/azure/cli/command_modules/appservice/custom.py b/src/azure-cli/azure/cli/command_modules/appservice/custom.py index 04bfc64a9ab..09f09208e30 100644 --- a/src/azure-cli/azure/cli/command_modules/appservice/custom.py +++ b/src/azure-cli/azure/cli/command_modules/appservice/custom.py @@ -511,9 +511,86 @@ def update_app_settings_functionapp(cmd, resource_group_name, name, settings=Non return update_app_settings(cmd, resource_group_name, name, settings, slot, slot_settings) +def _parse_simple_key_value_setting(s, dest): + """ + Parse simple key=value settings format. + + Parameters + ---------- + s : str + The setting string to parse. + dest : dict + Dictionary to store the parsed setting. + + Returns + ------- + bool + True if parsing succeeded, False otherwise. + """ + if ('=' in s and not s.lstrip().startswith(('{"', "[", "{")) and + not s.startswith('@')): # @ indicates file input + try: + setting_name, value = s.split('=', 1) + dest[setting_name] = value + return True + except ValueError: + pass # Fall back to JSON parsing if split fails + return False + + +def _parse_json_setting(s, result, slot_result, setting_type): + """ + Parse JSON format settings. + + Parameters: + s (str): The input string containing JSON-formatted settings. + result (dict): A dictionary to store the parsed key-value pairs from the settings. + slot_result (dict): A dictionary to store slot setting flags for each key. + setting_type (str): The type of settings being parsed, either "SlotSettings" or "Settings". + + Returns: + bool: True if parsing was successful, False otherwise. + """ + try: + temp = shell_safe_json_parse(s) + if isinstance(temp, list): # Accept the output of the "list" command + for t in temp: + if 'slotSetting' in t.keys(): + slot_result[t['name']] = t['slotSetting'] + elif setting_type == "SlotSettings": + slot_result[t['name']] = True + result[t['name']] = t['value'] + else: + # Handle JSON objects: setting_type is either "SlotSettings" or "Settings" + # Different logic needed for slot settings vs regular settings + if setting_type == "SlotSettings": + # For slot settings JSON objects, add values to result and mark as slot settings + result.update(temp) + for key in temp: + slot_result[key] = True + else: + # For regular settings JSON objects, add values to result only + result.update(temp) + return True + except InvalidArgumentValueError: + return False + + +def _parse_fallback_key_value_setting(s, result): + """Parse key=value as fallback when JSON parsing fails.""" + try: + setting_name, value = s.split('=', 1) + except ValueError as ex: + raise InvalidArgumentValueError( + f"Invalid setting format: '{s}'. Expected 'key=value' format or valid JSON.", + recommendation="Use 'key=value' format or provide valid JSON like '{\"key\": \"value\"}'." + ) from ex + result[setting_name] = value + + def update_app_settings(cmd, resource_group_name, name, settings=None, slot=None, slot_settings=None): if not settings and not slot_settings: - raise MutuallyExclusiveArgumentError('Usage Error: --settings |--slot-settings') + raise MutuallyExclusiveArgumentError('Please provide either --settings or --slot-settings parameter.') settings = settings or [] slot_settings = slot_settings or [] @@ -521,50 +598,25 @@ def update_app_settings(cmd, resource_group_name, name, settings=None, slot=None app_settings = _generic_site_operation(cmd.cli_ctx, resource_group_name, name, 'list_application_settings', slot) result, slot_result = {}, {} - # pylint: disable=too-many-nested-blocks + for src, dest, setting_type in [(settings, result, "Settings"), (slot_settings, slot_result, "SlotSettings")]: for s in src: - # Check if this looks like a simple key=value pair without JSON/dict syntax - # If so, parse it directly to avoid unnecessary warnings from ast.literal_eval - if ('=' in s and not s.lstrip().startswith(('{"', "[", "{")) and - not s.startswith('@')): # @ indicates file input - try: - setting_name, value = s.split('=', 1) - dest[setting_name] = value - continue - except ValueError: - pass # Fall back to JSON parsing if split fails + # Try simple key=value parsing first + if _parse_simple_key_value_setting(s, dest): + continue - try: - temp = shell_safe_json_parse(s) - if isinstance(temp, list): # a bit messy, but we'd like accept the output of the "list" command - for t in temp: - if 'slotSetting' in t.keys(): - slot_result[t['name']] = t['slotSetting'] - elif setting_type == "SlotSettings": - slot_result[t['name']] = True - result[t['name']] = t['value'] - else: - dest.update(temp) - except CLIError: - setting_name, value = s.split('=', 1) - dest[setting_name] = value - result.update(dest) + # Try JSON parsing + if _parse_json_setting(s, result, slot_result, setting_type): + continue + + # Fallback to key=value parsing with error handling + _parse_fallback_key_value_setting(s, result) for setting_name, value in result.items(): app_settings.properties[setting_name] = value client = web_client_factory(cmd.cli_ctx) - -# TODO: Centauri currently return wrong payload for update appsettings, remove this once backend has the fix. - if is_centauri_functionapp(cmd, resource_group_name, name): - update_application_settings_polling(cmd, resource_group_name, name, app_settings, slot, client) - result = _generic_site_operation(cmd.cli_ctx, resource_group_name, name, 'list_application_settings', slot) - else: - result = _generic_settings_operation(cmd.cli_ctx, resource_group_name, name, - 'update_application_settings', - app_settings, slot, client) - + # Process slot configurations before updating application settings to ensure proper configuration order. app_settings_slot_cfg_names = [] if slot_result: slot_cfg_names = client.web_apps.list_slot_configuration_names(resource_group_name, name) @@ -578,6 +630,15 @@ def update_app_settings(cmd, resource_group_name, name, settings=None, slot=None app_settings_slot_cfg_names = slot_cfg_names.app_setting_names client.web_apps.update_slot_configuration_names(resource_group_name, name, slot_cfg_names) +# TODO: Centauri currently return wrong payload for update appsettings, remove this once backend has the fix. + if is_centauri_functionapp(cmd, resource_group_name, name): + update_application_settings_polling(cmd, resource_group_name, name, app_settings, slot, client) + result = _generic_site_operation(cmd.cli_ctx, resource_group_name, name, 'list_application_settings', slot) + else: + result = _generic_settings_operation(cmd.cli_ctx, resource_group_name, name, + 'update_application_settings', + app_settings, slot, client) + return _build_app_settings_output(result.properties, app_settings_slot_cfg_names, redact=True) @@ -597,7 +658,7 @@ def update_application_settings_polling(cmd, resource_group_name, name, app_sett time.sleep(5) r = send_raw_request(cmd.cli_ctx, method='get', url=poll_url) else: - raise CLIError(ex) + raise AzureResponseError(f"Failed to update application settings: {str(ex)}") from ex def add_azure_storage_account(cmd, resource_group_name, name, custom_id, storage_type, account_name, diff --git a/src/azure-cli/azure/cli/command_modules/appservice/tests/latest/test_webapp_commands_thru_mock.py b/src/azure-cli/azure/cli/command_modules/appservice/tests/latest/test_webapp_commands_thru_mock.py index e916920594e..ee657453c56 100644 --- a/src/azure-cli/azure/cli/command_modules/appservice/tests/latest/test_webapp_commands_thru_mock.py +++ b/src/azure-cli/azure/cli/command_modules/appservice/tests/latest/test_webapp_commands_thru_mock.py @@ -10,6 +10,9 @@ from azure.mgmt.web import WebSiteManagementClient from knack.util import CLIError +from azure.cli.core.azclierror import (InvalidArgumentValueError, + MutuallyExclusiveArgumentError, + AzureResponseError) from azure.cli.command_modules.appservice.custom import (set_deployment_user, update_git_token, add_hostname, update_site_configs, @@ -27,7 +30,9 @@ list_snapshots, restore_snapshot, create_managed_ssl_cert, - add_github_actions) + add_github_actions, + update_app_settings, + update_application_settings_polling) # pylint: disable=line-too-long from azure.cli.core.profiles import ResourceType @@ -463,6 +468,141 @@ def test_create_managed_ssl_cert(self, generic_site_op_mock, client_factory_mock certificate_envelope=cert_def) + def test_update_app_settings_error_handling_no_parameters(self): + """Test that MutuallyExclusiveArgumentError is raised when neither settings nor slot_settings are provided.""" + cmd_mock = _get_test_cmd() + + # Test missing both parameters - should fail early without calling any services + with self.assertRaisesRegex(MutuallyExclusiveArgumentError, + "Please provide either --settings or --slot-settings parameter"): + update_app_settings(cmd_mock, 'test-rg', 'test-app') + + @mock.patch('azure.cli.command_modules.appservice.custom._generic_site_operation') + @mock.patch('azure.cli.command_modules.appservice.custom.shell_safe_json_parse') + def test_update_app_settings_error_handling_invalid_format(self, mock_json_parse, mock_site_op): + """Test that InvalidArgumentValueError is raised for invalid setting formats.""" + cmd_mock = _get_test_cmd() + + # Setup minimal mocks needed to reach the error handling code + mock_app_settings = mock.MagicMock() + mock_app_settings.properties = {} + mock_site_op.return_value = mock_app_settings + + # Mock shell_safe_json_parse to raise InvalidArgumentValueError (simulating invalid JSON) + mock_json_parse.side_effect = InvalidArgumentValueError("Invalid JSON format") + + # Test invalid format that can't be parsed as JSON or key=value + invalid_setting = "invalid_format_no_equals_no_json" + expected_message = r"Invalid setting format.*Expected 'key=value' format or valid JSON" + + with self.assertRaisesRegex(InvalidArgumentValueError, expected_message): + update_app_settings(cmd_mock, 'test-rg', 'test-app', settings=[invalid_setting]) + + @mock.patch('azure.cli.command_modules.appservice.custom._generic_site_operation') + @mock.patch('azure.cli.command_modules.appservice.custom.shell_safe_json_parse') + def test_update_app_settings_error_handling_invalid_format_no_equals(self, mock_json_parse, mock_site_op): + """Test ValueError path when shell_safe_json_parse raises InvalidArgumentValueError and string contains no '='.""" + cmd_mock = _get_test_cmd() + + # Setup minimal mocks needed to reach the error handling code + mock_app_settings = mock.MagicMock() + mock_app_settings.properties = {} + mock_site_op.return_value = mock_app_settings + + # Mock shell_safe_json_parse to raise InvalidArgumentValueError + mock_json_parse.side_effect = InvalidArgumentValueError("Invalid JSON format") + + # Test invalid format with no equals sign - this should trigger ValueError in split('=', 1) + invalid_setting_no_equals = "invalidformatthatcontainsnoequalsign" + expected_message = r"Invalid setting format.*Expected 'key=value' format or valid JSON" + + with self.assertRaisesRegex(InvalidArgumentValueError, expected_message): + update_app_settings(cmd_mock, 'test-rg', 'test-app', settings=[invalid_setting_no_equals]) + + @mock.patch('azure.cli.command_modules.appservice.custom._generic_site_operation') + @mock.patch('azure.cli.command_modules.appservice.custom.web_client_factory') + @mock.patch('azure.cli.command_modules.appservice.custom.is_centauri_functionapp') + @mock.patch('azure.cli.command_modules.appservice.custom._generic_settings_operation') + @mock.patch('azure.cli.command_modules.appservice.custom._build_app_settings_output') + def test_update_app_settings_success_key_value_format(self, mock_build, mock_settings_op, mock_centauri, + mock_client_factory, mock_site_op): + """Test successful processing of key=value format settings.""" + cmd_mock = _get_test_cmd() + + # Setup mocks + mock_app_settings = mock.MagicMock() + mock_app_settings.properties = {} + mock_site_op.return_value = mock_app_settings + + mock_client = mock.MagicMock() + mock_client_factory.return_value = mock_client + mock_centauri.return_value = False + mock_settings_op.return_value = mock_app_settings + mock_build.return_value = {"KEY1": "value1", "KEY2": "value2"} + + # Test valid key=value format + result = update_app_settings(cmd_mock, 'test-rg', 'test-app', + settings=['KEY1=value1', 'KEY2=value2']) + + # Verify the function completed successfully + self.assertEqual(result["KEY1"], "value1") + self.assertEqual(result["KEY2"], "value2") + mock_build.assert_called_once() + + @mock.patch('azure.cli.command_modules.appservice.custom.send_raw_request') + def test_update_application_settings_polling_error_handling(self, mock_send_request): + """Test that AzureResponseError is raised in polling function when appropriate.""" + cmd_mock = _get_test_cmd() + + # Mock an exception that doesn't have the expected structure + class MockException(Exception): + def __init__(self): + self.response = mock.MagicMock() + self.response.status_code = 400 # Not 202 + self.response.headers = {} + + # Mock _generic_settings_operation to raise the exception + with mock.patch('azure.cli.command_modules.appservice.custom._generic_settings_operation') as mock_settings_op, \ + self.assertRaisesRegex(AzureResponseError, "Failed to update application settings"): + mock_settings_op.side_effect = MockException() + update_application_settings_polling(cmd_mock, 'test-rg', 'test-app', + mock.MagicMock(), None, mock.MagicMock()) + + @mock.patch('azure.cli.command_modules.appservice.custom._generic_site_operation') + @mock.patch('azure.cli.command_modules.appservice.custom.web_client_factory') + @mock.patch('azure.cli.command_modules.appservice.custom.is_centauri_functionapp') + @mock.patch('azure.cli.command_modules.appservice.custom._generic_settings_operation') + @mock.patch('azure.cli.command_modules.appservice.custom._build_app_settings_output') + def test_update_app_settings_success_with_slot_settings(self, mock_build, mock_settings_op, mock_centauri, + mock_client_factory, mock_site_op): + """Test successful processing with slot settings.""" + cmd_mock = _get_test_cmd() + + # Setup mocks + mock_app_settings = mock.MagicMock() + mock_app_settings.properties = {} + mock_site_op.return_value = mock_app_settings + + mock_client = mock.MagicMock() + mock_slot_config = mock.MagicMock() + mock_slot_config.app_setting_names = [] + mock_client.web_apps.list_slot_configuration_names.return_value = mock_slot_config + mock_client_factory.return_value = mock_client + mock_centauri.return_value = False + mock_settings_op.return_value = mock_app_settings + mock_build.return_value = {"SLOT_KEY": "slot_value"} + + # Test with slot settings + result = update_app_settings(cmd_mock, 'test-rg', 'test-app', + settings=['REGULAR_KEY=regular_value'], + slot_settings=['SLOT_KEY=slot_value']) + + # Verify slot configuration was updated + mock_client.web_apps.list_slot_configuration_names.assert_called_once() + mock_client.web_apps.update_slot_configuration_names.assert_called_once() + mock_build.assert_called_once() + + class FakedResponse: # pylint: disable=too-few-public-methods def __init__(self, status_code): self.status_code = status_code