# -------------------------------------------------------------------------------------------- # Copyright (c) Microsoft Corporation. All rights reserved. # Licensed under the MIT License. See License.txt in the project root for license information. # -------------------------------------------------------------------------------------------- import os import sys from datetime import datetime from .scenario_tests import AbstractPreparer, SingleValueReplacer from .base import LiveScenarioTest from .exceptions import CliTestError from .reverse_dependency import get_dummy_cli from .utilities import StorageAccountKeyReplacer, GraphClientPasswordReplacer KEY_RESOURCE_GROUP = 'rg' KEY_VIRTUAL_NETWORK = 'vnet' KEY_VNET_NIC = 'nic' # This preparer's traffic is not recorded. # As a result when tests are run in record mode, sdk calls cannot be made to return the prepared resource group. # Rather the deterministic prepared resource's information should be returned. class NoTrafficRecordingPreparer(AbstractPreparer): from .base import execute as _raw_execute def __init__(self, *args, **kwargs): super().__init__(disable_recording=True, *args, **kwargs) def live_only_execute(self, cli_ctx, command, expect_failure=False): # 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. _ = self.moniker try: if self.test_class_instance.in_recording: return self._raw_execute(cli_ctx, command, expect_failure) except AttributeError: # A test might not have an in_recording attribute. Run live if this is an instance of LiveScenarioTest if isinstance(self.test_class_instance, LiveScenarioTest): return self._raw_execute(cli_ctx, command, expect_failure) return None # Resource Group Preparer and its shorthand decorator class ResourceGroupPreparer(NoTrafficRecordingPreparer, SingleValueReplacer): def __init__(self, name_prefix='clitest.rg', parameter_name='resource_group', parameter_name_for_location='resource_group_location', location='westus', dev_setting_name='AZURE_CLI_TEST_DEV_RESOURCE_GROUP_NAME', dev_setting_location='AZURE_CLI_TEST_DEV_RESOURCE_GROUP_LOCATION', random_name_length=75, key='rg', subscription=None, additional_tags=None): if ' ' in name_prefix: raise CliTestError('Error: Space character in resource group name prefix \'%s\'' % name_prefix) super().__init__(name_prefix, random_name_length) self.cli_ctx = get_dummy_cli() self.location = location self.subscription = subscription self.parameter_name = parameter_name self.parameter_name_for_location = parameter_name_for_location self.key = key self.additional_tags = additional_tags self.dev_setting_name = os.environ.get(dev_setting_name, None) self.dev_setting_location = os.environ.get(dev_setting_location, location) def create_resource(self, name, **kwargs): if self.dev_setting_name: self.test_class_instance.kwargs[self.key] = self.dev_setting_name return {self.parameter_name: self.dev_setting_name, self.parameter_name_for_location: self.dev_setting_location} test_class_path = sys.modules[self.test_class_instance.__module__].__file__.split(os.sep) # get index of the module name for main repo if 'command_modules' in test_class_path: index_of_module = test_class_path.index('command_modules') + 1 # get index of the extension name for extension repo elif 'src' in test_class_path: index_of_module = test_class_path.index('src') + 1 else: index_of_module = -1 module = test_class_path[index_of_module] if index_of_module >= 0 else 'unknown' tags = {'product': 'azurecli', 'cause': 'automation test', 'date': datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%SZ'), 'test': self.test_class_instance._testMethodName, 'module': module} if 'ENV_JOB_NAME' in os.environ: tags['job'] = os.environ['ENV_JOB_NAME'] tags = ' '.join(['{}={}'.format(key, value) for key, value in tags.items()]) if self.additional_tags is not None: tags = tags.join(['{}={}'.format(key, value) for key, value in self.additional_tags.items()]) template = 'az group create --location {} --name {} --tag ' + tags if self.subscription: template += ' --subscription {} '.format(self.subscription) self.live_only_execute(self.cli_ctx, template.format(self.location, name)) self.test_class_instance.kwargs[self.key] = name return {self.parameter_name: name, self.parameter_name_for_location: self.location} def remove_resource(self, name, **kwargs): # delete group if test is being recorded and if the group is not a dev rg if not self.dev_setting_name: template = 'az group delete --name {} --yes --no-wait ' if self.subscription: template += ' --subscription {} '.format(self.subscription) self.live_only_execute(self.cli_ctx, template.format(name)) # Storage Account Preparer and its shorthand decorator # pylint: disable=too-many-instance-attributes class StorageAccountPreparer(NoTrafficRecordingPreparer, SingleValueReplacer): def __init__(self, name_prefix='clitest', sku='Standard_LRS', location='westus', kind='Storage', allow_blob_public_access=False, allow_shared_key_access=None, hns=False, length=24, parameter_name='storage_account', resource_group_parameter_name='resource_group', skip_delete=True, dev_setting_name='AZURE_CLI_TEST_DEV_STORAGE_ACCOUNT_NAME', key='sa'): super().__init__(name_prefix, length) self.cli_ctx = get_dummy_cli() self.location = location self.sku = sku self.kind = kind self.allow_blob_public_access = allow_blob_public_access self.allow_shared_key_access = allow_shared_key_access self.hns = hns self.resource_group_parameter_name = resource_group_parameter_name self.skip_delete = skip_delete self.parameter_name = parameter_name self.key = key self.dev_setting_name = os.environ.get(dev_setting_name, None) def create_resource(self, name, **kwargs): group = self._get_resource_group(**kwargs) if not self.dev_setting_name: template = 'az storage account create -n {} -g {} -l {} --sku {} --kind {} --https-only' if not self.allow_blob_public_access: template += ' --allow-blob-public-access false' else: template += ' --allow-blob-public-access true' if self.allow_shared_key_access is not None: if not self.allow_shared_key_access: template += ' --allow-shared-key-access false' else: template += ' --allow-shared-key-access true' if self.hns: template += ' --hns' self.live_only_execute(self.cli_ctx, template.format( name, group, self.location, self.sku, self.kind, self.hns)) else: name = self.dev_setting_name try: account_key = self.live_only_execute(self.cli_ctx, 'storage account keys list -n {} -g {} --query "[0].value" -otsv' .format(name, group)).output except AttributeError: # live only execute returns None if playing from record account_key = None self.test_class_instance.kwargs[self.key] = name return {self.parameter_name: name, self.parameter_name + '_info': (name, account_key or StorageAccountKeyReplacer.KEY_REPLACEMENT)} def remove_resource(self, name, **kwargs): if not self.skip_delete and not self.dev_setting_name: group = self._get_resource_group(**kwargs) self.live_only_execute(self.cli_ctx, 'az storage account delete -n {} -g {} --yes'.format(name, group)) def _get_resource_group(self, **kwargs): try: return kwargs.get(self.resource_group_parameter_name) except KeyError: template = 'To create a storage account a resource group is required. Please add ' \ 'decorator @{} in front of this storage account preparer.' raise CliTestError(template.format(ResourceGroupPreparer.__name__)) # KeyVault Preparer and its shorthand decorator # pylint: disable=too-many-instance-attributes class KeyVaultPreparer(NoTrafficRecordingPreparer, SingleValueReplacer): def __init__(self, name_prefix='clitest', sku='standard', location='westus', parameter_name='key_vault', resource_group_parameter_name='resource_group', skip_delete=False, skip_purge=False, dev_setting_name='AZURE_CLI_TEST_DEV_KEY_VAULT_NAME', key='kv', name_len=24, additional_params=None): super().__init__(name_prefix, name_len) self.cli_ctx = get_dummy_cli() self.location = location self.sku = sku self.resource_group_parameter_name = resource_group_parameter_name self.skip_delete = skip_delete self.skip_purge = skip_purge self.parameter_name = parameter_name self.key = key self.additional_params = additional_params self.dev_setting_name = os.environ.get(dev_setting_name, None) def create_resource(self, name, **kwargs): if not self.dev_setting_name: group = self._get_resource_group(**kwargs) template = 'az keyvault create -n {} -g {} -l {} --sku {} ' if self.additional_params: template += self.additional_params if '--retention-days' not in template: template += ' --retention-days 7' self.live_only_execute(self.cli_ctx, template.format(name, group, self.location, self.sku)) self.test_class_instance.kwargs[self.key] = name return {self.parameter_name: name} self.test_class_instance.kwargs[self.key] = self.dev_setting_name return {self.parameter_name: self.dev_setting_name} def remove_resource(self, name, **kwargs): if not self.skip_delete and not self.dev_setting_name: group = self._get_resource_group(**kwargs) self.live_only_execute(self.cli_ctx, 'az keyvault delete -n {} -g {}'.format(name, group)) from azure.core.exceptions import HttpResponseError if self.skip_purge: return try: self.live_only_execute(self.cli_ctx, 'az keyvault purge -n {} -l {}'.format(name, self.location)) except HttpResponseError: # purge operation will fail with HttpResponseError when --enable-purge-protection pass def _get_resource_group(self, **kwargs): try: return kwargs.get(self.resource_group_parameter_name) except KeyError: template = 'To create a KeyVault a resource group is required. Please add ' \ 'decorator @{} in front of this KeyVault preparer.' raise CliTestError(template.format(ResourceGroupPreparer.__name__)) # Managed HSM Preparer and its shorthand decorator # pylint: disable=too-many-instance-attributes class ManagedHSMPreparer(NoTrafficRecordingPreparer, SingleValueReplacer): def __init__(self, certs_path, name_prefix='clitest', location='uksouth', key='hsm', name_len=24, parameter_name='managed_hsm', resource_group_parameter_name='resource_group', administrators=None, roles=[], additional_params=None, skip_delete=False, skip_purge=False): super().__init__(name_prefix, name_len) self.cli_ctx = get_dummy_cli() self.location = location self.resource_group_parameter_name = resource_group_parameter_name self.parameter_name = parameter_name self.key = key self.certs_path = certs_path self.administrators = administrators self.roles = roles self.additional_params = additional_params self.skip_delete = skip_delete self.skip_purge = skip_purge def create_resource(self, name, **kwargs): group = self._get_resource_group(**kwargs) administrators = self.administrators or self._get_signed_in_user() template = 'az keyvault create --hsm-name {} -g {} -l {} --administrators {} --retention-days 7' if self.additional_params: template += self.additional_params self.live_only_execute(self.cli_ctx, template.format(name, group, self.location, administrators)) # After creating MHSM, All data plane commands are disabled until the HSM is activated. # To activate the HSM, we must download the Security Domain. # See https://learn.microsoft.com/en-us/azure/key-vault/managed-hsm/quick-create-cli#activate-your-managed-hsm for details if self.certs_path: cert0 = os.path.join(self.certs_path, 'cert_0.cer').replace('\\', '\\\\') cert1 = os.path.join(self.certs_path, 'cert_1.cer').replace('\\', '\\\\') cert2 = os.path.join(self.certs_path, 'cert_2.cer').replace('\\', '\\\\') security_domain = os.path.join(self.certs_path, f'{name}-SD.json').replace('\\', '\\\\') activate_template = f'az keyvault security-domain download --hsm-name {name} --sd-wrapping-keys {cert0} {cert1} {cert2} --sd-quorum 2 --security-domain-file {security_domain}' self.live_only_execute(self.cli_ctx, activate_template) if self.roles: for role in self.roles: role_assignment_template = f'az keyvault role assignment create --hsm-name {name} --role "{role}" --assignee {administrators} --scope "/"' self.live_only_execute(self.cli_ctx, role_assignment_template) self.test_class_instance.kwargs[self.key] = name return {self.parameter_name: name} def remove_resource(self, name, **kwargs): security_domain = os.path.join(self.certs_path, f'{name}-SD.json').replace('\\', '\\\\') if os.path.exists(security_domain): os.remove(security_domain) from azure.core.exceptions import HttpResponseError try: if not self.skip_delete: group = self._get_resource_group(**kwargs) self.live_only_execute(self.cli_ctx, 'az keyvault delete --hsm-name {} -g {}'.format(name, group)) if self.skip_purge: return self.live_only_execute(self.cli_ctx, 'az keyvault purge --hsm-name {} -l {}'.format(name, self.location)) except HttpResponseError: # delete operation will fail with HttpResponseError when there's ongoing updates # purge operation will fail with HttpResponseError when --enable-purge-protection pass def _get_signed_in_user(self): try: user_info = self.live_only_execute(self.cli_ctx, 'ad signed-in-user show').get_output_in_json() return user_info['id'] if user_info else None except Exception: return None def _get_resource_group(self, **kwargs): try: return kwargs.get(self.resource_group_parameter_name) except KeyError: template = 'To create a Managed HSM, a resource group is required. Please add ' \ 'decorator @{} in front of this ManagedHSM preparer.' raise CliTestError(template.format(ResourceGroupPreparer.__name__)) # Role based access control service principal preparer # pylint: disable=too-many-instance-attributes class RoleBasedServicePrincipalPreparer(NoTrafficRecordingPreparer, SingleValueReplacer): 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().__init__(name_prefix, 24) self.cli_ctx = get_dummy_cli() self.skip_assignment = skip_assignment self.result = {} self.parameter_name = parameter_name self.parameter_password = parameter_password self.dev_setting_sp_name = os.environ.get(dev_setting_sp_name, None) self.dev_setting_sp_password = os.environ.get(dev_setting_sp_password, None) self.key = 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} 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} def remove_resource(self, name, **kwargs): if not self.dev_setting_sp_name: self.live_only_execute(self.cli_ctx, 'az ad sp delete --id {}'.format(self.result.get('appId'))) # Managed Application preparer # pylint: disable=too-many-instance-attributes class ManagedApplicationPreparer(AbstractPreparer, SingleValueReplacer): from .base import execute def __init__(self, name_prefix='clitest', parameter_name='aad_client_app_id', parameter_secret='aad_client_app_secret', app_name='app_name', dev_setting_app_name='AZURE_CLI_TEST_DEV_APP_NAME', dev_setting_app_secret='AZURE_CLI_TEST_DEV_APP_SECRET', key='app'): super().__init__(name_prefix, 24) self.cli_ctx = get_dummy_cli() self.parameter_name = parameter_name self.parameter_secret = parameter_secret self.result = {} self.app_name = app_name self.dev_setting_app_name = os.environ.get(dev_setting_app_name, None) self.dev_setting_app_secret = os.environ.get(dev_setting_app_secret, None) self.key = key def create_resource(self, name, **kwargs): if not self.dev_setting_app_name: template = 'az ad app create --display-name {} --key-type Password --password {} --identifier-uris ' \ 'http://{}' self.result = self.execute(self.cli_ctx, template.format(name, name, name)).get_output_in_json() self.test_class_instance.kwargs[self.key] = name return {self.parameter_name: self.result['appId'], self.parameter_secret: name} self.test_class_instance.kwargs[self.key] = name return {self.parameter_name: self.dev_setting_app_name, self.parameter_secret: self.dev_setting_app_secret} def remove_resource(self, name, **kwargs): if not self.dev_setting_app_name: self.execute(self.cli_ctx, 'az ad app delete --id {}'.format(self.result['appId'])) # pylint: disable=too-many-instance-attributes class VirtualNetworkPreparer(NoTrafficRecordingPreparer, SingleValueReplacer): def __init__(self, name_prefix='clitest.vn', location='westus', parameter_name='virtual_network', resource_group_parameter_name='resource_group', resource_group_key=KEY_RESOURCE_GROUP, dev_setting_name='AZURE_CLI_TEST_DEV_VIRTUAL_NETWORK_NAME', random_name_length=24, key=KEY_VIRTUAL_NETWORK): if ' ' in name_prefix: raise CliTestError( 'Error: Space character in name prefix \'%s\'' % name_prefix) super().__init__( name_prefix, random_name_length) self.cli_ctx = get_dummy_cli() self.location = location self.parameter_name = parameter_name self.key = key self.resource_group_parameter_name = resource_group_parameter_name self.resource_group_key = resource_group_key self.dev_setting_name = os.environ.get(dev_setting_name, None) def create_resource(self, name, **kwargs): if self.dev_setting_name: self.test_class_instance.kwargs[self.key] = name return {self.parameter_name: self.dev_setting_name, } tags = {'product': 'azurecli', 'cause': 'automation', 'date': datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%SZ')} if 'ENV_JOB_NAME' in os.environ: tags['job'] = os.environ['ENV_JOB_NAME'] tags = ' '.join(['{}={}'.format(key, value) for key, value in tags.items()]) template = 'az network vnet create --resource-group {} --location {} --name {} --subnet-name default --tag ' + tags self.live_only_execute(self.cli_ctx, template.format(self._get_resource_group(**kwargs), self.location, name)) self.test_class_instance.kwargs[self.key] = name return {self.parameter_name: name} def remove_resource(self, name, **kwargs): if not self.dev_setting_name: from azure.core.exceptions import HttpResponseError try: self.live_only_execute( self.cli_ctx, 'az network vnet delete --name {} --resource-group {}'.format(name, self._get_resource_group(**kwargs))) except HttpResponseError: # deletion of vnet may fail as service could create subresources like IPConfig. We could rely on the deletion of resource group to delete the vnet. pass def _get_resource_group(self, **kwargs): try: return kwargs.get(self.resource_group_parameter_name) except KeyError: template = 'To create a VirtualNetwork a resource group is required. Please add ' \ 'decorator @{} in front of this VirtualNetwork preparer.' raise CliTestError(template.format(VirtualNetworkPreparer.__name__)) # pylint: disable=too-many-instance-attributes class VnetNicPreparer(NoTrafficRecordingPreparer, SingleValueReplacer): def __init__(self, name_prefix='clitest.nic', parameter_name='subnet', resource_group_parameter_name=KEY_RESOURCE_GROUP, vnet_parameter_name=KEY_VIRTUAL_NETWORK, dev_setting_name='AZURE_CLI_TEST_DEV_VNET_NIC_NAME', key=KEY_VNET_NIC): if ' ' in name_prefix: raise CliTestError( 'Error: Space character in name prefix \'%s\'' % name_prefix) super().__init__(name_prefix, 15) self.cli_ctx = get_dummy_cli() self.parameter_name = parameter_name self.key = key self.resource_group_parameter_name = resource_group_parameter_name self.vnet_parameter_name = vnet_parameter_name self.dev_setting_name = os.environ.get(dev_setting_name, None) def create_resource(self, name, **kwargs): if self.dev_setting_name: self.test_class_instance.kwargs[self.key] = name return {self.parameter_name: self.dev_setting_name, } template = 'az network nic create --resource-group {} --name {} --vnet-name {} --subnet default ' self.live_only_execute(self.cli_ctx, template.format( self._get_resource_group(**kwargs), name, self._get_virtual_network(**kwargs))) self.test_class_instance.kwargs[self.key] = name return {self.parameter_name: name} def remove_resource(self, name, **kwargs): if not self.dev_setting_name: self.live_only_execute( self.cli_ctx, 'az network nic delete --name {} --resource-group {}'.format(name, self._get_resource_group(**kwargs))) def _get_resource_group(self, **kwargs): try: return kwargs.get(self.resource_group_parameter_name) except KeyError: template = 'To create a VirtualNetworkNic a resource group is required. Please add ' \ 'decorator @{} in front of this VirtualNetworkNic preparer.' raise CliTestError(template.format(VnetNicPreparer.__name__)) def _get_virtual_network(self, **kwargs): try: return kwargs.get(self.vnet_parameter_name) except KeyError: template = 'To create a VirtualNetworkNic a virtual network is required. Please add ' \ 'decorator @{} in front of this VirtualNetworkNic preparer.' raise CliTestError(template.format(VnetNicPreparer.__name__)) class LogAnalyticsWorkspacePreparer(NoTrafficRecordingPreparer, SingleValueReplacer): def __init__(self, name_prefix='laworkspace', location='eastus2euap', parameter_name='laworkspace', resource_group_parameter_name='resource_group', skip_delete=False, get_shared_key=False): super().__init__(name_prefix, 15) self.cli_ctx = get_dummy_cli() self.location = location self.parameter_name = parameter_name self.resource_group_parameter_name = resource_group_parameter_name self.skip_delete = skip_delete self.get_shared_key = get_shared_key def create_resource(self, name, **kwargs): group = self._get_resource_group(**kwargs) template = ('az monitor log-analytics workspace create -l {} -g {} -n {}') try: customer_id = self.live_only_execute(self.cli_ctx, template.format(self.location, group, name)).get_output_in_json()["customerId"] except AttributeError: # live only execute returns None if playing from record customer_id = None if self.get_shared_key: get_share_key_template = ('az monitor log-analytics workspace get-shared-keys -g {} -n {}') try: log_shared_key = self.live_only_execute(self.cli_ctx, get_share_key_template.format(group, name)).get_output_in_json()["primarySharedKey"] except AttributeError: # live only execute returns None if playing from record log_shared_key = None return {self.parameter_name: name, self.parameter_name + '_customer_id': (customer_id or 'veryFakedCustomerId=='), self.parameter_name + '_shared_key': (log_shared_key or 'veryFakedPrivateSharedKey==')} return {self.parameter_name: name} def remove_resource(self, name, **kwargs): if not self.skip_delete: group = self._get_resource_group(**kwargs) template = ('az monitor log-analytics workspace delete -g {} -n {} --yes') self.live_only_execute(self.cli_ctx, template.format(group, name)) def _get_resource_group(self, **kwargs): try: return kwargs.get(self.resource_group_parameter_name) except KeyError: template = 'To create a log analytics workspace a resource group is required. Please add ' \ 'decorator @{} in front of this preparer.' raise CliTestError(template.format(ResourceGroupPreparer.__name__, self.resource_group_parameter_name)) # Utility def is_preparer_func(fn): return getattr(fn, '__is_preparer', False)