# 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 : 26 # Methods Covered : 26 # Examples Total : 26 # Examples Tested : 20 # Coverage % : 77 # ---------------------- # Current Operation Coverage: # Operations: 1/1 # Redis: 10/13 # FirewallRules: 4/4 # PatchSchedules: 4/4 # LinkedServer: 1/4 import time import unittest import azure.mgmt.redis from devtools_testutils import AzureMgmtTestCase, RandomNameResourceGroupPreparer AZURE_LOCATION = 'eastus' class MgmtRedisTest(AzureMgmtTestCase): def setUp(self): super(MgmtRedisTest, self).setUp() self.mgmt_client = self.create_mgmt_client( azure.mgmt.redis.RedisManagementClient ) if self.is_live: from azure.mgmt.network import NetworkManagementClient self.network_client = self.create_mgmt_client( NetworkManagementClient ) def create_virtual_network(self, group_name, location, network_name, subnet_name): azure_operation_poller = self.network_client.virtual_networks.begin_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.begin_create_or_update( group_name, network_name, subnet_name, {'address_prefix': '10.0.0.0/24'} ) subnet_info = async_subnet_creation.result() return subnet_info @unittest.skip('hard to test') @RandomNameResourceGroupPreparer(location=AZURE_LOCATION) def test_redis(self, resource_group): # UNIQUE = resource_group.name[-4:] SUBSCRIPTION_ID = self.settings.SUBSCRIPTION_ID TENANT_ID = self.settings.TENANT_ID RESOURCE_GROUP = resource_group.name NAME = "myosgkxy" VIRTUAL_NETWORK_NAME = "myVirtualNetwork" SUBNET_NAME = "mySubnet" # CACHE_NAME = "myCache" CACHE_NAME = NAME RULE_NAME = "myRule" DEFAULT = "default" LINKED_SERVER_NAME = "myLinkedServer" REDIS_NAME = "myRedis" if self.is_live: self.create_virtual_network(RESOURCE_GROUP, AZURE_LOCATION, VIRTUAL_NETWORK_NAME, SUBNET_NAME) #-------------------------------------------------------------------------- # /Redis/put/RedisCacheCreate[put] #-------------------------------------------------------------------------- BODY = { "location": AZURE_LOCATION, "zones": [ "1" ], "sku": { "name": "Premium", "family": "P", "capacity": "1" }, "enable_non_ssl_port": True, "shard_count": "2", # "replicas_per_master": "2", "redis_configuration": { "maxmemory-policy": "allkeys-lru" }, "subnet_id": "/subscriptions/" + SUBSCRIPTION_ID + "/resourceGroups/" + RESOURCE_GROUP + "/providers/Microsoft.Network/virtualNetworks/" + VIRTUAL_NETWORK_NAME + "/subnets/" + SUBNET_NAME, "static_ip": "10.0.0.5", "minimum_tls_version": "1.2" } result = self.mgmt_client.redis.begin_create(resource_group_name=RESOURCE_GROUP, name=NAME, parameters=BODY) result = result.result() #-------------------------------------------------------------------------- # /PatchSchedules/put/RedisCachePatchSchedulesCreateOrUpdate[put] #-------------------------------------------------------------------------- BODY = { "schedule_entries": [ { "day_of_week": "Monday", "start_hour_utc": "12", "maintenance_window": "PT5H" }, { "day_of_week": "Tuesday", "start_hour_utc": "12" } ] } result = self.mgmt_client.patch_schedules.create_or_update(resource_group_name=RESOURCE_GROUP, name=NAME, default=DEFAULT, parameters=BODY) if self.is_live: time.sleep(1800) #-------------------------------------------------------------------------- # /FirewallRules/put/RedisCacheFirewallRuleCreate[put] #-------------------------------------------------------------------------- BODY = { "start_ip": "10.0.1.1", "end_ip": "10.0.1.4" } result = self.mgmt_client.firewall_rules.create_or_update(resource_group_name=RESOURCE_GROUP, cache_name=CACHE_NAME, rule_name=RULE_NAME, parameters=BODY) #-------------------------------------------------------------------------- # /LinkedServer/put/LinkedServer_Create[put] #-------------------------------------------------------------------------- BODY = { "linked_redis_cache_id": "/subscriptions/" + SUBSCRIPTION_ID + "/resourceGroups/" + RESOURCE_GROUP + "/providers/Microsoft.Cache/Redis/" + REDIS_NAME, "linked_redis_cache_location": "West US", "server_role": "Secondary" } # result = self.mgmt_client.linked_server.begin_create(resource_group_name=RESOURCE_GROUP, name=NAME, linked_server_name=LINKED_SERVER_NAME, parameters=BODY) # result = result.result() #-------------------------------------------------------------------------- # /LinkedServer/get/LinkedServer_Get[get] #-------------------------------------------------------------------------- # result = self.mgmt_client.linked_server.get(resource_group_name=RESOURCE_GROUP, name=NAME, linked_server_name=LINKED_SERVER_NAME) #-------------------------------------------------------------------------- # /FirewallRules/get/RedisCacheFirewallRuleGet[get] #-------------------------------------------------------------------------- result = self.mgmt_client.firewall_rules.get(resource_group_name=RESOURCE_GROUP, cache_name=CACHE_NAME, rule_name=RULE_NAME) #-------------------------------------------------------------------------- # /PatchSchedules/get/RedisCachePatchSchedulesGet[get] #-------------------------------------------------------------------------- # result = self.mgmt_client.patch_schedules.get(resource_group_name=RESOURCE_GROUP, name=NAME, default=DEFAULT) #-------------------------------------------------------------------------- # /Redis/get/RedisCacheGet[get] #-------------------------------------------------------------------------- # result = self.mgmt_client.redis.list_upgrade_notifications(resource_group_name=RESOURCE_GROUP, name=NAME, history="5000") #-------------------------------------------------------------------------- # /PatchSchedules/get/RedisCachePatchSchedulesList[get] #-------------------------------------------------------------------------- result = self.mgmt_client.patch_schedules.list_by_redis_resource(resource_group_name=RESOURCE_GROUP, cache_name=CACHE_NAME) #-------------------------------------------------------------------------- # /FirewallRules/get/RedisCacheFirewallRulesList[get] #-------------------------------------------------------------------------- result = self.mgmt_client.firewall_rules.list_by_redis_resource(resource_group_name=RESOURCE_GROUP, cache_name=CACHE_NAME) #-------------------------------------------------------------------------- # /LinkedServer/get/LinkedServer_List[get] #-------------------------------------------------------------------------- result = self.mgmt_client.linked_server.list(resource_group_name=RESOURCE_GROUP, name=NAME) #-------------------------------------------------------------------------- # /Redis/get/RedisCacheGet[get] #-------------------------------------------------------------------------- result = self.mgmt_client.redis.get(resource_group_name=RESOURCE_GROUP, name=NAME) #-------------------------------------------------------------------------- # /Redis/get/RedisCacheListByResourceGroup[get] #-------------------------------------------------------------------------- result = self.mgmt_client.redis.list_by_resource_group(resource_group_name=RESOURCE_GROUP) #-------------------------------------------------------------------------- # /Redis/get/RedisCacheList[get] #-------------------------------------------------------------------------- result = self.mgmt_client.redis.list() #-------------------------------------------------------------------------- # /Operations/get/Operations_List[get] #-------------------------------------------------------------------------- result = self.mgmt_client.operations.list() #-------------------------------------------------------------------------- # /Redis/post/RedisCacheRegenerateKey[post] #-------------------------------------------------------------------------- BODY = { "key_type": "Primary" } result = self.mgmt_client.redis.regenerate_key(resource_group_name=RESOURCE_GROUP, name=NAME, parameters=BODY) #-------------------------------------------------------------------------- # /Redis/post/RedisCacheForceReboot[post] #-------------------------------------------------------------------------- BODY = { "shard_id": "0", "reboot_type": "AllNodes" } result = self.mgmt_client.redis.force_reboot(resource_group_name=RESOURCE_GROUP, name=NAME, parameters=BODY) #-------------------------------------------------------------------------- # /Redis/post/RedisCacheListKeys[post] #-------------------------------------------------------------------------- result = self.mgmt_client.redis.list_keys(resource_group_name=RESOURCE_GROUP, name=NAME) #-------------------------------------------------------------------------- # /Redis/post/RedisCacheImport[post] #-------------------------------------------------------------------------- BODY = { "format": "RDB", "files": [ "http://fileuris.contoso.com/pathtofile1" ] } # result = self.mgmt_client.redis.begin_import_data(resource_group_name=RESOURCE_GROUP, name=NAME, parameters=BODY) # result = result.result() #-------------------------------------------------------------------------- # /Redis/post/RedisCacheExport[post] #-------------------------------------------------------------------------- BODY = { "format": "RDB", "prefix": "datadump1", "container": "https://contosostorage.blob.core.window.net/urltoBlobContainer?sasKeyParameters" } # result = self.mgmt_client.redis.begin_export_data(resource_group_name=RESOURCE_GROUP, name=NAME, parameters=BODY) # result = result.result() #-------------------------------------------------------------------------- # /Redis/patch/RedisCacheUpdate[patch] #-------------------------------------------------------------------------- BODY = { "enable_non_ssl_port": True } result = self.mgmt_client.redis.update(resource_group_name=RESOURCE_GROUP, name=NAME, parameters=BODY) #-------------------------------------------------------------------------- # /Redis/post/RedisCacheList[post] #-------------------------------------------------------------------------- BODY = { "type": "Microsoft.Cache/Redis", "name": "cacheName" } result = self.mgmt_client.redis.check_name_availability(parameters=BODY) #-------------------------------------------------------------------------- # /LinkedServer/delete/LinkedServerDelete[delete] #-------------------------------------------------------------------------- # result = self.mgmt_client.linked_server.delete(resource_group_name=RESOURCE_GROUP, name=NAME, linked_server_name=LINKED_SERVER_NAME) #-------------------------------------------------------------------------- # /FirewallRules/delete/RedisCacheFirewallRuleDelete[delete] #-------------------------------------------------------------------------- result = self.mgmt_client.firewall_rules.delete(resource_group_name=RESOURCE_GROUP, cache_name=CACHE_NAME, rule_name=RULE_NAME) #-------------------------------------------------------------------------- # /PatchSchedules/delete/RedisCachePatchSchedulesDelete[delete] #-------------------------------------------------------------------------- result = self.mgmt_client.patch_schedules.delete(resource_group_name=RESOURCE_GROUP, name=NAME, default=DEFAULT) #-------------------------------------------------------------------------- # /Redis/delete/RedisCacheDelete[delete] #-------------------------------------------------------------------------- result = self.mgmt_client.redis.begin_delete(resource_group_name=RESOURCE_GROUP, name=NAME) result = result.result() #------------------------------------------------------------------------------ if __name__ == '__main__': unittest.main()