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

Implement and test Route53Provider.process_desired_zone checking of NA-CA-*

Browse Source
This commit is contained in:
Ross McFarland
2021-08-17 20:26:51 -07:00
parent 536c0c68ec
commit c81450682c
3 changed files with 182 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
octodns/provider/base.py
View File

@@ -12,6 +12,10 @@ from ..zone import Zone
from .plan import Plan
class ProviderException(Exception):
pass
class BaseProvider(BaseSource):
def __init__(self, id, apply_disabled=False,
@@ -44,6 +48,12 @@ class BaseProvider(BaseSource):
'''
return []
def supports_warn_or_except(self, msg):
# TODO: base class param to control warn vs except
if False:
raise ProviderException(msg)
self.log.warning(msg)
def process_desired_zone(self, desired):
return desired

39
octodns/provider/route53.py
View File

@@ -19,6 +19,7 @@ from six import text_type
from ..equality import EqualityTupleMixin
from ..record import Record, Update
from ..record.geo import GeoCodes
from ..zone import Zone
from .base import BaseProvider
octal_re = re.compile(r'\\(\d\d\d)')
@@ -924,6 +925,44 @@ class Route53Provider(BaseProvider):
return data
def process_desired_zone(self, desired):
ret = Zone(desired.name, desired.sub_zones)
for record in desired.records:
if getattr(record, 'dynamic', False):
# Make a copy of the record in case we have to muck with it
record = record.copy()
from pprint import pprint
pprint((record, record.data))
dynamic = record.dynamic
rules = []
for i, rule in enumerate(dynamic.rules):
geos = rule.data.get('geos', [])
if not geos:
rules.append(rule)
continue
filtered_geos = [g for g in geos
if not g.startswith('NA-CA-')]
if not filtered_geos:
# We've removed all geos, we'll have to skip this rule
msg = 'NA-CA-* not supported resulting in ' \
'empty geo target, skipping rule {}'.format(i)
self.supports_warn_or_except(msg)
continue
elif geos != filtered_geos:
msg = 'NA-CA-* not supported resulting in ' \
'empty geo target, skipping rule {}'.format(i)
self.supports_warn_or_except(msg)
rule.data['geos'] = filtered_geos
rules.append(rule)
dynamic.rules = rules
pprint((record, record.data))
ret.add_record(record)
return super(Route53Provider, self).process_desired_zone(ret)
def populate(self, zone, target=False, lenient=False):
self.log.debug('populate: name=%s, target=%s, lenient=%s', zone.name,
target, lenient)

133
tests/test_octodns_provider_route53.py
View File

@@ -394,6 +394,139 @@ class TestRoute53Provider(TestCase):
return (provider, stubber)
def test_process_desired_zone(self):
provider, stubber = self._get_stubbed_fallback_auth_provider()
# No records, essentially a no-op
desired = Zone('unit.tests.', [])
got = provider.process_desired_zone(desired)
self.assertEquals(desired.records, got.records)
# Record without any geos
desired = Zone('unit.tests.', [])
record = Record.new(desired, 'a', {
'ttl': 30,
'type': 'A',
'value': '1.2.3.4',
'dynamic': {
'pools': {
'one': {
'values': [{
'value': '2.2.3.4',
}],
},
},
'rules': [{
'pool': 'one',
}],
},
})
desired.add_record(record)
got = provider.process_desired_zone(desired)
self.assertEquals(desired.records, got.records)
self.assertEquals(1, len(list(got.records)[0].dynamic.rules))
self.assertFalse('geos' in list(got.records)[0].dynamic.rules[0].data)
# Record where all geos are supported
desired = Zone('unit.tests.', [])
record = Record.new(desired, 'a', {
'ttl': 30,
'type': 'A',
'value': '1.2.3.4',
'dynamic': {
'pools': {
'one': {
'values': [{
'value': '1.2.3.4',
}],
},
'two': {
'values': [{
'value': '2.2.3.4',
}],
},
},
'rules': [{
'geos': ['EU', 'NA-US-OR'],
'pool': 'two',
}, {
'pool': 'one',
}],
},
})
desired.add_record(record)
got = provider.process_desired_zone(desired)
self.assertEquals(2, len(list(got.records)[0].dynamic.rules))
self.assertEquals(['EU', 'NA-US-OR'],
list(got.records)[0].dynamic.rules[0].data['geos'])
self.assertFalse('geos' in list(got.records)[0].dynamic.rules[1].data)
# Record with NA-CA-* only rule which is removed
desired = Zone('unit.tests.', [])
record = Record.new(desired, 'a', {
'ttl': 30,
'type': 'A',
'value': '1.2.3.4',
'dynamic': {
'pools': {
'one': {
'values': [{
'value': '1.2.3.4',
}],
},
'two': {
'values': [{
'value': '2.2.3.4',
}],
},
},
'rules': [{
'geos': ['NA-CA-BC'],
'pool': 'two',
}, {
'pool': 'one',
}],
},
})
desired.add_record(record)
got = provider.process_desired_zone(desired)
self.assertEquals(1, len(list(got.records)[0].dynamic.rules))
self.assertFalse('geos' in list(got.records)[0].dynamic.rules[0].data)
# Record with NA-CA-* rule combined with other geos, filtered
desired = Zone('unit.tests.', [])
record = Record.new(desired, 'a', {
'ttl': 30,
'type': 'A',
'value': '1.2.3.4',
'dynamic': {
'pools': {
'one': {
'values': [{
'value': '1.2.3.4',
}],
},
'two': {
'values': [{
'value': '2.2.3.4',
}],
},
},
'rules': [{
'geos': ['EU', 'NA-CA-NB', 'NA-US-OR'],
'pool': 'two',
}, {
'pool': 'one',
}],
},
})
desired.add_record(record)
got = provider.process_desired_zone(desired)
self.assertEquals(2, len(list(got.records)[0].dynamic.rules))
self.assertEquals(['EU', 'NA-US-OR'],
list(got.records)[0].dynamic.rules[0].data['geos'])
self.assertFalse('geos' in list(got.records)[0].dynamic.rules[1].data)
def test_populate_with_fallback(self):
provider, stubber = self._get_stubbed_fallback_auth_provider()