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

POC supports & dynamic checking in _process_desired_zone

Browse Source
This commit is contained in:
Ross McFarland
2021-08-23 12:26:09 -07:00
parent dfd3a12f92
commit 47de105a29
3 changed files with 67 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
octodns/provider/base.py
View File

@ -50,12 +50,24 @@ class BaseProvider(BaseSource):
that are made to have them logged or throw errors depending on the that are made to have them logged or throw errors depending on the
provider configuration. provider configuration.
''' '''
if self.SUPPORTS_MUTLIVALUE_PTR:
# nothing do here
return desired
for record in desired.records: for record in desired.records:
if record._type == 'PTR' and len(record.values) > 1: if record._type not in self.SUPPORTS:
msg = '{} records not supported for {}'.format(record._type,
record.fqdn)
fallback = 'omitting record'
self.supports_warn_or_except(msg, fallback)
desired.remove_record(record)
elif getattr(record, 'dynamic', False) and not self.SUPPORTS_DYNAMIC:
msg = 'dynamic records not supported for {}'\
.format(record.fqdn)
fallback = 'falling back to simple record'
self.supports_warn_or_except(msg, fallback)
record = record.copy()
record.dynamic = None
desired.add_record(record, replace=True)
elif record._type == 'PTR' and len(record.values) > 1 and \
not self.SUPPORTS_MUTLIVALUE_PTR:
# replace with a single-value copy # replace with a single-value copy
msg = 'multi-value PTR records not supported for {}' \ msg = 'multi-value PTR records not supported for {}' \
.format(record.fqdn) .format(record.fqdn)

2
tests/helpers.py
View File

@ -107,7 +107,7 @@ class PlannableProvider(BaseProvider):
SUPPORTS_GEO = False SUPPORTS_GEO = False
SUPPORTS_DYNAMIC = False SUPPORTS_DYNAMIC = False
SUPPORTS = set(('A',)) SUPPORTS = set(('A', 'AAAA', 'TXT'))
def __init__(self, *args, **kwargs): def __init__(self, *args, **kwargs):
super(PlannableProvider, self).__init__(*args, **kwargs) super(PlannableProvider, self).__init__(*args, **kwargs)

53
tests/test_octodns_provider_base.py
View File

@ -21,7 +21,10 @@ from octodns.zone import Zone
class HelperProvider(BaseProvider): class HelperProvider(BaseProvider):
log = getLogger('HelperProvider') log = getLogger('HelperProvider')
SUPPORTS = set(('A',)) SUPPORTS = set(('A', 'PTR'))
SUPPORTS_MUTLIVALUE_PTR = False
SUPPORTS_DYNAMIC = False
id = 'test' id = 'test'
strict_supports = False strict_supports = False
@ -234,6 +237,10 @@ class TestBaseProvider(TestCase):
self.assertFalse(plan) self.assertFalse(plan)
def test_process_desired_zone(self): def test_process_desired_zone(self):
provider = HelperProvider('test')
# SUPPORTS_MUTLIVALUE_PTR
provider.SUPPORTS_MUTLIVALUE_PTR = False
zone1 = Zone('unit.tests.', []) zone1 = Zone('unit.tests.', [])
record1 = Record.new(zone1, 'ptr', { record1 = Record.new(zone1, 'ptr', {
'type': 'PTR', 'type': 'PTR',
@ -242,11 +249,51 @@ class TestBaseProvider(TestCase):
}) })
zone1.add_record(record1) zone1.add_record(record1)
zone2 = HelperProvider('hasptr')._process_desired_zone(zone1) zone2 = provider._process_desired_zone(zone1.copy())
record2 = list(zone2.records)[0] record2 = list(zone2.records)[0]
self.assertEqual(len(record2.values), 1) self.assertEqual(len(record2.values), 1)
provider.SUPPORTS_MUTLIVALUE_PTR = True
zone2 = provider._process_desired_zone(zone1.copy())
record2 = list(zone2.records)[0]
from pprint import pprint
pprint([
record1, record2
])
self.assertEqual(len(record2.values), 2)
# SUPPORTS_DYNAMIC
provider.SUPPORTS_DYNAMIC = False
zone1 = Zone('unit.tests.', [])
record1 = Record.new(zone1, 'a', {
'dynamic': {
'pools': {
'one': {
'values': [{
'value': '1.1.1.1',
}],
},
},
'rules': [{
'pool': 'one',
}],
},
'type': 'A',
'ttl': 3600,
'values': ['2.2.2.2'],
})
self.assertTrue(record1.dynamic)
zone1.add_record(record1)
zone2 = provider._process_desired_zone(zone1.copy())
record2 = list(zone2.records)[0]
self.assertFalse(record2.dynamic)
provider.SUPPORTS_DYNAMIC = True
zone2 = provider._process_desired_zone(zone1.copy())
record2 = list(zone2.records)[0]
self.assertTrue(record2.dynamic)
def test_safe_none(self): def test_safe_none(self):
# No changes is safe # No changes is safe
Plan(None, None, [], True).raise_if_unsafe() Plan(None, None, [], True).raise_if_unsafe()