Merge pull request #107 from release-engineering/simplify_sas

Azure: Allow partial SAS comparison

Author: JAVGan

cloudpub/ms_azure/service.py

@@ -620,7 +620,7 @@ def publish(self, metadata: AzurePublishingMetadata) -> None:
             tech_config.disk_versions = [disk_version]
 
         # We just want to append a new image if the SAS is not already present.
-        elif not is_sas_present(tech_config, metadata.image_path):
+        elif not is_sas_present(tech_config, metadata.image_path, metadata.check_base_sas_only):
             # Here we can have the metadata.disk_version set or empty.
             # When set we want to get the existing disk_version which matches its value.
             log.debug("Scanning the disk versions from %s" % metadata.destination)

cloudpub/ms_azure/utils.py

@@ -51,6 +51,9 @@ def __init__(
             legacy_sku_id (str, optional):
                 Only required when ``support_legacy == True``. The SKU ID for Gen1.
                 Defaults to ``{sku_id}-gen1``
+            check_base_sas_only (bool, optional):
+                Indicates to skip checking SAS parameters when set as ``True``.
+                Default to ``False``
             **kwargs
                 Arguments for :class:`~cloudpub.common.PublishingMetadata`.
         """
@@ -60,6 +63,7 @@ def __init__(
         self.support_legacy = support_legacy
         self.recommended_sizes = recommended_sizes or []
         self.legacy_sku_id = kwargs.pop("legacy_sku_id", None)
+        self.check_base_sas_only = kwargs.pop("check_base_sas_only", False)
 
         if generation == "V1" or not support_legacy:
             self.legacy_sku_id = None
@@ -113,7 +117,7 @@ def get_image_type_mapping(architecture: str, generation: str) -> str:
     return gen_map.get(generation, "")
 
 
-def is_sas_eq(sas1: str, sas2: str) -> bool:
+def is_sas_eq(sas1: str, sas2: str, base_only=False) -> bool:
     """
     Compare 2 SAS URI and determine where they're equivalent.
 
@@ -127,10 +131,12 @@ def is_sas_eq(sas1: str, sas2: str) -> bool:
     This comparison is necessary as each time a SAS URI is generated it returns a different value.
 
     Args:
-        sas1:
+        sas1 (str):
             The left SAS to compare the equivalency
-        sas2:
+        sas2 (str):
             The right SAS to compare the equivalency
+        base_only (bool):
+            When True it will only compare the base SAS and not its arguments. Defaults to False.
 
     Returns:
         True when both SAS URIs are equivalent, False otherwise.
@@ -147,25 +153,26 @@ def is_sas_eq(sas1: str, sas2: str) -> bool:
         log.debug("Got different base SAS: %s - Expected: %s" % (base_sas1, base_sas2))
         return False
 
-    # Parameters lengh differs
-    if len(params_sas1) != len(params_sas2):
-        log.debug(
-            "Got different lengh of SAS parameters: len(%s) - Expected len(%s)"
-            % (params_sas1, params_sas2)
-        )
-        return False
-
-    # Parameters values differs
-    for k, v in params_sas1.items():
-        if v != params_sas2.get(k, None):
-            log.debug("The SAS parameter %s doesn't match %s." % (v, params_sas2.get(k, None)))
+    if not base_only:
+        # Parameters lengh differs
+        if len(params_sas1) != len(params_sas2):
+            log.debug(
+                "Got different lengh of SAS parameters: len(%s) - Expected len(%s)"
+                % (params_sas1, params_sas2)
+            )
             return False
 
+        # Parameters values differs
+        for k, v in params_sas1.items():
+            if v != params_sas2.get(k, None):
+                log.debug("The SAS parameter %s doesn't match %s." % (v, params_sas2.get(k, None)))
+                return False
+
     # Equivalent SAS
     return True
 
 
-def is_sas_present(tech_config: VMIPlanTechConfig, sas_uri: str) -> bool:
+def is_sas_present(tech_config: VMIPlanTechConfig, sas_uri: str, base_only: bool = False) -> bool:
     """
     Check whether the given SAS URI is already present in the disk_version.
 
@@ -174,12 +181,14 @@ def is_sas_present(tech_config: VMIPlanTechConfig, sas_uri: str) -> bool:
             The plan's technical configuraion to seek the SAS_URI.
         sas_uri (str)
             The SAS URI to check whether it's present or not in disk version.
+        base_only (bool):
+            When True it will only compare the base SAS and not its arguments. Defaults to False.
     Returns:
         bool: True when the SAS is present in the plan, False otherwise.
     """
     for disk_version in tech_config.disk_versions:
         for img in disk_version.vm_images:
-            if is_sas_eq(img.source.os_disk.uri, sas_uri):
+            if is_sas_eq(img.source.os_disk.uri, sas_uri, base_only):
                 return True
     return False
 

tests/ms_azure/test_service.py

@@ -1076,7 +1076,9 @@ def test_publish_nodiskversion(
         mock_filter.assert_called_once_with(
             product=product_obj, resource="virtual-machine-plan-technical-configuration"
         )
-        mock_is_sas.assert_called_once_with(technical_config_obj, metadata_azure_obj.image_path)
+        mock_is_sas.assert_called_once_with(
+            technical_config_obj, metadata_azure_obj.image_path, False
+        )
         mock_prep_img.assert_not_called()
         mock_upd_sku.assert_called_once_with(
             disk_versions=expected_tech_config.disk_versions,
@@ -1134,6 +1136,7 @@ def test_publish_saspresent(
         mock_is_sas.assert_called_once_with(
             technical_config_obj,
             metadata_azure_obj.image_path,
+            False,
         )
         mock_prep_img.assert_not_called()
         mock_disk_scratch.assert_not_called()
@@ -1202,6 +1205,7 @@ def test_publish_novmimages(
         mock_is_sas.assert_called_once_with(
             technical_config_obj,
             metadata_azure_obj.image_path,
+            False,
         )
         mock_prep_img.assert_not_called()
         mock_disk_scratch.assert_not_called()
@@ -1267,6 +1271,7 @@ def test_publish_disk_has_images(
         mock_is_sas.assert_called_once_with(
             technical_config_obj,
             metadata_azure_obj.image_path,
+            False,
         )
         mock_prep_img.assert_called_once_with(
             metadata=metadata_azure_obj,
@@ -1389,6 +1394,7 @@ def test_publish_live_x64_only(
         mock_is_sas.assert_called_once_with(
             technical_config_obj,
             metadata_azure_obj.image_path,
+            False,
         )
         mock_prep_img.assert_called_once_with(
             metadata=metadata_azure_obj,
@@ -1479,6 +1485,7 @@ def test_publish_live_arm64_only(
         mock_is_sas.assert_called_once_with(
             technical_config_obj,
             metadata_azure_obj.image_path,
+            False,
         )
         mock_prep_img.assert_called_once_with(
             metadata=metadata_azure_obj,

tests/ms_azure/test_utils.py

@@ -102,31 +102,65 @@ def test_get_image_type_mapping(self, metadata_azure_obj: AzurePublishingMetadat
         assert res == "x64Gen2"
 
     @pytest.mark.parametrize(
-        "sas1,sas2,expected",
+        "sas1,sas2,base_only,expected",
         [
-            ("https://foo.com/bar", "https://foo.com/bar?foo=bar", False),
-            ("https://foo.com/bar?foo=bar&st=aaaaa", "https://foo.com/bar?foo=bar&st=bbb", True),
+            ("https://foo.com/bar", "https://foo.com/bar?foo=bar", False, False),
+            (
+                "https://foo.com/bar?foo=bar&st=aaaaa",
+                "https://foo.com/bar?foo=bar&st=bbb",
+                False,
+                True,
+            ),
             (
                 "https://foo.com/bar?foo=bar&st=a&se=b&sig=c&bar=foo",
                 "https://foo.com/bar?foo=bar&st=d&se=e&sig=f&bar=foo",
+                False,
                 True,
             ),
             (
                 "https://foo.com/bar?foo=bar&st=a&se=b&sig=c&bar=foo",
                 "https://foo.com/bar?foo=foo&st=d&se=e&sig=f",
                 False,
+                False,
+            ),
+            (
+                "https://foo.com/bar?foo=bar&st=aaaaa",
+                "https://bar.com/foo?foo=bar&st=bbb",
+                False,
+                False,
+            ),
+            (
+                "https://foo.com/bar?bar=foo&st=aaaaa",
+                "https://foo.com/bar?foo=bar&st=bbb",
+                False,
+                False,
+            ),
+            (
+                "https://foo.com/bar?foo=bar&st=aaaaa",
+                "https://bar.com/foo?foo=bar&st=bbb",
+                True,
+                False,
+            ),
+            (
+                "https://foo.com/bar?bar=foo&st=aaaaa",
+                "https://foo.com/bar?foo=bar&st=bbb",
+                True,
+                True,
             ),
-            ("https://foo.com/bar?foo=bar&st=aaaaa", "https://bar.com/foo?foo=bar&st=bbb", False),
-            ("https://foo.com/bar?bar=foo&st=aaaaa", "https://foo.com/bar?foo=bar&st=bbb", False),
         ],
     )
     def test_is_sas_present(
-        self, sas1: str, sas2: str, expected: bool, technical_config_obj: VMIPlanTechConfig
+        self,
+        sas1: str,
+        sas2: str,
+        base_only: bool,
+        expected: bool,
+        technical_config_obj: VMIPlanTechConfig,
     ) -> None:
         # Test positive
         technical_config_obj.disk_versions[0].vm_images[0].source.os_disk.uri = sas1
 
-        res = is_sas_present(tech_config=technical_config_obj, sas_uri=sas2)
+        res = is_sas_present(tech_config=technical_config_obj, sas_uri=sas2, base_only=base_only)
         assert res is expected
 
     def test_prepare_vm_images_gen1(