From dba4f9a184cc9008373c250509394363c2c77f61 Mon Sep 17 00:00:00 2001 From: Yuchao Yan Date: Wed, 18 Mar 2026 14:51:52 +0800 Subject: [PATCH] Remove disabled test files for azure-mgmt-communication --- .../tests/disable_test_mgmt_communication.py | 104 ------------------ ..._test_mgmt_communication_list_resources.py | 98 ----------------- ...est_mgmt_communication_notification_hub.py | 97 ---------------- 3 files changed, 299 deletions(-) delete mode 100644 sdk/communication/azure-mgmt-communication/tests/disable_test_mgmt_communication.py delete mode 100644 sdk/communication/azure-mgmt-communication/tests/disable_test_mgmt_communication_list_resources.py delete mode 100644 sdk/communication/azure-mgmt-communication/tests/disable_test_mgmt_communication_notification_hub.py diff --git a/sdk/communication/azure-mgmt-communication/tests/disable_test_mgmt_communication.py b/sdk/communication/azure-mgmt-communication/tests/disable_test_mgmt_communication.py deleted file mode 100644 index 2e25867f753c..000000000000 --- a/sdk/communication/azure-mgmt-communication/tests/disable_test_mgmt_communication.py +++ /dev/null @@ -1,104 +0,0 @@ -# pylint: disable=line-too-long,useless-suppression -# 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 -import pytest - -import azure.mgmt.communication -from devtools_testutils import AzureMgmtTestCase, ResourceGroupPreparer -from azure.mgmt.communication.models import CommunicationServiceResource -from azure.mgmt.communication.models import KeyType -from azure.mgmt.communication.models import TaggedResource -from azure.mgmt.communication.models import RegenerateKeyParameters - -AZURE_LOCATION = "westus" -COMMUNICATION_SERVICE_LOCATION = "global" -COMMUNICATION_SERVICE_DATA_LOCATION = "UnitedStates" -DISABLE_MGMT_TESTS = True -DISABLE_REASON = "Temporary issue causing the tests to fail" - - -class MgmtCommunicationTest(AzureMgmtTestCase): - - def setUp(self): - super(MgmtCommunicationTest, self).setUp() - self.communication_client = self.create_mgmt_client( - azure.mgmt.communication.CommunicationServiceManagementClient - ) - - @pytest.mark.skipif(DISABLE_MGMT_TESTS, reason=DISABLE_REASON) - @ResourceGroupPreparer(location=AZURE_LOCATION) - def test_communication_crud(self, resource_group): - GROUP_NAME = resource_group.name - resource_name = self.get_resource_name("test-resource-crud") - - resource = CommunicationServiceResource( - location=COMMUNICATION_SERVICE_LOCATION, data_location=COMMUNICATION_SERVICE_DATA_LOCATION - ) - resource = self.communication_client.communication_service.begin_create_or_update( - GROUP_NAME, resource_name, resource - ).result() - - self.assertEqual(resource.name, resource_name) - self.assertEqual(resource.provisioning_state, "Succeeded") - self.assertIsNotNone(resource.immutable_resource_id) - self.assertEqual(resource.location, COMMUNICATION_SERVICE_LOCATION) - self.assertEqual(resource.data_location, COMMUNICATION_SERVICE_DATA_LOCATION) - self.assertIsNone(resource.notification_hub_id) - self.assertIsNone(resource.tags) - - resource = self.communication_client.communication_service.get(GROUP_NAME, resource_name) - self.assertEqual(resource.name, resource_name) - self.assertEqual(resource.provisioning_state, "Succeeded") - self.assertIsNotNone(resource.immutable_resource_id) - self.assertEqual(resource.location, COMMUNICATION_SERVICE_LOCATION) - self.assertEqual(resource.data_location, COMMUNICATION_SERVICE_DATA_LOCATION) - self.assertIsNone(resource.notification_hub_id) - self.assertIsNone(resource.tags) - - tags = {"tags": {"tag1": "tag1val", "tag2": "tag2val"}} - resource = self.communication_client.communication_service.update( - GROUP_NAME, resource_name, TaggedResource(**tags) - ) - self.assertEqual(resource.name, resource_name) - self.assertEqual(resource.provisioning_state, "Succeeded") - self.assertIsNotNone(resource.immutable_resource_id) - self.assertEqual(resource.location, COMMUNICATION_SERVICE_LOCATION) - self.assertEqual(resource.data_location, COMMUNICATION_SERVICE_DATA_LOCATION) - self.assertIsNone(resource.notification_hub_id) - self.assertIsNotNone(resource.tags) - self.assertEqual(resource.tags["tag1"], "tag1val") - self.assertEqual(resource.tags["tag2"], "tag2val") - - keys = self.communication_client.communication_service.list_keys(GROUP_NAME, resource_name) - self.assertIsNotNone(keys.primary_key) - self.assertIsNotNone(keys.secondary_key) - self.assertIsNotNone(keys.primary_connection_string) - self.assertIsNotNone(keys.secondary_connection_string) - - key_type = {"key_type": "Primary"} - keys_regenerated_primary = self.communication_client.communication_service.regenerate_key( - GROUP_NAME, resource_name, RegenerateKeyParameters(**key_type) - ) - self.assertNotEqual(keys.primary_key, keys_regenerated_primary.primary_key) - self.assertNotEqual(keys.primary_connection_string, keys_regenerated_primary.primary_connection_string) - - key_type = {"key_type": "Secondary"} - keys_regenerated_secondary = self.communication_client.communication_service.regenerate_key( - GROUP_NAME, resource_name, RegenerateKeyParameters(**key_type) - ) - self.assertNotEqual(keys.secondary_key, keys_regenerated_secondary.secondary_key) - self.assertNotEqual(keys.secondary_connection_string, keys_regenerated_secondary.secondary_connection_string) - - # Delete can take a long time to return when running live. Disable polling requirement until we can determine why. - self.communication_client.communication_service.begin_delete(GROUP_NAME, resource_name, polling=False) - - -# ------------------------------------------------------------------------------ -if __name__ == "__main__": - unittest.main() diff --git a/sdk/communication/azure-mgmt-communication/tests/disable_test_mgmt_communication_list_resources.py b/sdk/communication/azure-mgmt-communication/tests/disable_test_mgmt_communication_list_resources.py deleted file mode 100644 index 837bb23b568f..000000000000 --- a/sdk/communication/azure-mgmt-communication/tests/disable_test_mgmt_communication_list_resources.py +++ /dev/null @@ -1,98 +0,0 @@ -# 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 -import pytest - -import azure.mgmt.communication -from devtools_testutils import AzureMgmtTestCase, ResourceGroupPreparer -from azure.mgmt.communication.models import CommunicationServiceResource -from azure.mgmt.communication.models import KeyType -from azure.mgmt.communication.models import TaggedResource -from azure.mgmt.communication.models import RegenerateKeyParameters - -AZURE_LOCATION = "westus" -COMMUNICATION_SERVICE_LOCATION = "global" -COMMUNICATION_SERVICE_DATA_LOCATION = "UnitedStates" -DISABLE_MGMT_TESTS = True -DISABLE_REASON = "Temporary issue causing the tests to fail" - - -class MgmtCommunicationTest(AzureMgmtTestCase): - - def setUp(self): - super(MgmtCommunicationTest, self).setUp() - self.communication_client = self.create_mgmt_client( - azure.mgmt.communication.CommunicationServiceManagementClient - ) - - @pytest.mark.skipif(DISABLE_MGMT_TESTS, reason=DISABLE_REASON) - @ResourceGroupPreparer(location=AZURE_LOCATION) - def test_communication_list_by_subscription(self, resource_group): - GROUP_NAME = resource_group.name - resource_name = self.get_resource_name("test-resource-list-by-subscription") - - resource = CommunicationServiceResource( - location=COMMUNICATION_SERVICE_LOCATION, data_location=COMMUNICATION_SERVICE_DATA_LOCATION - ) - resource = self.communication_client.communication_service.begin_create_or_update( - GROUP_NAME, resource_name, resource - ).result() - - self.assertEqual(resource.name, resource_name) - self.assertEqual(resource.provisioning_state, "Succeeded") - self.assertIsNotNone(resource.immutable_resource_id) - self.assertEqual(resource.location, COMMUNICATION_SERVICE_LOCATION) - self.assertEqual(resource.data_location, COMMUNICATION_SERVICE_DATA_LOCATION) - self.assertIsNone(resource.notification_hub_id) - self.assertIsNone(resource.tags) - - resources = self.communication_client.communication_service.list_by_subscription() - self.assertIsNotNone(resources) - - # Verify that the resource we just created is in the list - resource_found = False - for resource in resources: - if resource.name == resource_name: - resource_found = True - self.assertTrue(resource_found) - - @pytest.mark.skipif(DISABLE_MGMT_TESTS, reason=DISABLE_REASON) - @ResourceGroupPreparer(location=AZURE_LOCATION) - def test_communication_list_by_rg(self, resource_group): - GROUP_NAME = resource_group.name - resource_name = self.get_resource_name("test-resource-list-by-rg") - - resource = CommunicationServiceResource( - location=COMMUNICATION_SERVICE_LOCATION, data_location=COMMUNICATION_SERVICE_DATA_LOCATION - ) - resource = self.communication_client.communication_service.begin_create_or_update( - GROUP_NAME, resource_name, resource - ).result() - - self.assertEqual(resource.name, resource_name) - self.assertEqual(resource.provisioning_state, "Succeeded") - self.assertIsNotNone(resource.immutable_resource_id) - self.assertEqual(resource.location, COMMUNICATION_SERVICE_LOCATION) - self.assertEqual(resource.data_location, COMMUNICATION_SERVICE_DATA_LOCATION) - self.assertIsNone(resource.notification_hub_id) - self.assertIsNone(resource.tags) - - resources = self.communication_client.communication_service.list_by_resource_group(GROUP_NAME) - self.assertIsNotNone(resources) - - # Verify that the resource we just created is in the list - resource_found = False - for resource in resources: - if resource.name == resource_name: - resource_found = True - self.assertTrue(resource_found) - - -# ------------------------------------------------------------------------------ -if __name__ == "__main__": - unittest.main() diff --git a/sdk/communication/azure-mgmt-communication/tests/disable_test_mgmt_communication_notification_hub.py b/sdk/communication/azure-mgmt-communication/tests/disable_test_mgmt_communication_notification_hub.py deleted file mode 100644 index ac0c104c8209..000000000000 --- a/sdk/communication/azure-mgmt-communication/tests/disable_test_mgmt_communication_notification_hub.py +++ /dev/null @@ -1,97 +0,0 @@ -# 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 -import time -import pytest - -import azure.mgmt.communication -import azure.mgmt.notificationhubs -from devtools_testutils import AzureMgmtTestCase, ResourceGroupPreparer -from azure.mgmt.communication.models import CommunicationServiceResource -from azure.mgmt.communication.models import KeyType -from azure.mgmt.communication.models import TaggedResource -from azure.mgmt.communication.models import RegenerateKeyParameters -from azure.mgmt.notificationhubs.models import SharedAccessAuthorizationRuleCreateOrUpdateParameters - -AZURE_LOCATION = "westus" -COMMUNICATION_SERVICE_LOCATION = "global" -COMMUNICATION_SERVICE_DATA_LOCATION = "UnitedStates" -DISABLE_MGMT_TESTS = True -DISABLE_REASON = "Temporary issue causing the tests to fail" - - -class MgmtCommunicationTest(AzureMgmtTestCase): - - def setUp(self): - super(MgmtCommunicationTest, self).setUp() - self.communication_client = self.create_mgmt_client( - azure.mgmt.communication.CommunicationServiceManagementClient - ) - self.notificationhubs_client = self.create_mgmt_client( - azure.mgmt.notificationhubs.NotificationHubsManagementClient - ) - - @pytest.mark.skipif(DISABLE_MGMT_TESTS, reason=DISABLE_REASON) - @ResourceGroupPreparer(location=AZURE_LOCATION) - def test_communication_link_notif_hub(self, resource_group): - GROUP_NAME = resource_group.name - namespace_name = self.get_resource_name("test-namespace-for-comm") - notification_hub_name = self.get_resource_name("test-notification-hub-for-comm") - resource_name = self.get_resource_name("test-resource-link-notif-hub") - - # Create the Notification Hubs resource that will be linked to the Communication Service resource - self.notificationhubs_client.namespaces.create_or_update( - GROUP_NAME, namespace_name, {"location": AZURE_LOCATION} - ) - namespace = self.notificationhubs_client.namespaces.get(GROUP_NAME, namespace_name) - while namespace.status == "Created": - if self.is_live == True: - time.sleep(30) - namespace = self.notificationhubs_client.namespaces.get(GROUP_NAME, namespace_name) - notification_hubs = self.notificationhubs_client.notification_hubs.create_or_update( - GROUP_NAME, namespace_name, notification_hub_name, {"location": AZURE_LOCATION} - ) - - # Create auth rule - authorization_rule = {"properties": {"rights": ["Listen"]}} - authorization_rule_name = "TestMgmtCommunicationLinkNotificationHub" - self.notificationhubs_client.notification_hubs.create_or_update_authorization_rule( - GROUP_NAME, - namespace_name, - notification_hub_name, - authorization_rule_name, - SharedAccessAuthorizationRuleCreateOrUpdateParameters(**authorization_rule), - ) - - # Obtain connection string - keys = self.notificationhubs_client.notification_hubs.list_keys( - GROUP_NAME, namespace_name, notification_hub_name, authorization_rule_name - ) - notification_hub_connection_string = keys.primary_connection_string - - # Create Communication Service resource for test - resource = CommunicationServiceResource( - location=COMMUNICATION_SERVICE_LOCATION, data_location=COMMUNICATION_SERVICE_DATA_LOCATION - ) - resource = self.communication_client.communication_service.begin_create_or_update( - GROUP_NAME, resource_name, resource - ).result() - - # Link Notification Hub's connection string to Communication Service - linked_notification_hub = self.communication_client.communication_service.link_notification_hub( - GROUP_NAME, - resource_name, - {"resource_id": notification_hubs.id, "connection_string": notification_hub_connection_string}, - ) - self.assertIsNotNone(linked_notification_hub.resource_id) - self.assertEqual(linked_notification_hub.resource_id, notification_hubs.id) - - -# ------------------------------------------------------------------------------ -if __name__ == "__main__": - unittest.main()