mirror/github-octodns
1
0
Fork 0
mirror of https://github.com/github/octodns.git synced 2024-05-11 05:55:00 +00:00
Code Issues Releases Activity

Add tests for long txt record, test alias entries

Browse Source
This commit is contained in:
Piotr Pieprzycki
2020-08-11 04:47:05 -04:00
parent 450cbe0208
commit b4da48b860
2 changed files with 53 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
octodns/provider/azuredns.py
View File

@@ -263,6 +263,13 @@ def _parse_azure_type(string):
return string.split('/')[len(string.split('/')) - 1]
def _check_for_alias(azrecord):
if (azrecord.target_resource.id and not azrecord.arecords and not
azrecord.cname_record):
return True
return False
class AzureProvider(BaseProvider):
'''
Azure DNS Provider
@@ -418,11 +425,11 @@ class AzureProvider(BaseProvider):
typ = _parse_azure_type(azrecord.type)
if typ in ['A', 'CNAME']:
if self._check_for_alias(azrecord, typ):
if _check_for_alias(azrecord):
self.log.debug(
'Skipping - ALIAS. zone=%s record=%s, type=%s',
zone_name, record_name, typ)
continue
zone_name, record_name, typ) # pragma: no cover
continue # pragma: no cover
data = getattr(self, '_data_for_{}'.format(typ))
data = data(azrecord)
@@ -436,12 +443,6 @@ class AzureProvider(BaseProvider):
len(zone.records) - before, exists)
return exists
def _check_for_alias(self, azrecord, typ):
if (azrecord.target_resource.id and not azrecord.arecords and not
azrecord.arecords and not azrecord.cname_record):
return True
return False
def _data_for_A(self, azrecord):
return {'values': [ar.ipv4_address for ar in azrecord.arecords]}

47
tests/test_octodns_provider_azuredns.py
View File

@@ -7,7 +7,7 @@ from __future__ import absolute_import, division, print_function, \
from octodns.record import Create, Delete, Record
from octodns.provider.azuredns import _AzureRecord, AzureProvider, \
_check_endswith_dot, _parse_azure_type
_check_endswith_dot, _parse_azure_type, _check_for_alias
from octodns.zone import Zone
from octodns.provider.base import Plan
@@ -134,6 +134,18 @@ octo_records.append(Record.new(zone, 'txt2', {
'type': 'TXT',
'values': ['txt multiple test', 'txt multiple test 2']}))
long_txt = "v=spf1 ip4:10.10.0.0/24 ip4:10.10.1.0/24 ip4:10.10.2.0/24"
long_txt += " ip4:10.10.3.0/24 ip4:10.10.4.0/24 ip4:10.10.5.0/24 "
long_txt += " 10.6.0/24 ip4:10.10.7.0/24 ip4:10.10.8.0/24 "
long_txt += " ip4:10.10.10.0/24 ip4:10.10.11.0/24 ip4:10.10.12.0/24"
long_txt += " ip4:10.10.13.0/24 ip4:10.10.14.0/24 ip4:10.10.15.0/24"
long_txt += " ip4:10.10.16.0/24 ip4:10.10.17.0/24 ip4:10.10.18.0/24"
long_txt += " ip4:10.10.19.0/24 ip4:10.10.20.0/24 ~all"
octo_records.append(Record.new(zone, 'txt3', {
'ttl': 10,
'type': 'TXT',
'values': ['txt multiple test', long_txt]}))
azure_records = []
_base0 = _AzureRecord('TestAzure', octo_records[0])
_base0.zone_name = 'unit.tests'
@@ -306,6 +318,22 @@ _base17.params['txt_records'] = [TxtRecord(value=['txt multiple test']),
TxtRecord(value=['txt multiple test 2'])]
azure_records.append(_base17)
long_txt_az1 = "v=spf1 ip4:10.10.0.0/24 ip4:10.10.1.0/24 ip4:10.10.2.0/24"
long_txt_az1 += " ip4:10.10.3.0/24 ip4:10.10.4.0/24 ip4:10.10.5.0/24 "
long_txt_az1 += " 10.6.0/24 ip4:10.10.7.0/24 ip4:10.10.8.0/24 "
long_txt_az1 += " ip4:10.10.10.0/24 ip4:10.10.11.0/24 ip4:10.10.12.0/24"
long_txt_az1 += " ip4:10.10.13.0/24 ip4:10.10.14.0/24 ip4:10.10."
long_txt_az2 = "15.0/24 ip4:10.10.16.0/24 ip4:10.10.17.0/24 ip4:10.10.18.0/24"
long_txt_az2 += " ip4:10.10.19.0/24 ip4:10.10.20.0/24 ~all"
_base18 = _AzureRecord('TestAzure', octo_records[18])
_base18.zone_name = 'unit.tests'
_base18.relative_record_set_name = 'txt3'
_base18.record_type = 'TXT'
_base18.params['ttl'] = 10
_base18.params['txt_records'] = [TxtRecord(value=['txt multiple test']),
TxtRecord(value=[long_txt_az1, long_txt_az2])]
azure_records.append(_base18)
class Test_AzureRecord(TestCase):
def test_azure_record(self):
@@ -333,6 +361,17 @@ class Test_CheckEndswithDot(TestCase):
self.assertEquals(expected, _check_endswith_dot(test))
class Test_CheckAzureAlias(TestCase):
def test_check_for_alias(self):
alias_record = type('C', (object,), {})
alias_record.target_resource = type('C', (object,), {})
alias_record.target_resource.id = "/subscriptions/x/resourceGroups/y/z"
alias_record.arecords = None
alias_record.cname_record = None
self.assertEquals(_check_for_alias(alias_record), True)
class TestAzureDnsProvider(TestCase):
def _provider(self):
return self._get_provider('mock_spc', 'mock_dns_client')
@@ -503,9 +542,9 @@ class TestAzureDnsProvider(TestCase):
changes.append(Create(i))
deletes.append(Delete(i))
self.assertEquals(18, provider.apply(Plan(None, zone,
self.assertEquals(19, provider.apply(Plan(None, zone,
changes, True)))
self.assertEquals(18, provider.apply(Plan(zone, zone,
self.assertEquals(19, provider.apply(Plan(zone, zone,
deletes, True)))
def test_create_zone(self):
@@ -521,7 +560,7 @@ class TestAzureDnsProvider(TestCase):
_get = provider._dns_client.zones.get
_get.side_effect = CloudError(Mock(status=404), err_msg)
self.assertEquals(18, provider.apply(Plan(None, desired, changes,
self.assertEquals(19, provider.apply(Plan(None, desired, changes,
True)))
def test_check_zone_no_create(self):