[Compute] Fix #34083: az vm create: Preserve ARM deployment errors when HTTP response is consumed

Root cause: src/azure-cli/azure/cli/command_modules/vm/tests/latest/test_vm_create_errors.py:VMCreateErrorsScenarioTest.test_vm_create_windows_os_disk_too_small
requested Standard_D2s_v3 in westus2. The supplied completed live run
establishes that this SKU is unavailable to the test subscription, so
SkuNotAvailable prevents reaching the intended InvalidParameter
disk-size error. Earlier attempts selected a commonly used SKU; this
correction instead uses the review's subscription-specific evidence that
Standard_D2s_v5 is allowed in westus2. No additional handler or
authentication defect is established.

Change: Changed only the negative Windows VM scenario's size literal
from Standard_D2s_v3 to Standard_D2s_v5. Retained westus2, the 1-GB OS
disk and every existing assertion. The success scenario's
Standard_D2s_v3 in eastus, production code, authentication and
recordings remain unchanged. Restored report after checkpoint omitted
disposable report artifacts; no additional source changes made.

Alternatives considered:
- Repeat Standard_D2s_v3: rejected because completed live evidence
  establishes that it is unavailable to this subscription.
- Change region, other scenario sizes, disk size or assertions: rejected
  because the review requires preserving those exactly.

Verified:
- Trusted validation passed (1 command).
- New or changed tests fail on the base revision and pass with the
  change:
  `src/azure-cli-core/azure/cli/core/commands/tests/test_arm.py::TestTemplateBasedException::test_template_exception_consumed_invalid_template`,
  `src/azure-cli-core/azure/cli/core/commands/tests/test_arm.py::TestTemplateBasedException::test_template_exception_consumed_message_without_parsed_error`,
  `src/azure-cli-core/azure/cli/core/commands/tests/test_arm.py::TestTemplateBasedException::test_template_exception_consumed_nested_error`.
- Unit tests: 12 executed, 0 skipped.
- flake8 and pylint passed.
- Pre-publication gates passed: release-artifact, generated-ownership,
  command-help, suppression, independent-review.
- Scenario recordings must be refreshed by a live test run.
- Repository CI and live tests run on this PR.

Author: a0x1ab

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

@@ -105,9 +105,14 @@ def raise_subdivision_deployment_error(error_message, error_code=None):
 
 
 def handle_template_based_exception(ex):
+    from azure.core.exceptions import HttpResponseError
+
     try:
         raise CLIError(ex.inner_exception.error.message)
     except AttributeError:
+        if isinstance(ex, HttpResponseError):
+            # The SDK message includes parsed error details; the response stream may be consumed.
+            raise_subdivision_deployment_error(ex.message, ex.error.code if ex.error else None)
         if hasattr(ex, 'response'):
             raise_subdivision_deployment_error(ex.response.internal_response.text, ex.error.code if ex.error else None)
         else:

src/azure-cli-core/azure/cli/core/commands/tests/test_arm.py

@@ -0,0 +1,175 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+import json
+import unittest
+from types import SimpleNamespace
+from unittest import mock
+
+from azure.core.exceptions import HttpResponseError
+from azure.core.pipeline.transport import HttpRequest, RequestsTransportResponse
+from knack.util import CLIError
+from requests import Response
+
+from azure.cli.core.azclierror import DeploymentError, InvalidTemplateError
+from azure.cli.core.commands.arm import handle_template_based_exception
+
+
+class TestTemplateBasedException(unittest.TestCase):
+
+    @staticmethod
+    def _sdk_error(error=None, message=None, consumed=True):
+        response = Response()
+        response.status_code = 400
+        response.reason = 'Bad Request'
+        response.headers['Content-Type'] = 'application/json'
+        response._content = json.dumps({'error': error} if error else {}).encode('utf-8')
+        transport_response = RequestsTransportResponse(
+            HttpRequest('PUT', 'https://management.azure.com/deployments/test'), response)
+        exception = HttpResponseError(response=transport_response, message=message)
+        if consumed:
+            # Match a streamed requests response whose body was read without caching it.
+            response._content = False
+            response._content_consumed = True
+        return exception
+
+    def test_template_exception_consumed_sku_error(self):
+        message = (
+            "The requested VM size Standard_B2s is currently not available in location eastus. "
+            "Please try another size or deploy to a different location or different zone.")
+        exception = self._sdk_error({'code': 'SkuNotAvailable', 'message': message})
+        self.assertIsNone(exception.inner_exception)
+        with self.assertRaisesRegex(RuntimeError, 'already consumed'):
+            _ = exception.response.internal_response.text
+
+        with self.assertRaises(DeploymentError) as raised:
+            handle_template_based_exception(exception)
+        self.assertEqual(str(raised.exception), exception.message)
+        self.assertIn('SkuNotAvailable', str(raised.exception))
+        self.assertIn(message, str(raised.exception))
+
+    def test_template_exception_consumed_sku_errors_across_regions_and_sizes(self):
+        for region in ('eastus', 'westeurope', 'northeurope'):
+            for size in ('Standard_B2s', 'Standard_B2ms', 'Standard_D2s_v3',
+                         'Standard_D2s_v5', 'Standard_B1s'):
+                with self.subTest(region=region, size=size):
+                    message = (
+                        'The requested VM size {} is currently not available in location {}. '
+                        'Please try another size or deploy to a different location or different zone.'
+                    ).format(size, region)
+                    exception = self._sdk_error({'code': 'SkuNotAvailable', 'message': message})
+                    self.assertIsNone(exception.inner_exception)
+
+                    with self.assertRaises(DeploymentError) as raised:
+                        handle_template_based_exception(exception)
+                    self.assertEqual(str(raised.exception), exception.message)
+                    self.assertIn('SkuNotAvailable', str(raised.exception))
+                    self.assertIn(message, str(raised.exception))
+
+    def test_template_exception_consumed_quota_error(self):
+        message = (
+            'Operation could not be completed as it results in exceeding approved '
+            'standardDSv5Family Cores quota. Location: eastus, Current Limit: 0, Current Usage: 0, '
+            'Additional Required: 2. Please request a quota increase.')
+        exception = self._sdk_error({'code': 'QuotaExceeded', 'message': message})
+
+        with self.assertRaises(DeploymentError) as raised:
+            handle_template_based_exception(exception)
+        self.assertEqual(str(raised.exception), exception.message)
+        self.assertIn('QuotaExceeded', str(raised.exception))
+        self.assertIn(message, str(raised.exception))
+
+    def test_template_exception_consumed_nested_error(self):
+        message = 'At least one resource deployment operation failed.'
+        detail_message = 'Allocation failed. Please retry later or use a different VM size.'
+        exception = self._sdk_error({
+            'code': 'DeploymentFailed',
+            'message': message,
+            'details': [{
+                'code': 'ResourceDeploymentFailure',
+                'message': 'The resource write operation failed.',
+                'details': [{'code': 'AllocationFailed', 'message': detail_message}]
+            }]
+        })
+
+        with self.assertRaises(DeploymentError) as raised:
+            handle_template_based_exception(exception)
+        self.assertEqual(str(raised.exception), exception.message)
+        for expected in ('DeploymentFailed', message, 'ResourceDeploymentFailure',
+                         'The resource write operation failed.', 'AllocationFailed', detail_message):
+            self.assertIn(expected, str(raised.exception))
+
+    def test_template_exception_consumed_invalid_template(self):
+        message = 'The template deployment is not valid.'
+        exception = self._sdk_error({'code': 'InvalidTemplateDeployment', 'message': message})
+
+        with self.assertRaises(InvalidTemplateError) as raised:
+            handle_template_based_exception(exception)
+        self.assertIn('InvalidTemplateDeployment', str(raised.exception))
+        self.assertIn(message, str(raised.exception))
+
+    def test_template_exception_consumed_message_without_parsed_error(self):
+        message = '(QuotaExceeded) Current Limit: 0, Current Usage: 0, Additional Required: 2.'
+        exception = self._sdk_error(message=message)
+        self.assertIsNone(exception.error)
+
+        with self.assertRaises(DeploymentError) as raised:
+            handle_template_based_exception(exception)
+        self.assertEqual(str(raised.exception), message)
+
+    def test_template_exception_sdk_error_without_response(self):
+        message = 'Deployment failed before a response was available.'
+        exception = HttpResponseError(message=message)
+
+        with self.assertRaises(DeploymentError) as raised:
+            handle_template_based_exception(exception)
+        self.assertEqual(str(raised.exception), message)
+
+    def test_template_exception_sdk_error_does_not_reread_response(self):
+        exception = self._sdk_error({'code': 'SkuNotAvailable', 'message': 'Try another VM size.'})
+        with mock.patch.object(exception.response, 'text', side_effect=AssertionError('Body reread')) as text:
+            with self.assertRaises(DeploymentError) as raised:
+                handle_template_based_exception(exception)
+        text.assert_not_called()
+        self.assertEqual(str(raised.exception), exception.message)
+
+    def test_template_exception_unconsumed_error(self):
+        message = 'The deployment conflicts with an existing operation.'
+        exception = self._sdk_error({'code': 'Conflict', 'message': message}, consumed=False)
+
+        with self.assertRaises(DeploymentError) as raised:
+            handle_template_based_exception(exception)
+        self.assertIn('Conflict', str(raised.exception))
+        self.assertIn(message, str(raised.exception))
+
+    def test_template_exception_inner_error_preserved(self):
+        exception = self._sdk_error({'code': 'DeploymentFailed', 'message': 'Outer failure.'})
+        exception.inner_exception = SimpleNamespace(error=SimpleNamespace(message='Inner failure.'))
+
+        with self.assertRaises(CLIError) as raised:
+            handle_template_based_exception(exception)
+        self.assertIs(type(raised.exception), CLIError)
+        self.assertEqual(str(raised.exception), 'Inner failure.')
+
+    def test_template_exception_legacy_response_preserved(self):
+        for code, error_type in [('DeploymentFailed', DeploymentError),
+                                 ('InvalidTemplateDeployment', InvalidTemplateError)]:
+            with self.subTest(code=code):
+                exception = SimpleNamespace(
+                    inner_exception=None,
+                    error=SimpleNamespace(code=code),
+                    response=SimpleNamespace(internal_response=SimpleNamespace(text='Original response body.')))
+
+                with self.assertRaises(error_type) as raised:
+                    handle_template_based_exception(exception)
+                self.assertEqual(str(raised.exception), 'Original response body.')
+
+    def test_template_exception_unrelated_error_preserved(self):
+        exception = ValueError('Unrelated VM error.')
+
+        with self.assertRaises(CLIError) as raised:
+            handle_template_based_exception(exception)
+        self.assertIs(type(raised.exception), CLIError)
+        self.assertEqual(str(raised.exception), 'Unrelated VM error.')

src/azure-cli/azure/cli/command_modules/vm/tests/latest/test_vm_create_errors.py

@@ -0,0 +1,100 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+from azure.cli.core.azclierror import logger as error_logger
+from azure.cli.testsdk import ResourceGroupPreparer, ScenarioTest
+from azure.cli.testsdk.patches import patch_main_exception_handler
+
+
+class VMCreateErrorsScenarioTest(ScenarioTest):
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        # Keep the real CLI error renderer: the default harness patch only raises an exception.
+        self.recording_patches.remove(patch_main_exception_handler)
+        self.replay_patches.remove(patch_main_exception_handler)
+
+    @ResourceGroupPreparer(name_prefix='cli_test_vm_create_success_', location='eastus')
+    def test_vm_create_windows_standard_ssd_no_public_ip(self, resource_group):
+        self.kwargs.update({
+            'vm': 'windowsvm',
+            'admin_username': 'clitestuser',
+            'admin_password': 'TestPassword1234!',
+        })
+
+        result = self.cmd(
+            'vm create -g {rg} -n {vm} --location eastus '
+            '--image Win2022Datacenter --size Standard_D2s_v3 '
+            '--storage-sku StandardSSD_LRS --public-ip-address "" '
+            '--admin-username {admin_username} --admin-password {admin_password}',
+            checks=[
+                self.exists('id'),
+                self.check('resourceGroup', '{rg}'),
+                self.check('location', 'eastus'),
+                self.check('powerState', 'VM running'),
+                self.check('publicIpAddress', ''),
+                self.exists('privateIpAddress'),
+            ])
+        self.assertEqual(result.exit_code, 0)
+        self.assertTrue(result.output.strip())
+        created_vm = result.get_output_in_json()
+
+        vm = self.cmd('vm show -g {rg} -n {vm}', checks=[
+            self.check('id', created_vm['id']),
+            self.check('provisioningState', 'Succeeded'),
+            self.check('hardwareProfile.vmSize', 'Standard_D2s_v3'),
+            self.check('storageProfile.imageReference.publisher', 'MicrosoftWindowsServer'),
+            self.check('storageProfile.imageReference.offer', 'WindowsServer'),
+            self.check('storageProfile.imageReference.sku', '2022-datacenter-g2'),
+            self.exists('storageProfile.imageReference.version'),
+            self.check('storageProfile.osDisk.osType', 'Windows'),
+            self.check('storageProfile.osDisk.managedDisk.storageAccountType', 'StandardSSD_LRS'),
+        ]).get_output_in_json()
+
+        self.cmd('vm get-instance-view -g {rg} -n {vm}', checks=[
+            self.check("instanceView.statuses[?code=='ProvisioningState/succeeded'] | length(@)", 1),
+            self.check("instanceView.statuses[?code=='PowerState/running'] | length(@)", 1),
+        ])
+
+        nics = vm['networkProfile']['networkInterfaces']
+        self.assertTrue(nics)
+        for nic in nics:
+            self.kwargs['nic_id'] = nic['id']
+            attached_nic = self.cmd('network nic show --ids {nic_id}').get_output_in_json()
+            ip_configurations = attached_nic['ipConfigurations']
+            self.assertTrue(ip_configurations)
+            for ip_configuration in ip_configurations:
+                self.assertFalse(ip_configuration.get('publicIpAddress'))
+
+    @ResourceGroupPreparer(name_prefix='cli_test_vm_create_disk_error_', location='westus2')
+    def test_vm_create_windows_os_disk_too_small(self, resource_group):
+        self.kwargs.update({
+            'vm': 'invaliddiskvm',
+            'admin_username': 'clitestuser',
+            'admin_password': 'TestPassword1234!',
+        })
+
+        # This size is passed into the deployment template; Compute rejects shrinking the image disk.
+        with self.assertLogs(error_logger, level='WARNING') as logs:
+            result = self.cmd(
+                'vm create -g {rg} -n {vm} --location westus2 '
+                '--image Win2022Datacenter --size Standard_D2s_v5 '
+                '--storage-sku StandardSSD_LRS --public-ip-address "" '
+                '--admin-username {admin_username} --admin-password {admin_password} '
+                '--os-disk-size-gb 1',
+                expect_failure=True)
+
+        self.assertNotEqual(result.exit_code, 0)
+        self.assertEqual(result.output.strip(), '')
+        self.assertIsNone(self.cli_ctx.result.result)
+        self.assertIsNotNone(self.cli_ctx.result.error)
+        errors = '\n'.join(logs.output)
+        self.assertIn('InvalidParameter', errors)
+        self.assertRegex(errors.lower(), r'disk size\s+1\b')
+        self.assertIn('smaller than', errors.lower())
+        self.assertIn('image', errors.lower())
+        for internal_error in ('Traceback', 'RuntimeError', 'already consumed', 'NoneType'):
+            self.assertNotIn(internal_error, errors)
+            self.assertNotIn(internal_error, str(self.cli_ctx.result.error))