# 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 azure.mgmt.dns import azure.mgmt.network from devtools_testutils import ( AzureMgmtRecordedTestCase, ResourceGroupPreparer, recorded_by_proxy, AzureMgmtPreparer, FakeResource ) import pytest @pytest.mark.live_test_only class VirtualNetworkPreparer(AzureMgmtPreparer): def __init__(self, name_prefix='pvtzonevnet'): super(VirtualNetworkPreparer, self).__init__(name_prefix, 24) def create_resource(self, name, **kwargs): registration_virtual_network_name = name + 'reg' resolution_virtual_network_name = name + 'res' if self.is_live: resource_group_name = kwargs['resource_group'].name location_name = kwargs['location'] registration_network = self.test_class_instance.network_client.virtual_networks.begin_create_or_update( resource_group_name, registration_virtual_network_name, { 'location': location_name, 'address_space': { 'address_prefixes': ['10.0.0.0/8'] }, 'subnets': [ { 'name': 'default', 'address_prefix': '10.0.0.0/24' } ] } ).result() resolution_network = self.test_class_instance.network_client.virtual_networks.begin_create_or_update( resource_group_name, resolution_virtual_network_name, { 'location': location_name, 'address_space': { 'address_prefixes': ['10.0.0.0/8'] }, 'subnets': [ { 'name': 'default', 'address_prefix': '10.0.0.0/24' } ] } ).result() else: registration_network = FakeResource(name=registration_virtual_network_name, id='') resolution_network = FakeResource(name=resolution_virtual_network_name, id='') return { 'registration_virtual_network': registration_network, 'resolution_virtual_network': resolution_network } @pytest.mark.live_test_only class TestMgmtDns(AzureMgmtRecordedTestCase): def setup_method(self, method): self.dns_client = self.create_mgmt_client( azure.mgmt.dns.DnsManagementClient, ) self.network_client = self.create_mgmt_client( azure.mgmt.network.NetworkManagementClient, ) @ResourceGroupPreparer() @recorded_by_proxy def test_public_zone(self, resource_group, location): zone_name = self.get_resource_name('pydns.com') # Zones are a 'global' resource. zone = self.dns_client.zones.create_or_update( resource_group.name, zone_name, { 'zone_type': 'Public', 'location': 'global' } ) assert zone.name == zone_name zone = self.dns_client.zones.get( resource_group.name, zone.name ) assert zone.name == zone_name zones = list(self.dns_client.zones.list_by_resource_group( resource_group.name )) assert len(zones) >= 1 zones = list(self.dns_client.zones.list()) assert len(zones) >= 1 # Record set record_set_name = self.get_resource_name('record_set') self.dns_client.record_sets.create_or_update( resource_group.name, zone.name, record_set_name, 'A', { "ttl": 300, "arecords": [ { "ipv4_address": "1.2.3.4" } ] } ) self.dns_client.record_sets.update( resource_group.name, zone.name, record_set_name, 'A', { "ttl": 300, "arecords": [ { "ipv4_address": "1.2.3.4" }, { "ipv4_address": "5.6.7.8" } ] } ) record_set = self.dns_client.record_sets.get( resource_group.name, zone.name, record_set_name, 'A' ) assert record_set is not None record_sets = list(self.dns_client.record_sets.list_by_type( resource_group.name, zone.name, 'A' )) assert len(record_sets) == 1 # NS and SOA records are created by default in public zones. record_sets = list(self.dns_client.record_sets.list_by_dns_zone( resource_group.name, zone.name )) assert len(record_sets) == 3 self.dns_client.record_sets.delete( resource_group.name, zone.name, record_set_name, 'A' ) async_delete = self.dns_client.zones.begin_delete( resource_group.name, zone.name ) async_delete.wait() @unittest.skip("using this API is no longer allowed") @ResourceGroupPreparer() @VirtualNetworkPreparer() @recorded_by_proxy def test_private_zone(self, resource_group, location, registration_virtual_network, resolution_virtual_network): zone_name = self.get_resource_name('pydns.com') # Zones are a 'global' resource. zone = self.dns_client.zones.create_or_update( resource_group.name, zone_name, { 'zone_type': 'Private', 'location': 'global', 'registration_virtual_networks': [ { 'id': registration_virtual_network.id } ], 'resolution_virtual_networks': [ { 'id': resolution_virtual_network.id } ] } ) assert zone.name == zone_name zone = self.dns_client.zones.get( resource_group.name, zone.name ) assert zone.name == zone_name zones = list(self.dns_client.zones.list_by_resource_group( resource_group.name )) assert len(zones) >= 1 zones = list(self.dns_client.zones.list()) assert len(zones) >= 1 # Record set record_set_name = self.get_resource_name('record_set') self.dns_client.record_sets.create_or_update( resource_group.name, zone.name, record_set_name, 'A', { "ttl": 300, "arecords": [ { "ipv4_address": "1.2.3.4" } ] } ) self.dns_client.record_sets.update( resource_group.name, zone.name, record_set_name, 'A', { "ttl": 300, "arecords": [ { "ipv4_address": "1.2.3.4" }, { "ipv4_address": "5.6.7.8" } ] } ) record_set = self.dns_client.record_sets.get( resource_group.name, zone.name, record_set_name, 'A' ) assert record_set is not None record_sets = list(self.dns_client.record_sets.list_by_type( resource_group.name, zone.name, 'A' )) assert len(record_sets) == 1 # SOA record is created by default in private zones. record_sets = list(self.dns_client.record_sets.list_by_dns_zone( resource_group.name, zone.name )) assert len(record_sets) == 2 self.dns_client.record_sets.delete( resource_group.name, zone.name, record_set_name, 'A' ) async_delete = self.dns_client.zones.begin_delete( resource_group.name, zone.name ) async_delete.wait() #------------------------------------------------------------------------------ if __name__ == '__main__': unittest.main()