Remove credentials from the integration test recordings

src/aosm/HISTORY.rst

@@ -5,6 +5,7 @@ Release History
 
 unreleased
 ++++++++++
+* Added a `--clean` flag to the NSD `delete` command. This flag will delete the NSDG on top of the other resources deleted by the `delete` command.
 * Added integration tests for `publish` and `delete` commands.
 * Added a `--force` flag to the aosm `build` and `delete` commands. This command will force the `build` or `delete` commands to proceed without waiting on user input to confirm.
 * `az aosm nfd build` options `--order-params` and `--interactive` to help users choose which NF parameters to expose as deployParameters. Feature added that allows CNF value mappings file to be generated if none is supplied.

src/aosm/azext_aosm/custom.py

@@ -53,6 +53,7 @@ def build_definition(
     :type cmd: _type_
     :param config_file: path to the file
     :param definition_type: VNF, CNF
+    :param force: force the build even if the design has already been built
     """
 
     # Read the config from the given file
@@ -208,6 +209,7 @@ def delete_published_definition(
     :param clean: if True, will delete the NFDG, artifact stores and publisher too.
         Defaults to False. Only works if no resources have those as a parent.     Use
         with care.
+    :param force: if True, will not prompt for confirmation before deleting the resources.
     """
     config = _get_config_from_file(
         config_file=config_file, configuration_type=definition_type
@@ -284,6 +286,7 @@ def build_design(cmd, client: HybridNetworkManagementClient, config_file: str, f
     :param client:
     :type client: HybridNetworkManagementClient
     :param config_file: path to the file
+    :param force: force the build, even if the design has already been built
     """
 
     api_clients = ApiClients(
@@ -308,6 +311,7 @@ def delete_published_design(
     cmd,
     client: HybridNetworkManagementClient,
     config_file,
+    clean=False,
     force=False,
 ):
     """
@@ -317,6 +321,8 @@ def delete_published_design(
     :param clean: if True, will delete the NSDG, artifact stores and publisher too.
                   Defaults to False. Only works if no resources have those as a parent.
                     Use with care.
+    :param clean: if True, will delete the NSDG on top of the other resources.
+    :param force: if True, will not prompt for confirmation before deleting the resources.
     """
     config = _get_config_from_file(config_file=config_file, configuration_type=NSD)
 
@@ -325,7 +331,7 @@ def delete_published_design(
     )
 
     destroyer = ResourceDeleter(api_clients, config)
-    destroyer.delete_nsd(force=force)
+    destroyer.delete_nsd(clean=clean, force=force)
 
 
 def publish_design(

src/aosm/azext_aosm/delete/delete.py

@@ -87,7 +87,7 @@ def delete_nfd(self, clean: bool = False, force: bool = False) -> None:
                 self.delete_artifact_store("sa")
             self.delete_publisher()
 
-    def delete_nsd(self, force: bool = False) -> None:
+    def delete_nsd(self, clean: bool = False, force: bool = False) -> None:
         """
         Delete the NSDV and manifests.
 
@@ -102,6 +102,8 @@ def delete_nsd(self, force: bool = False) -> None:
                 f" {self.config.acr_manifest_names} and configuration group schema"
                 f" {self.config.cg_schema_name}?"
             )
+            if clean:
+                print(f"Because of the --clean flag, the NSDG {self.config.nsdg_name} will also be deleted.")
             print("There is no undo. Type 'delete' to confirm")
             if not input_ack("delete", "Confirm delete:"):
                 print("Not proceeding with delete")
@@ -110,6 +112,8 @@ def delete_nsd(self, force: bool = False) -> None:
         self.delete_nsdv()
         self.delete_artifact_manifest("acr")
         self.delete_config_group_schema()
+        if clean:
+            self.delete_nsdg()
 
     def delete_nfdv(self):
         assert isinstance(self.config, NFConfiguration)

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

@@ -0,0 +1,50 @@
+from azure.cli.testsdk.scenario_tests import (
+    RecordingProcessor
+)
+from azure.cli.testsdk.scenario_tests.utilities import is_text_payload
+import json
+
+MOCK_ACR_TOKEN = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+MOCK_SAS_URI = "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
+
+
+class AcrTokenReplacer(RecordingProcessor):
+    def process_response(self, response):
+        ACR_TOKEN = "acrToken"
+        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"]["string"] = json.dumps(response_body)
+            except TypeError:
+                pass
+        return response
+
+
+class SasUriReplacer(RecordingProcessor):
+    def process_response(self, response):
+        CONTAINER_CREDENTIALS = "containerCredentials"
+        CONTAINER_SAS_URI = "containerSasUri"
+        if not (is_text_payload(response) and response["body"]["string"]):
+            return response
+
+        response_body = json.loads(response["body"]["string"])
+        try:
+            if CONTAINER_CREDENTIALS not in response_body:
+                return response
+
+            credentials_list = response_body[CONTAINER_CREDENTIALS]
+            new_credentials_list = []
+
+            for credential in credentials_list:
+                if CONTAINER_SAS_URI in credential:
+                    credential[CONTAINER_SAS_URI] = MOCK_SAS_URI
+                new_credentials_list.append(credential)
+
+            response_body[CONTAINER_CREDENTIALS] = new_credentials_list
+            response["body"]["string"] = json.dumps(response_body)
+        except TypeError:
+            pass
+
+        return response