Fix integration tests (#67)

* Fix integration tests

* Update src/aosm/development.md

Co-authored-by: Cyclam <[email protected]>

* Update src/aosm/azext_aosm/tests/latest/test_aosm_cnf_publish_and_delete.py

Co-authored-by: Cyclam <[email protected]>

* Update src/aosm/azext_aosm/tests/latest/test_aosm_cnf_publish_and_delete.py

Co-authored-by: Cyclam <[email protected]>

* Markup changes

---------

Co-authored-by: Cyclam <[email protected]>
Co-authored-by: Sunny Carter <[email protected]>

src/aosm/azext_aosm/deploy/deploy_with_arm.py

@@ -541,26 +541,42 @@ def validate_and_deploy_arm_template(
                 :raise RuntimeError if validation or deploy fails
         :return: Output dictionary from the bicep template.
         """
-        # Get current time from the time module and remove all digits after the decimal point
+        # Get current time from the time module and remove all digits after the decimal
+        # point
         current_time = str(time.time()).split(".", maxsplit=1)[0]
 
         # Add a timestamp to the deployment name to ensure it is unique
         deployment_name = f"AOSM_CLI_deployment_{current_time}"
 
-        validation = self.api_clients.resource_client.deployments.begin_validate(
-            resource_group_name=resource_group,
-            deployment_name=deployment_name,
-            parameters={
-                "properties": {
-                    "mode": "Incremental",
-                    "template": template,
-                    "parameters": parameters,
-                }
-            },
-        )
-        validation_res = LongRunningOperation(
-            self.cli_ctx, "Validating ARM template..."
-        )(validation)
+        # Validation is automatically re-attempted in live runs, but not in test
+        # playback, causing them to fail. This explicitly re-attempts validation to
+        # ensure the tests pass
+        validation_res = None
+        for validation_attempt in range(2):
+            try:
+                validation = self.api_clients.resource_client.deployments.begin_validate(
+                    resource_group_name=resource_group,
+                    deployment_name=deployment_name,
+                    parameters={
+                        "properties": {
+                            "mode": "Incremental",
+                            "template": template,
+                            "parameters": parameters,
+                        }
+                    },
+                )
+                validation_res = LongRunningOperation(
+                    self.cli_ctx, "Validating ARM template..."
+                )(validation)
+                break
+            except Exception:  # pylint: disable=broad-except
+                if validation_attempt == 1:
+                    raise
+
+        if not validation_res:
+            # Don't expect to hit this but it appeases mypy
+            raise RuntimeError(f"Validation of template {template} failed.")
+
         logger.debug("Validation Result %s", validation_res)
         if validation_res.error:
             # Validation failed so don't even try to deploy

src/aosm/azext_aosm/tests/latest/recording_processors.py

@@ -7,22 +7,31 @@
 # the recordings so that we can avoid checking in secrets to the repo.
 # --------------------------------------------------------------------------------------------
 
-from azure.cli.testsdk.scenario_tests import RecordingProcessor
+from azure.cli.testsdk.scenario_tests import (
+    RecordingProcessor
+)
 from azure.cli.testsdk.scenario_tests.utilities import is_text_payload
 import json
+import re
 
-MOCK_ACR_TOKEN = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
-MOCK_SAS_URI = "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
+MOCK_TOKEN = "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
+MOCK_SAS_URI = "https://xxxxxxxxxxxxxxx.blob.core.windows.net"
+MOCK_STORAGE_ACCOUNT_SR = "&si=StorageAccountAccessPolicy&sr=xxxxxxxxxxxxxxxxxxxx"
+BLOB_STORE_URI_REGEX = r"https:\/\/[a-zA-Z0-9]+\.blob\.core\.windows\.net"
+STORAGE_ACCOUNT_SR_REGEX = r"&si=StorageAccountAccessPolicy&sr=.*"
 
 
-class AcrTokenReplacer(RecordingProcessor):
+class TokenReplacer(RecordingProcessor):
     def process_response(self, response):
         ACR_TOKEN = "acrToken"
+        ACCESS_TOKEN = "access_token"
         if is_text_payload(response) and response["body"]["string"]:
             try:
                 response_body = json.loads(response["body"]["string"])
                 if ACR_TOKEN in response_body:
-                    response_body[ACR_TOKEN] = MOCK_ACR_TOKEN
+                    response_body[ACR_TOKEN] = MOCK_TOKEN
+                if ACCESS_TOKEN in response_body:
+                    response_body[ACCESS_TOKEN] = MOCK_TOKEN
                 response["body"]["string"] = json.dumps(response_body)
             except TypeError:
                 pass
@@ -46,7 +55,8 @@ def process_response(self, response):
 
             for credential in credentials_list:
                 if CONTAINER_SAS_URI in credential:
-                    credential[CONTAINER_SAS_URI] = MOCK_SAS_URI
+                    credential[CONTAINER_SAS_URI] = re.sub(BLOB_STORE_URI_REGEX, MOCK_SAS_URI, credential[CONTAINER_SAS_URI])
+                    credential[CONTAINER_SAS_URI] = re.sub(STORAGE_ACCOUNT_SR_REGEX, MOCK_STORAGE_ACCOUNT_SR, credential[CONTAINER_SAS_URI])
                 new_credentials_list.append(credential)
 
             response_body[CONTAINER_CREDENTIALS] = new_credentials_list
@@ -55,3 +65,15 @@ def process_response(self, response):
             pass
 
         return response
+
+
+class BlobStoreUriReplacer(RecordingProcessor):
+    def process_request(self, request):
+        try:
+            request.uri = re.sub(BLOB_STORE_URI_REGEX, MOCK_SAS_URI, request.uri)
+            request.uri = re.sub(STORAGE_ACCOUNT_SR_REGEX, MOCK_STORAGE_ACCOUNT_SR, request.uri)
+
+        except TypeError:
+            pass
+
+        return request