{Test} Add email replacer for cli scenario test (#31262)

* Add email address replacer for test framework

* add process request

* fix

* re.sub

Author: evelyn-ys

src/azure-cli-testsdk/azure/cli/testsdk/base.py

@@ -24,7 +24,7 @@
                       patch_progress_controller, patch_get_current_system_username)
 from .exceptions import CliExecutionError
 from .utilities import (find_recording_dir, StorageAccountKeyReplacer, GraphClientPasswordReplacer,
-                        MSGraphClientPasswordReplacer, AADAuthRequestFilter)
+                        MSGraphClientPasswordReplacer, AADAuthRequestFilter, EmailAddressReplacer)
 from .reverse_dependency import get_dummy_cli
 
 logger = logging.getLogger('azure.cli.testsdk')
@@ -93,6 +93,7 @@ def __init__(self, method_name, config_file=None, recording_name=None,
         default_recording_processors = [
             SubscriptionRecordingProcessor(MOCKED_SUBSCRIPTION_ID),
             AADAuthRequestFilter(),
+            EmailAddressReplacer(),
             LargeRequestBodyProcessor(),
             LargeResponseBodyProcessor(),
             DeploymentNameReplacer(),

src/azure-cli-testsdk/azure/cli/testsdk/utilities.py

@@ -4,6 +4,7 @@
 # --------------------------------------------------------------------------------------------
 
 import os
+import re
 from contextlib import contextmanager
 
 from .scenario_tests import (create_random_name as create_random_name_base, RecordingProcessor)
@@ -210,6 +211,29 @@ def process_response(self, response):
         return response
 
 
+class EmailAddressReplacer(RecordingProcessor):
+    """Replace email address like xxx@microsoft.com with test@example.com"""
+
+    EMAIL_REPLACEMENT = 'test@example.com'
+
+    def _replace_email_address(self, text):
+        pattern = r'[\w.%#+-]+[%40|@|_]microsoft.com'
+        return re.sub(pattern, self.EMAIL_REPLACEMENT, text)
+
+    def process_request(self, request):
+        request.uri = self._replace_email_address(request.uri)
+        if request.body:
+            body = _byte_to_str(request.body)
+            request.body = self._replace_email_address(body)
+        return request
+
+    def process_response(self, response):
+        if response['body']['string']:
+            body = _byte_to_str(response['body']['string'])
+            response['body']['string'] = self._replace_email_address(body)
+        return response
+
+
 class AADAuthRequestFilter(RecordingProcessor):
     """Remove oauth authentication requests and responses from recording.
     This is derived from OAuthRequestResponsesFilter.