add AKSCustomRoleBasedServicePrincipalPreparer

src/azure-cli/azure/cli/command_modules/acs/tests/latest/custom_preparers.py

@@ -9,9 +9,11 @@
 from azure.cli.testsdk.preparers import (
     ResourceGroupPreparer,
     VirtualNetworkPreparer,
+    RoleBasedServicePrincipalPreparer,
     KEY_RESOURCE_GROUP,
     KEY_VIRTUAL_NETWORK,
 )
+from azure.cli.testsdk.utilities import GraphClientPasswordReplacer
 
 
 class AKSCustomResourceGroupPreparer(ResourceGroupPreparer):
@@ -135,3 +137,65 @@ def _update_address_prefixes(self, **kwargs):
             self.address_prefixes = kwargs.get(
                 self.address_prefixes_parameter_name
             )
+
+
+class AKSCustomRoleBasedServicePrincipalPreparer(
+    RoleBasedServicePrincipalPreparer
+):
+    def __init__(
+        self,
+        name_prefix="clitest",
+        skip_assignment=True,
+        parameter_name="sp_name",
+        parameter_password="sp_password",
+        dev_setting_sp_name="AZURE_CLI_TEST_DEV_SP_NAME",
+        dev_setting_sp_password="AZURE_CLI_TEST_DEV_SP_PASSWORD",
+        key="sp",
+    ):
+        super(AKSCustomRoleBasedServicePrincipalPreparer, self).__init__(
+            name_prefix,
+            skip_assignment,
+            parameter_name,
+            parameter_password,
+            dev_setting_sp_name,
+            dev_setting_sp_password,
+            key,
+        )
+
+    def create_resource(self, name, **kwargs):
+        if not self.dev_setting_sp_name:
+            command = "az ad sp create-for-rbac -n {}{}".format(
+                name, " --skip-assignment" if self.skip_assignment else ""
+            )
+
+            try:
+                self.result = self.live_only_execute(
+                    self.cli_ctx, command
+                ).get_output_in_json()
+            except AttributeError:  # live only execute returns None if playing from record
+                pass
+
+            self.test_class_instance.kwargs[self.key] = name
+            self.test_class_instance.kwargs[
+                "{}_pass".format(self.key)
+            ] = self.parameter_password
+            return {
+                self.parameter_name: name,
+                self.parameter_password: self.result.get("password")
+                or GraphClientPasswordReplacer.PWD_REPLACEMENT,
+            }
+        else:
+            # call AbstractPreparer.moniker to make resource counts and self.resource_moniker consistent between live and
+            # play-back. see SingleValueReplacer.process_request, AbstractPreparer.__call__._preparer_wrapper
+            # and ScenarioTest.create_random_name. This is so that when self.create_random_name is called for the
+            # first time during live or playback, it would have the same value.
+            # In short, the default sp preparer in live mode does not call moniker, which leads to inconsistent counts.
+            _ = self.moniker
+            self.test_class_instance.kwargs[self.key] = self.dev_setting_sp_name
+            self.test_class_instance.kwargs[
+                "{}_pass".format(self.key)
+            ] = self.dev_setting_sp_password
+            return {
+                self.parameter_name: self.dev_setting_sp_name,
+                self.parameter_password: self.dev_setting_sp_password,
+            }

src/azure-cli/azure/cli/command_modules/acs/tests/latest/test_aks_commands.py

@@ -20,7 +20,7 @@
 from azure.cli.command_modules.acs._consts import CONST_KUBE_DASHBOARD_ADDON_NAME
 from .recording_processors import KeyReplacer
 
-from .custom_preparers import AKSCustomResourceGroupPreparer, AKSCustomVirtualNetworkPreparer
+from .custom_preparers import AKSCustomResourceGroupPreparer, AKSCustomVirtualNetworkPreparer, AKSCustomRoleBasedServicePrincipalPreparer
 # flake8: noqa
 
 
@@ -4834,9 +4834,11 @@ def test_aks_create_network_cidr(self, resource_group, resource_group_location):
         self.cmd(
             'aks delete -g {resource_group} -n {name} --yes --no-wait', checks=[self.is_empty()])
 
+    # live only due to dependency `_add_role_assignment` is not mocked
+    @live_only()
     @AllowLargeResponse()
     @AKSCustomResourceGroupPreparer(random_name_length=17, name_prefix='clitest', location='westus2')
-    @RoleBasedServicePrincipalPreparer()
+    @AKSCustomRoleBasedServicePrincipalPreparer()
     def test_aks_create_attach_acr(self, resource_group, resource_group_location, sp_name, sp_password):
         aks_name = self.create_random_name('cliakstest', 16)
         acr_name = self.create_random_name('cliaksacr', 16)