# coding: utf-8 # ------------------------------------------------------------------------- # Copyright (c) Microsoft Corporation. All rights reserved. # Licensed under the MIT License. See License.txt in the project root for # license information. # -------------------------------------------------------------------------- import unittest from azure.mgmt.containerregistry.v2017_10_01.models import ( Registry, RegistryUpdateParameters, StorageAccountProperties, Sku, SkuName, SkuTier, ProvisioningState, PasswordName, WebhookCreateParameters, WebhookUpdateParameters, WebhookAction, WebhookStatus, ) import azure.mgmt.storage from devtools_testutils import AzureMgmtTestCase, FakeStorageAccount, ResourceGroupPreparer, StorageAccountPreparer FAKE_STORAGE = FakeStorageAccount(name="pyacr", id="") DEFAULT_LOCATION = "westcentralus" DEFAULT_REPLICATION_LOCATION = "southcentralus" DEFAULT_WEBHOOK_SERVICE_URI = "http://www.microsoft.com" DEFAULT_WEBHOOK_SCOPE = "hello-world" DEFAULT_KEY_VALUE_PAIR = {"key": "value"} class MgmtACRTest(AzureMgmtTestCase): def setUp(self): super(MgmtACRTest, self).setUp() self.client = self.create_mgmt_client(azure.mgmt.containerregistry.ContainerRegistryManagementClient) @ResourceGroupPreparer(location=DEFAULT_LOCATION) @StorageAccountPreparer(name_prefix="pyacr", location=DEFAULT_LOCATION, playback_fake_resource=FAKE_STORAGE) def _disabled_test_classic_registry(self, resource_group, location, storage_account): registry_name = self.get_resource_name("pyacr") name_status = self.client.registries.check_name_availability(registry_name) self.assertTrue(name_status.name_available) # Create a classic registry registry = self.client.registries.create( resource_group_name=resource_group.name, registry_name=registry_name, registry=Registry( location=location, sku=Sku(name=SkuName.classic), storage_account=StorageAccountProperties(id=storage_account.id), ), ).result() self.assertEqual(registry.name, registry_name) self.assertEqual(registry.location, location) self.assertEqual(registry.sku.name, SkuName.classic.value) self.assertEqual(registry.sku.tier, SkuTier.classic.value) self.assertEqual(registry.provisioning_state, ProvisioningState.succeeded.value) self.assertEqual(registry.admin_user_enabled, False) self._core_registry_scenario(registry_name, resource_group.name) @ResourceGroupPreparer(location=DEFAULT_LOCATION) def test_managed_registry(self, resource_group, location): registry_name = self.get_resource_name("pyacr") name_status = self.client.registries.check_name_availability(registry_name) self.assertTrue(name_status.name_available) # Create a managed registry self._create_managed_registry(registry_name, resource_group.name, location) self._core_registry_scenario(registry_name, resource_group.name) def _core_registry_scenario(self, registry_name, resource_group_name): registries = list(self.client.registries.list_by_resource_group(resource_group_name)) self.assertEqual(len(registries), 1) # Update the registry with new tags and enable admin user registry = self.client.registries.update( resource_group_name=resource_group_name, registry_name=registry_name, registry_update_parameters=RegistryUpdateParameters(tags=DEFAULT_KEY_VALUE_PAIR, admin_user_enabled=True), ).result() self.assertEqual(registry.name, registry_name) self.assertEqual(registry.tags, DEFAULT_KEY_VALUE_PAIR) self.assertEqual(registry.admin_user_enabled, True) registry = self.client.registries.get(resource_group_name, registry_name) self.assertEqual(registry.name, registry_name) self.assertEqual(registry.tags, DEFAULT_KEY_VALUE_PAIR) self.assertEqual(registry.admin_user_enabled, True) credentials = self.client.registries.list_credentials(resource_group_name, registry_name) self.assertEqual(len(credentials.passwords), 2) credentials = self.client.registries.regenerate_credential( resource_group_name, registry_name, PasswordName.password ) self.assertEqual(len(credentials.passwords), 2) if registry.sku.name == SkuName.premium.value: usages = self.client.registries.list_usages(resource_group_name, registry_name) self.assertTrue(len(usages.value) > 1) self.client.registries.delete(resource_group_name, registry_name).wait() def _create_managed_registry(self, registry_name, resource_group_name, location): registry = self.client.registries.create( resource_group_name=resource_group_name, registry_name=registry_name, registry=Registry(location=location, sku=Sku(name=SkuName.premium)), ).result() self.assertEqual(registry.name, registry_name) self.assertEqual(registry.location, location) self.assertEqual(registry.sku.name, SkuName.premium.value) self.assertEqual(registry.sku.tier, SkuTier.premium.value) self.assertEqual(registry.provisioning_state, ProvisioningState.succeeded.value) self.assertEqual(registry.admin_user_enabled, False) self.assertEqual(registry.storage_account, None) @ResourceGroupPreparer(location=DEFAULT_LOCATION) def test_webhook(self, resource_group, location): registry_name = self.get_resource_name("pyacr") webhook_name = self.get_resource_name("pyacr") # Create a managed registry self._create_managed_registry(registry_name, resource_group.name, location) # Create a webhook webhook = self.client.webhooks.create( resource_group_name=resource_group.name, registry_name=registry_name, webhook_name=webhook_name, webhook_create_parameters=WebhookCreateParameters( location=location, service_uri=DEFAULT_WEBHOOK_SERVICE_URI, actions=[WebhookAction.push] ), ).result() self.assertEqual(webhook.name, webhook_name) self.assertEqual(webhook.location, location) self.assertEqual(webhook.provisioning_state, ProvisioningState.succeeded.value) self.assertEqual(webhook.actions, [WebhookAction.push.value]) self.assertEqual(webhook.status, WebhookStatus.enabled.value) webhooks = list(self.client.webhooks.list(resource_group.name, registry_name)) self.assertEqual(len(webhooks), 1) # Update the webhook with custom headers, scope, and new tags webhook = self.client.webhooks.update( resource_group_name=resource_group.name, registry_name=registry_name, webhook_name=webhook_name, webhook_update_parameters=WebhookUpdateParameters( tags=DEFAULT_KEY_VALUE_PAIR, custom_headers=DEFAULT_KEY_VALUE_PAIR, scope=DEFAULT_WEBHOOK_SCOPE ), ).result() self.assertEqual(webhook.name, webhook_name) self.assertEqual(webhook.tags, DEFAULT_KEY_VALUE_PAIR) self.assertEqual(webhook.scope, DEFAULT_WEBHOOK_SCOPE) webhook = self.client.webhooks.get(resource_group.name, registry_name, webhook_name) self.assertEqual(webhook.name, webhook_name) self.assertEqual(webhook.tags, DEFAULT_KEY_VALUE_PAIR) self.assertEqual(webhook.scope, DEFAULT_WEBHOOK_SCOPE) webhook_config = self.client.webhooks.get_callback_config(resource_group.name, registry_name, webhook_name) self.assertEqual(webhook_config.service_uri, DEFAULT_WEBHOOK_SERVICE_URI) self.assertEqual(webhook_config.custom_headers, DEFAULT_KEY_VALUE_PAIR) self.client.webhooks.ping(resource_group.name, registry_name, webhook_name) self.client.webhooks.list_events(resource_group.name, registry_name, webhook_name) self.client.webhooks.delete(resource_group.name, registry_name, webhook_name).wait() self.client.registries.delete(resource_group.name, registry_name).wait() @ResourceGroupPreparer(location=DEFAULT_LOCATION) def _disabled_test_replication(self, resource_group, location): registry_name = self.get_resource_name("pyacr") replication_name = DEFAULT_REPLICATION_LOCATION # Create a managed registry self._create_managed_registry(registry_name, resource_group.name, location) # Create a replication replication = self.client.replications.create( resource_group_name=resource_group.name, registry_name=registry_name, replication_name=replication_name, location=DEFAULT_REPLICATION_LOCATION, ).result() self.assertEqual(replication.name, replication_name) self.assertEqual(replication.location, DEFAULT_REPLICATION_LOCATION) self.assertEqual(replication.provisioning_state, ProvisioningState.succeeded.value) replications = list(self.client.replications.list(resource_group.name, registry_name)) self.assertEqual(len(replications), 2) # 2 because a replication in home region is auto created # Update the replication with new tags replication = self.client.replications.update( resource_group_name=resource_group.name, registry_name=registry_name, replication_name=replication_name, tags=DEFAULT_KEY_VALUE_PAIR, ).result() self.assertEqual(replication.name, replication_name) self.assertEqual(replication.tags, DEFAULT_KEY_VALUE_PAIR) replication = self.client.replications.get(resource_group.name, registry_name, replication_name) self.assertEqual(replication.name, replication_name) self.assertEqual(replication.tags, DEFAULT_KEY_VALUE_PAIR) self.client.replications.delete(resource_group.name, registry_name, replication_name).wait() self.client.registries.delete(resource_group.name, registry_name).wait() # ------------------------------------------------------------------------------ if __name__ == "__main__": unittest.main()