# 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. #-------------------------------------------------------------------------- # TEST SCENARIO COVERAGE # ---------------------- # Methods Total : 8 # Methods Covered : 8 # Examples Total : 8 # Examples Tested : 8 # Coverage % : 100 # ---------------------- import unittest import azure.mgmt.redhatopenshift from devtools_testutils import AzureMgmtTestCase, ResourceGroupPreparer AZURE_LOCATION = 'australiaeast' @unittest.skip("skip test") class MgmtAzureRedHatOpenShiftClientTest(AzureMgmtTestCase): def setUp(self): super(MgmtAzureRedHatOpenShiftClientTest, self).setUp() self.mgmt_client = self.create_mgmt_client( azure.mgmt.redhatopenshift.AzureRedHatOpenShiftClient ) if self.is_live: from azure.mgmt.network import NetworkManagementClient self.network_client = self.create_mgmt_client( NetworkManagementClient ) from azure.mgmt.authorization import AuthorizationManagementClient self.authorization_client = self.create_mgmt_client( AuthorizationManagementClient ) from azure.mgmt.containerregistry import ContainerRegistryManagementClient self.acr_client = self.create_mgmt_client( ContainerRegistryManagementClient ) def create_virtual_network(self, group_name, location, network_name, subnet_name): azure_operation_poller = self.network_client.virtual_networks.create_or_update( group_name, network_name, { 'location': location, 'address_space': { 'address_prefixes': ['10.0.0.0/16'] } }, ) result_create = azure_operation_poller.result() async_subnet_creation = self.network_client.subnets.create_or_update( group_name, network_name, subnet_name, subnet_parameters={'address_prefix': '10.0.0.0/24', "private_link_service_network_policies": "Disabled", "private_endpoint_network_policies": "Disabled", "service_endpoints": [ { "service": "Microsoft.ContainerRegistry" } ] } ) subnet_info = async_subnet_creation.result() return subnet_info def create_subnet(self, group_name, location, network_name, subnet_name): async_subnet_creation = self.network_client.subnets.create_or_update( group_name, network_name, subnet_name, subnet_parameters={'address_prefix': '10.0.1.0/24', "private_link_service_network_policies": "Disabled", "private_endpoint_network_policies": "Disabled", "service_endpoints": [ { "service": "Microsoft.ContainerRegistry" } ] } ) subnet_info = async_subnet_creation.result() subnet_info = self.network_client.subnets.get(group_name, network_name, subnet_name) print(str(subnet_info)) return subnet_info def assign_role(self, service_principal_id, scope, name, full_id): BODY = { "role_definition_id": full_id, "principal_id": service_principal_id, "principal_type": "ServicePrincipal" } result = self.authorization_client.role_assignments.create(scope, role_assignment_name=name, parameters=BODY) @ResourceGroupPreparer(location=AZURE_LOCATION) def test_redhatopenshift(self, resource_group): SUBSCRIPTION_ID = self.settings.SUBSCRIPTION_ID TENANT_ID = self.settings.TENANT_ID RESOURCE_GROUP = resource_group.name RESOURCE_NAME = "zimsclusterxx" VIRTUAL_NETWORK_NAME = "myvirtualnetwork" SUBNET_NAME = "mysubnet" SUBNET_NAME_2 = "mysubnet2" if self.is_live: SUBNET = self.create_virtual_network(RESOURCE_GROUP, AZURE_LOCATION, VIRTUAL_NETWORK_NAME, SUBNET_NAME) SUBNET_2 = self.create_subnet(RESOURCE_GROUP, AZURE_LOCATION, VIRTUAL_NETWORK_NAME, SUBNET_NAME_2) self.assign_role(self.settings.SERVICE_PRINCIPAL_ID, # SP Object ID "/subscriptions/" + SUBSCRIPTION_ID + "/resourceGroups/" + RESOURCE_GROUP + "/providers/Microsoft.Network/virtualNetworks/" + VIRTUAL_NETWORK_NAME, "1fa638dc-b769-420d-b822-340abb216e78", "/subscriptions/" + SUBSCRIPTION_ID + "/providers/Microsoft.Authorization/roleDefinitions/" + "b24988ac-6180-42a0-ab88-20f7382dd24c") self.assign_role(self.settings.ARO_SERVICE_PRINCIPAL_ID, "/subscriptions/" + SUBSCRIPTION_ID + "/resourceGroups/" + RESOURCE_GROUP + "/providers/Microsoft.Network/virtualNetworks/" + VIRTUAL_NETWORK_NAME, "1fa638dc-b769-420d-b822-340abb216e77", "/subscriptions/" + SUBSCRIPTION_ID + "/providers/Microsoft.Authorization/roleDefinitions/" + "b24988ac-6180-42a0-ab88-20f7382dd24c") CLIENT_ID = self.settings.CLIENT_ID CLIENT_SECRET = self.settings.CLIENT_SECRET self.be_careful_with_service_principal else: CLIENT_ID = "00000000-0000-0000-0000-000000000000" CLIENT_SECRET = "xxxxxxxx" # /OpenShiftClusters/put/Creates or updates a OpenShift cluster with the specified subscription, resource group and resource name.[put] BODY = { "location": "australiaeast", "tags": { "key": "value" }, "cluster_profile": { "pull_secret": "", "domain": "ab0176mx", "resource_group_id": "/subscriptions/" + SUBSCRIPTION_ID + "/resourceGroups/" + "aro-ab0176mx" }, "service_principal_profile": { "client_id": CLIENT_ID, "client_secret": CLIENT_SECRET }, "network_profile": { "pod_cidr": "10.128.0.0/14", "service_cidr": "172.30.0.0/16" }, "master_profile": { "vm_size": "Standard_D8s_v3", "subnet_id": "/subscriptions/" + SUBSCRIPTION_ID + "/resourceGroups/" + RESOURCE_GROUP + "/providers/Microsoft.Network/virtualNetworks/" + VIRTUAL_NETWORK_NAME + "/subnets/" + SUBNET_NAME + "" }, "worker_profiles": [ { "name": "worker", "vm_size": "Standard_D4s_v3", "disk_size_gb": "128", "subnet_id": "/subscriptions/" + SUBSCRIPTION_ID + "/resourceGroups/" + RESOURCE_GROUP + "/providers/Microsoft.Network/virtualNetworks/" + VIRTUAL_NETWORK_NAME + "/subnets/" + SUBNET_NAME_2 + "", "count": "3" } ], "apiserver_profile": { "visibility": "Public" }, "ingress_profiles": [ { "name": "default", "visibility": "Public" } ] } result = self.mgmt_client.open_shift_clusters.create_or_update(resource_group_name=RESOURCE_GROUP, resource_name=RESOURCE_NAME, parameters=BODY) result = result.result() # /OpenShiftClusters/get/Gets a OpenShift cluster with the specified subscription, resource group and resource name.[get] result = self.mgmt_client.open_shift_clusters.get(resource_group_name=RESOURCE_GROUP, resource_name=RESOURCE_NAME) # /OpenShiftClusters/get/Lists OpenShift clusters in the specified subscription and resource group.[get] result = self.mgmt_client.open_shift_clusters.list_by_resource_group(resource_group_name=RESOURCE_GROUP) # /OpenShiftClusters/get/Lists OpenShift clusters in the specified subscription.[get] result = self.mgmt_client.open_shift_clusters.list() # /Operations/get/Lists all of the available RP operations.[get] result = self.mgmt_client.operations.list() # /OpenShiftClusters/post/Lists credentials of an OpenShift cluster with the specified subscription, resource group and resource name.[post] result = self.mgmt_client.open_shift_clusters.list_credentials(resource_group_name=RESOURCE_GROUP, resource_name=RESOURCE_NAME) # /OpenShiftClusters/patch/Creates or updates a OpenShift cluster with the specified subscription, resource group and resource name.[patch] BODY = { "tags": { "key": "value" }, "cluster_profile": { "pull_secret": "", "domain": "ab0176mx", "resource_group_id": "/subscriptions/" + SUBSCRIPTION_ID + "/resourceGroups/" + "aro-ab0176mx" }, "service_principal_profile": { "client_id": CLIENT_ID, "client_secret": CLIENT_SECRET }, "network_profile": { "pod_cidr": "10.128.0.0/14", "service_cidr": "172.30.0.0/16" }, "master_profile": { "vm_size": "Standard_D8s_v3", "subnet_id": "/subscriptions/" + SUBSCRIPTION_ID + "/resourceGroups/" + RESOURCE_GROUP + "/providers/Microsoft.Network/virtualNetworks/" + VIRTUAL_NETWORK_NAME + "/subnets/" + SUBNET_NAME + "" }, "worker_profiles": [ { # "name": "worker", "vm_size": "Standard_D4s_v3", "disk_size_gb": "128", "subnet_id": "/subscriptions/" + SUBSCRIPTION_ID + "/resourceGroups/" + RESOURCE_GROUP + "/providers/Microsoft.Network/virtualNetworks/" + VIRTUAL_NETWORK_NAME + "/subnets/" + SUBNET_NAME_2 + "", "count": "3" } ], "apiserver_profile": { "visibility": "Public" }, "ingress_profiles": [ { "name": "default", "visibility": "Public" } ] } result = self.mgmt_client.open_shift_clusters.update(resource_group_name=RESOURCE_GROUP, resource_name=RESOURCE_NAME, parameters=BODY) result = result.result() # /OpenShiftClusters/delete/Deletes a OpenShift cluster with the specified subscription, resource group and resource name.[delete] result = self.mgmt_client.open_shift_clusters.delete(resource_group_name=RESOURCE_GROUP, resource_name=RESOURCE_NAME) result = result.result() #------------------------------------------------------------------------------ if __name__ == '__main__': unittest.main()