minor fix

wangzelin007 · wangzelin007 · commit 2b61732dc167 · 2025-09-22T14:35:04.000+08:00
diff --git a/src/azure-cli-core/azure/cli/core/commands/__init__.py b/src/azure-cli-core/azure/cli/core/commands/__init__.py
@@ -699,7 +699,7 @@ def _what_if(self, args):
         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
+            from azure.cli.core.what_if import show_what_if
             try:
                 # Get subscription ID with priority: --subscription parameter > current login subscription
                 if '--subscription' in args:
@@ -708,12 +708,13 @@ def _what_if(self, 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")
+                    from azure.cli.core.commands.client_factory import get_subscription_id
+                    subscription_id = get_subscription_id(self.cli_ctx)
+                    print(f"DEBUG: Using current login subscription ID: {subscription_id}")
                 
                 args = ["az"] + args if args[0] != 'az' else args
                 command = " ".join(args)
-                what_if_result = what_if_preview(command, subscription_id=subscription_id)
+                what_if_result = show_what_if(self.cli_ctx, command, subscription_id=subscription_id)
 
                 # Ensure output format is set for proper formatting
                 # Default to 'json' if not already set
diff --git a/src/azure-cli-core/azure/cli/core/commands/parameters.py b/src/azure-cli-core/azure/cli/core/commands/parameters.py
@@ -272,7 +272,8 @@ 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."
+             "This will call the what-if service to compare the current state with the expected state after execution.",
+        is_preview=True
     )
     return what_if_type
 
diff --git a/src/azure-cli-core/azure/cli/core/what_if.py b/src/azure-cli-core/azure/cli/core/what_if.py
@@ -16,94 +16,186 @@
 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
+def show_what_if(cli_ctx, azcli_script: str, subscription_id: Optional[str] = None, no_pretty_print: bool = False):
+    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
 
-    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
+    payload = {
+        "azcli_script": azcli_script,
+        "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()
 
-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
+        FUNCTION_APP_URL = "https://azcli-script-insight.azurewebsites.net"
+        resource = cli_ctx.cloud.endpoints.active_directory_resource_id
+        profile = Profile(cli_ctx=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}")
 
-
-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
+        raw_results = response.json()
+    except ValueError as ex:
+        raise CLIError(f"Failed to parse response from what-if service: {ex}")
+
+    success = raw_results.get('success')
+    if success is False:
+        return raw_results
+    elif success is True:
+        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, cli_ctx.enable_color))
+        return what_if_result
+    else:
+        raise CLIError(f"Unexpected response from what-if service, got: {raw_results}")
+
+
+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
diff --git a/src/azure-cli/azure/cli/command_modules/sql/_params.py b/src/azure-cli/azure/cli/command_modules/sql/_params.py
@@ -1916,7 +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.')
+        c.argument('what_if', get_what_if_type())
 
         # Create args that will be used to build up the Server object
         create_args_for_complex_type(
diff --git a/src/azure-cli/azure/cli/command_modules/vm/_params.py b/src/azure-cli/azure/cli/command_modules/vm/_params.py
@@ -413,7 +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('what_if', get_what_if_type())
         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")
@@ -1063,7 +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('what_if', get_what_if_type())
             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')

Original file line number	Diff line number	Diff line change
`@@ -272,7 +272,8 @@ def get_what_if_type():`
`272`	`272`	`what_if_type = CLIArgumentType(`
`273`	`273`	`options_list=['--what-if'],`
`274`	`274`	`help="Preview the changes that will be made without actually executing the command. "`
`275`		`- "This will call the what-if service to compare the current state with the expected state after execution."`
	`275`	`+ "This will call the what-if service to compare the current state with the expected state after execution.",`
	`276`	`+ is_preview=True`
`276`	`277`	`)`
`277`	`278`	`return what_if_type`
`278`	`279`