[Draft] Support az what-if

src/azure-cli/azure/cli/command_modules/util/_help.py

@@ -92,3 +92,13 @@
 - name: List resource groups by bringing your own access token
   text: az demo byo-access-token --access-token "eyJ0eXAiO..." --subscription-id 00000000-0000-0000-0000-000000000000
 """
+
+helps['what-if'] = """
+type: command
+short-summary: Create a sandboxed what-if simulation of Azure CLI scripts to visualize infrastructure changes before execution.
+examples:
+- name: Simulate a what-if scenario for a provided script
+  text: az what-if --script-path "/path/to/your/script.sh"
+- name: Simulate a what-if scenario for a specific subscription
+  text: az what-if --script-path "/path/to/your/script.sh" --subscription 00000000-0000-0000-0000-000000000000
+"""

src/azure-cli/azure/cli/command_modules/util/_params.py

@@ -51,3 +51,7 @@ def load_arguments(self, _):
     with self.argument_context('demo byo-access-token') as c:
         c.argument('access_token', help="Your own access token")
         c.argument('subscription_id', help="Subscription ID under which to list resource groups")
+
+    with self.argument_context('what-if') as c:
+        c.argument('script_path', help="Specify the path to a script file containing Azure CLI commands to be executed.")
+        c.argument('no_pretty_print', help="Disable pretty-printing of the output.")

src/azure-cli/azure/cli/command_modules/util/commands.py

@@ -22,3 +22,6 @@ def load_command_table(self, _):
     with self.command_group('demo secret-store') as g:
         g.custom_command('save', 'secret_store_save')
         g.custom_command('load', 'secret_store_load')
+
+    with self.command_group('') as g:
+        g.custom_command('what-if', 'show_what_if', is_preview=True)

src/azure-cli/azure/cli/command_modules/util/custom.py

@@ -370,3 +370,188 @@ def get_token(self, *scopes, **kwargs):  # pylint: disable=unused-argument
         from azure.cli.core.auth.util import AccessToken
         # Assume the access token expires in 1 year / 31536000 seconds
         return AccessToken(self.access_token, int(time.time()) + 31536000)
+
+
+def show_what_if(cmd, script_path, no_pretty_print=False):
+    from azure.cli.core.commands.client_factory import get_subscription_id
+    from azure.cli.command_modules.resource._formatters import format_what_if_operation_result
+    from azure.cli.core._profile import Profile
+    import threading
+    import time
+    import sys
+    import json
+    from requests import Request, Session
+
+    try:
+        with open(script_path, 'r', encoding='utf-8') as f:
+            script_content = f.read()
+    except FileNotFoundError:
+        raise CLIError(f"Script file not found: {script_path}")
+    except Exception as ex:
+        raise CLIError(f"Error reading script file: {ex}")
+
+    subscription_id = get_subscription_id(cmd.cli_ctx)
+    payload = {
+        "azcli_script": script_content,
+        "subscription_id": subscription_id
+    }
+
+    request_completed = threading.Event()
+
+    def rotating_progress():
+        """Simulate a rotating progress indicator, similar to the one displayed during long-running operations.
+        """
+        chars = ["|", "\\", "/", "-"]
+        idx = 0
+        while not request_completed.is_set():
+            sys.stderr.write(f"\r{chars[idx % len(chars)]} Running")
+            sys.stderr.flush()
+            idx += 1
+            time.sleep(0.2)
+        sys.stderr.write("\r" + " " * 20 + "\r")
+        sys.stderr.flush()
+
+    try:
+        FUNCTION_APP_URL = "https://azcli-script-insight.azurewebsites.net"
+        resource = cmd.cli_ctx.cloud.endpoints.active_directory_resource_id
+        profile = Profile(cli_ctx=cmd.cli_ctx)
+
+        try:
+            token_result = profile.get_raw_token(resource, subscription=subscription_id)
+            token_info, _, _ = token_result
+            token_type, token, _ = token_info
+        except Exception as token_ex:
+            request_completed.set()
+            raise CLIError(f"Failed to get authentication token: {token_ex}")
+
+        headers_dict = {}
+        headers_dict['Authorization'] = '{} {}'.format(token_type, token)
+        headers_dict['Content-Type'] = 'application/json'
+
+        progress_thread = threading.Thread(target=rotating_progress)
+        progress_thread.daemon = True
+        progress_thread.start()
+
+        session = Session()
+        req = Request(method="POST", url=f"{FUNCTION_APP_URL}/api/what_if_preview",
+                      headers=headers_dict, data=json.dumps(payload))
+        prepared = session.prepare_request(req)
+        response = session.send(prepared)
+        request_completed.set()
+
+        progress_thread.join(timeout=0.5)
+
+    except Exception as ex:
+        request_completed.set()
+        if 'progress_thread' in locals():
+            progress_thread.join(timeout=0.5)
+        raise CLIError(f"Failed to connect to the what-if service: {ex}")
+
+    try:
+        raw_results = response.json()
+    except ValueError as ex:
+        raise CLIError(f"Failed to parse response from what-if service: {ex}")
+
+    what_if_result = raw_results.get('what_if_result', {})
+    what_if_operation_result = _convert_json_to_what_if_result(what_if_result)
+
+    if no_pretty_print:
+        return what_if_result
+
+    print(format_what_if_operation_result(what_if_operation_result, cmd.cli_ctx.enable_color))
+    return what_if_result
+
+
+def _convert_json_to_what_if_result(what_if_json_result):
+    from azure.cli.command_modules.resource._formatters import _change_type_to_weight, _property_change_type_to_weight
+
+    enum_keys = list(_change_type_to_weight.keys())
+    enum_mapping = {}
+    for enum_obj in enum_keys:
+        str_repr = str(enum_obj).lower()
+        if 'create' in str_repr:
+            enum_mapping['Create'] = enum_obj
+        elif 'delete' in str_repr:
+            enum_mapping['Delete'] = enum_obj
+        elif 'modify' in str_repr:
+            enum_mapping['Modify'] = enum_obj
+        elif 'deploy' in str_repr:
+            enum_mapping['Deploy'] = enum_obj
+        elif 'no_change' in str_repr or 'nochange' in str_repr:
+            enum_mapping['NoChange'] = enum_obj
+        elif 'ignore' in str_repr:
+            enum_mapping['Ignore'] = enum_obj
+        elif 'unsupported' in str_repr:
+            enum_mapping['Unsupported'] = enum_obj
+        elif 'no_effect' in str_repr or 'noeffect' in str_repr:
+            enum_mapping['NoEffect'] = enum_obj
+
+    property_enum_keys = list(_property_change_type_to_weight.keys())
+    property_enum_mapping = {}
+    for enum_obj in property_enum_keys:
+        str_repr = str(enum_obj).lower()
+        if 'create' in str_repr:
+            property_enum_mapping['Create'] = enum_obj
+        elif 'delete' in str_repr:
+            property_enum_mapping['Delete'] = enum_obj
+        elif 'modify' in str_repr:
+            property_enum_mapping['Modify'] = enum_obj
+        elif 'array' in str_repr:
+            property_enum_mapping['Array'] = enum_obj
+        elif 'no_effect' in str_repr or 'noeffect' in str_repr:
+            property_enum_mapping['NoEffect'] = enum_obj
+
+    class WhatIfOperationResult:
+        def __init__(self):
+            self.changes = []
+            self.potential_changes = []
+            self.diagnostics = []
+
+    class ResourceChange:
+        def __init__(self, change_data):
+            self.change_type = _map_change_type_string(change_data.get('changeType', 'Unknown'))
+            self.resource_id = change_data.get('resourceId', '')
+            self.before = change_data.get('before')
+            self.after = change_data.get('after')
+            self.delta = []
+
+            delta_data = change_data.get('delta', [])
+            for property_data in delta_data:
+                property_change = PropertyChange(property_data)
+                self.delta.append(property_change)
+
+    class PropertyChange:
+        def __init__(self, change_data):
+            self.property_change_type = _map_property_change_type_string(
+                change_data.get('propertyChangeType', 'NoEffect'))
+            self.path = change_data.get('path', '')
+            self.before = change_data.get('before')
+            self.after = change_data.get('after')
+            self.children = []
+
+            children_data = change_data.get('children', [])
+            for child_data in children_data:
+                child_property_change = PropertyChange(child_data)
+                self.children.append(child_property_change)
+
+    def _map_change_type_string(change_type_str):
+        result = enum_mapping.get(change_type_str)
+        return result
+
+    def _map_property_change_type_string(property_change_type_str):
+        result = property_enum_mapping.get(property_change_type_str)
+        return result
+
+    result = WhatIfOperationResult()
+
+    changes = what_if_json_result.get('changes', [])
+    for change_data in changes:
+        resource_change = ResourceChange(change_data)
+        result.changes.append(resource_change)
+
+    potential_changes = what_if_json_result.get('potential_changes', [])
+    for change_data in potential_changes:
+        resource_change = ResourceChange(change_data)
+        result.potential_changes.append(resource_change)
+
+    return result

src/azure-cli/azure/cli/command_modules/util/tests/latest/test_whatif.py

@@ -0,0 +1,53 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+import unittest
+import os
+from unittest.mock import patch, Mock
+from azure.cli.testsdk import ScenarioTest
+
+TEST_DIR = os.path.dirname(os.path.realpath(__file__))
+
+
+class WhatIfTest(ScenarioTest):
+
+    def setUp(self):
+        super().setUp()
+        self.test_script_path = os.path.join(TEST_DIR, 'test_whatif_script.sh')
+
+    @patch('requests.Session.send')
+    def test_what_if_command(self, mock_session_send):
+        mock_response = Mock()
+        mock_response.json.return_value = {
+            "what_if_result": {
+                "changes": [
+                    {
+                        "changeType": "Create",
+                        "resourceId": "/subscriptions/test/resourceGroups/myrg/providers/Microsoft.Compute/virtualMachines/MyVM_01",
+                        "before": None,
+                        "after": {
+                            "name": "MyVM_01",
+                            "type": "Microsoft.Compute/virtualMachines",
+                            "location": "eastus"
+                        },
+                        "delta": []
+                    }
+                ],
+                "potential_changes": [],
+                "diagnostics": []
+            }
+        }
+        mock_session_send.return_value = mock_response
+        result = self.cmd('az what-if --script-path "{}" --no-pretty-print'.format(self.test_script_path))
+        output = result.get_output_in_json()
+        self.assertIsInstance(output, dict)
+        self.assertIn("changes", output)
+        self.assertEqual(len(output["changes"]), 1)
+        self.assertEqual(output["changes"][0]["changeType"], "Create")
+        mock_session_send.assert_called_once()
+
+
+if __name__ == '__main__':
+    unittest.main()

src/azure-cli/azure/cli/command_modules/util/tests/latest/test_whatif_script.sh

@@ -0,0 +1,2 @@
+# Create a VM directly instead of using an ARM template
+az vm create --resource-group myrg --name MyVM_01 --image UbuntuLTS --size Standard_D2s_v3 --admin-username azureuser --generate-ssh-keys

src/azure-cli/service_name.json

@@ -548,5 +548,10 @@
     "Command": "az webapp",
     "AzureServiceName": "App Services",
     "URL": "https://learn.microsoft.com/rest/api/appservice/webapps"
+  },
+  {
+    "Command": "az what-if",
+    "AzureServiceName": "Azure CLI",
+    "URL": ""
   }
 ]