[Draft] Support --what-if argument

src/azure-cli-core/azure/cli/core/commands/__init__.py

@@ -506,6 +506,8 @@ class AzCliCommandInvoker(CommandInvoker):
 
     # pylint: disable=too-many-statements,too-many-locals,too-many-branches
     def execute(self, args):
+        args_copy = args[:]
+
         from knack.events import (EVENT_INVOKER_PRE_CMD_TBL_CREATE, EVENT_INVOKER_POST_CMD_TBL_CREATE,
                                   EVENT_INVOKER_CMD_TBL_LOADED, EVENT_INVOKER_PRE_PARSE_ARGS,
                                   EVENT_INVOKER_POST_PARSE_ARGS,
@@ -586,7 +588,8 @@ def execute(self, args):
             args[0] = '--help'
 
         self.parser.enable_autocomplete()
-
+        if '--what-if' in (args_copy):
+            return self._what_if(args_copy)
         self.cli_ctx.raise_event(EVENT_INVOKER_PRE_PARSE_ARGS, args=args)
         parsed_args = self.parser.parse_args(args)
         self.cli_ctx.raise_event(EVENT_INVOKER_POST_PARSE_ARGS, command=parsed_args.command, args=parsed_args)
@@ -691,6 +694,44 @@ def execute(self, args):
             table_transformer=self.commands_loader.command_table[parsed_args.command].table_transformer,
             is_query_active=self.data['query_active'])
 
+    def _what_if(self, args):
+        # DEBUG: Add logging to see if this method is called
+        print(f"DEBUG: _what_if called with command: {args}")
+        if '--what-if' in args:
+            print("DEBUG: Entering what-if mode")
+            from azure.cli.core.what_if import what_if_preview
+            try:
+                # Get subscription ID with priority: --subscription parameter > current login subscription
+                if '--subscription' in args:
+                    index = args.index('--subscription')
+                    if index + 1 < len(args):
+                        subscription_value = args[index + 1]
+                        subscription_id = subscription_value    
+                else:
+                    # Fallback to current login subscription TODO
+                    subscription_id = self.cli_ctx.data.get("subscription_id", "6b085460-5f21-477e-ba44-1035046e9101")
+
+                args = ["az"] + args if args[0] != 'az' else args
+                command = " ".join(args)
+                what_if_result = what_if_preview(command, subscription_id=subscription_id)
+
+                # Ensure output format is set for proper formatting
+                # Default to 'json' if not already set
+                if 'output' not in self.cli_ctx.invocation.data or self.cli_ctx.invocation.data['output'] is None:
+                    self.cli_ctx.invocation.data['output'] = 'json'
+
+                # Return the formatted what-if output as the result
+                # Similar to the normal flow in execute() method
+                return CommandResultItem(
+                    what_if_result, 
+                    table_transformer=None,
+                    is_query_active=self.data.get('query_active', False),
+                    exit_code=0
+                )
+            except Exception as ex:
+                # If what-if service fails, still show an informative message
+                return CommandResultItem(None, exit_code=1, error=CLIError('What-if preview failed: {str(ex)}\nNote: This was a preview operation. No actual changes were made.'))
+
     @staticmethod
     def _extract_parameter_names(args):
         # note: name start with more than 2 '-' will be treated as value e.g. certs in PEM format

src/azure-cli-core/azure/cli/core/commands/parameters.py

@@ -268,6 +268,15 @@ def get_location_type(cli_ctx):
     return location_type
 
 
+def get_what_if_type():
+    what_if_type = CLIArgumentType(
+        options_list=['--what-if'],
+        help="Preview the changes that will be made without actually executing the command. "
+             "This will call the what-if service to compare the current state with the expected state after execution."
+    )
+    return what_if_type
+
+
 deployment_name_type = CLIArgumentType(
     help=argparse.SUPPRESS,
     required=False,

src/azure-cli-core/azure/cli/core/what_if.py

@@ -0,0 +1,109 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+"""
+Module for handling what-if functionality in Azure CLI.
+This module provides the core logic for preview mode execution without actually running commands.
+
+IMPORTANT: The what-if service requires client-side authentication to operate under the 
+caller's subscription and permissions. Server-side authentication is not supported for 
+what-if operations as it would not provide access to the caller's subscription.
+
+This client now uses AzureCliCredential to obtain an access token for the caller's subscription.
+
+The what-if service will use your configured credentials to access your subscription
+and preview deployment changes under your permissions.
+"""
+
+import requests
+from typing import Dict, Any, Optional
+from azure.identity import AzureCliCredential
+from datetime import datetime, timezone
+from knack.log import get_logger
+
+logger = get_logger(__name__)
+
+# Configuration
+FUNCTION_APP_URL = "https://azcli-script-insight.azurewebsites.net"
+
+
+def get_azure_cli_access_token() -> Optional[str]:
+    """
+    Get access token for the caller's subscription using AzureCliCredential
+
+    Returns:
+        Access token string if successful, None if failed
+    """
+    token_info = get_azure_cli_token_info()
+    return token_info.get("accessToken") if token_info else None
+
+
+def get_azure_cli_token_info() -> Optional[Dict[str, Any]]:
+    """
+    Get complete token information using AzureCliCredential including expiration
+
+    Returns:
+        Dictionary with token info including accessToken, expiresOn, etc., or None if failed
+    """
+    try:
+        # Use AzureCliCredential for Azure CLI authentication
+        cli_credential = AzureCliCredential(process_timeout=30)
+
+        # Get access token for Azure Resource Manager
+        token = cli_credential.get_token("https://management.azure.com/.default")
+
+        token_info = {
+            "accessToken": token.token,
+            "expiresOn": datetime.fromtimestamp(token.expires_on, tz=timezone.utc).isoformat(),
+            "tokenType": "Bearer"
+        }
+
+        return token_info
+
+    except Exception as e:
+        logger.warning(f"Error getting access token with AzureCliCredential: {str(e)}")
+        return None
+
+
+def what_if_preview(azcli_script: str, subscription_id: Optional[str] = None) -> Dict[str, Any]:
+    """
+    Preview deployment changes using Azure what-if functionality
+
+    Args:
+        function_app_url: Base URL of your Azure Function App
+        azcli_script: Azure CLI script to analyze
+        subscription_id: Optional fallback subscription ID if not in script
+
+    Returns:
+        Dictionary with what-if preview result
+    """
+    url = f"{FUNCTION_APP_URL.rstrip('/')}/api/what_if_preview"
+
+    headers = {
+        'Content-Type': 'application/json',
+        'Accept': 'application/json'
+    }
+
+    # Get access token from Azure CLI
+    access_token = get_azure_cli_access_token()
+    if not access_token:
+        return {
+            "error": "Failed to get access token from Azure CLI. Please ensure you are logged in with 'az login'",
+            "details": "The what-if service requires client credentials to access your subscription. Please provide an access token.",
+            "success": False
+        }
+
+    # Use Authorization header for access token
+    headers['Authorization'] = f'Bearer {access_token}'
+
+    payload = {"azcli_script": azcli_script}
+    if subscription_id:
+        payload["subscription_id"] = subscription_id
+
+    try:
+        response = requests.post(url, json=payload, headers=headers, timeout=300)
+        return response.json()
+    except requests.RequestException as e:
+        raise e

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

@@ -40,6 +40,7 @@
     get_enum_type,
     get_resource_name_completion_list,
     get_location_type,
+    get_what_if_type,
     tags_type,
     resource_group_name_type
 )
@@ -1915,6 +1916,7 @@ def _configure_security_policy_storage_params(arg_ctx):
     with self.argument_context('sql server create') as c:
         c.argument('location',
                    arg_type=get_location_type_with_default_from_resource_group(self.cli_ctx))
+        c.argument('what_if', get_what_if_type(), help='Preview the changes that will be made without actually executing the command.')
 
         # Create args that will be used to build up the Server object
         create_args_for_complex_type(

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

@@ -4369,6 +4369,7 @@ def server_create(
         external_admin_principal_type=None,
         external_admin_sid=None,
         external_admin_name=None,
+        what_if=None,
         **kwargs):
     '''
     Creates a server.

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

@@ -13,7 +13,7 @@
 from azure.cli.core.commands.validators import (
     get_default_location_from_resource_group, validate_file_or_dict)
 from azure.cli.core.commands.parameters import (
-    get_location_type, get_resource_name_completion_list, tags_type, get_three_state_flag,
+    get_location_type, get_what_if_type, get_resource_name_completion_list, tags_type, get_three_state_flag,
     file_type, get_enum_type, zone_type, zones_type)
 from azure.cli.command_modules.vm._actions import _resource_not_exists
 from azure.cli.command_modules.vm._completers import (
@@ -413,6 +413,7 @@ def load_arguments(self, _):
         c.argument('workspace', is_preview=True, arg_group='Monitor', help='Name or ID of Log Analytics Workspace. If you specify the workspace through its name, the workspace should be in the same resource group with the vm, otherwise a new workspace will be created.')
 
     with self.argument_context('vm update') as c:
+        c.argument('what_if', get_what_if_type(), help='Preview the changes that will be made without actually executing the command.')
         c.argument('os_disk', min_api='2017-12-01', help="Managed OS disk ID or name to swap to")
         c.argument('write_accelerator', nargs='*', min_api='2017-12-01',
                    help="enable/disable disk write accelerator. Use singular value 'true/false' to apply across, or specify individual disks, e.g.'os=true 1=true 2=true' for os disk and data disks with lun of 1 & 2")
@@ -1062,6 +1063,7 @@ def load_arguments(self, _):
     for scope in ['vm create', 'vmss create']:
         with self.argument_context(scope) as c:
             c.argument('location', get_location_type(self.cli_ctx), help='Location in which to create VM and related resources. If default location is not configured, will default to the resource group\'s location')
+            c.argument('what_if', get_what_if_type(), help='Preview the changes that will be made without actually executing the command.')
             c.argument('tags', tags_type)
             c.argument('no_wait', help='Do not wait for the long-running operation to finish.')
             c.argument('validate', options_list=['--validate'], help='Generate and validate the ARM template without creating any resources.', action='store_true')

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

@@ -850,7 +850,7 @@ def create_vm(cmd, vm_name, resource_group_name, image=None, size='Standard_DS1_
               enable_user_redeploy_scheduled_events=None, zone_placement_policy=None, include_zones=None,
               exclude_zones=None, align_regional_disks_to_vm_zone=None, wire_server_mode=None, imds_mode=None,
               wire_server_access_control_profile_reference_id=None, imds_access_control_profile_reference_id=None,
-              key_incarnation_id=None):
+              key_incarnation_id=None, what_if=False):
 
     from azure.cli.core.commands.client_factory import get_subscription_id
     from azure.cli.core.util import random_string, hash_string