vm create: fix invalid detection logics on vhd blobs (Azure#6229)

src/command_modules/azure-cli-vm/HISTORY.rst

@@ -4,6 +4,7 @@ Release History
 ===============
 2.0.31
 ++++++
+* vm: fix an invalid detection logic on unmanaged blob uri
 * vm: support disk encryption w/o user provided service principals 
 * BREAKING CHANGE: do not use VM 'ManagedIdentityExtension' for MSI support
 * vmss: support eviction policy

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

@@ -6,6 +6,10 @@
 # pylint:disable=too-many-lines
 
 import os
+try:
+    from urllib.parse import urlparse
+except ImportError:
+    from urlparse import urlparse  # pylint: disable=import-error
 
 from knack.log import get_logger
 from knack.util import CLIError
@@ -198,15 +202,11 @@ def _parse_image_argument(cmd, namespace):
     from msrestazure.azure_exceptions import CloudError
     import re
 
-    # 1 - easy check for URI
-    if namespace.image.lower().endswith('.vhd'):
-        return 'uri'
-
-    # 2 - check if a fully-qualified ID (assumes it is an image ID)
+    # 1 - check if a fully-qualified ID (assumes it is an image ID)
     if is_valid_resource_id(namespace.image):
         return 'image_id'
 
-    # 3 - attempt to match an URN pattern
+    # 2 - attempt to match an URN pattern
     urn_match = re.match('([^:]*):([^:]*):([^:]*):([^:]*)', namespace.image)
     if urn_match:
         namespace.os_publisher = urn_match.group(1)
@@ -223,6 +223,10 @@ def _parse_image_argument(cmd, namespace):
 
         return 'urn'
 
+    # 3 - unmanaged vhd based images?
+    if urlparse(namespace.image).scheme:
+        return 'uri'
+
     # 4 - attempt to match an URN alias (most likely)
     from azure.cli.command_modules.vm._actions import load_images_from_aliases_doc
     images = load_images_from_aliases_doc(cmd.cli_ctx)
@@ -1145,15 +1149,10 @@ def process_image_create_namespace(cmd, namespace):
 
 def _figure_out_storage_source(cli_ctx, resource_group_name, source):
     from msrestazure.azure_exceptions import CloudError
-    try:
-        from urllib.parse import urlparse
-    except ImportError:
-        from urlparse import urlparse  # pylint: disable=import-error
-
     source_blob_uri = None
     source_disk = None
     source_snapshot = None
-    if urlparse(source).scheme:  # an exiting scheme means a blob uri, as ':' isn't allowed in a disk name
+    if urlparse(source).scheme:  # a uri?
         source_blob_uri = source
     elif '/disks/' in source.lower():
         source_disk = source

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

@@ -76,7 +76,7 @@ def test_figure_out_storage_source(self):
         self.assertEqual(src_blob_uri, test_data)
 
         test_data = '/subscriptions/0b1f6471-1bf0-4dda-aec3-cb9272f09590/resourceGroups/JAVACSMRG6017/providers/Microsoft.Compute/disks/ex.vhd'
-        src_blob_uri, src_disk, src_snapshot = _figure_out_storage_source(TestCli(), 'tg1', test_data)
+        src_blob_uri, src_disk, src_snapshot = _figure_out_storage_source(None, 'tg1', test_data)
         self.assertEqual(src_disk, test_data)
         self.assertFalse(src_snapshot)
         self.assertFalse(src_blob_uri)
@@ -214,6 +214,21 @@ def test_parse_staging_image_argument(self, logger_mock, client_factory_mock):
                                        "Configuring plan settings will be skipped", 'publisher1:offer1:sku1:1.0.0',
                                        'image not found')
 
+    def test_parse_unmanaged_image_argument(self):
+        np = mock.MagicMock()
+        np.image = 'https://foo.blob.core.windows.net/vhds/1'
+        cmd = mock.MagicMock()
+        # action & assert
+        self.assertEqual(_parse_image_argument(cmd, np), 'uri')
+
+    def test_parse_managed_image_argument(self):
+        np = mock.MagicMock()
+        np.image = '/subscriptions/123/resourceGroups/foo/providers/Microsoft.Compute/images/nixos-imag.vhd'
+        cmd = mock.MagicMock()
+
+        # action & assert
+        self.assertEqual(_parse_image_argument(cmd, np), 'image_id')
+
     def test_get_next_subnet_addr_suffix(self):
         result = _get_next_subnet_addr_suffix('10.0.0.0/16', '10.0.0.0/24', 24)
         self.assertEqual(result, '10.0.1.0/24')