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 _DynamicMixin's value objects, unit test them

Browse Source
This commit is contained in:
Ross McFarland
2018-12-03 15:51:30 -08:00
parent b650013ccb
commit 446f66f562
2 changed files with 232 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

101
octodns/record.py
View File

@@ -386,6 +386,47 @@ class _ValueMixin(object):
self.fqdn, self.value)
class _DynamicPool(object):
def __init__(self, _id, data):
self._id = _id
# TODO: actually parse this
self.data = data
def _data(self):
return self.data
class _DynamicRule(object):
def __init__(self, i, data):
self.i = i
# TODO: actually parse this
self.data = data
def _data(self):
return self.data
class _Dynamic(object):
def __init__(self, pools, rules):
self.pools = pools
self.rules = rules
def _data(self):
pools = {}
for _id, pool in self.pools.items():
pools[_id] = pool._data()
rules = []
for rule in self.rules:
rules.append(rule._data())
return {
'pools': pools,
'rules': rules,
}
class _DynamicMixin(object):
@classmethod
@@ -445,12 +486,62 @@ class _DynamicMixin(object):
def __init__(self, zone, name, data, *args, **kwargs):
super(_DynamicMixin, self).__init__(zone, name, data, *args,
**kwargs)
try:
self.dynamic = dict(data['dynamic'])
except:
self.dynamic = {}
# TODO: implement all this
self.dynamic = {}
if 'dynamic' not in data:
return
# pools
try:
pools = dict(data['dynamic']['pools'])
except:
pools = {}
for _id, pool in sorted(pools.items()):
pools[_id] = _DynamicPool(_id, pool)
# rules
try:
rules = list(data['dynamic']['rules'])
except:
rules = []
parsed = []
for i, rule in enumerate(rules):
parsed.append(_DynamicRule(i, rule))
# dynamic
self.dynamic = _Dynamic(pools, parsed)
def _data(self):
ret = super(_DynamicMixin, self)._data()
if self.dynamic:
ret['dynamic'] = self.dynamic._data()
return ret
def changes(self, other, target):
if target.SUPPORTS_DYNAMIC:
if self.dynamic != other.dynamic:
return Update(self, other)
return super(_DynamicMixin, self).changes(other, target)
def __repr__(self):
# TODO: improve this whole thing, we need multi-line...
if self.dynamic:
# TODO: this hack can't going to cut it, as part of said
# improvements the value types should deal with serializing their
# value
try:
values = self.values
except AttributeError:
values = self.value
return '<{} {} {}, {}, {}, {}>'.format(self.__class__.__name__,
self._type, self.ttl,
self.fqdn, values,
self.dynamic)
return super(_DynamicMixin, self).__repr__()
class Ipv4List(object):

138
tests/test_octodns_record.py
View File

@@ -1923,9 +1923,98 @@ class TestDynamicRecords(TestCase):
self.assertEquals('A', a._type)
self.assertEquals(a_data['ttl'], a.ttl)
self.assertEquals(a_data['values'], a.values)
self.assertEquals(a_data['dynamic'], a.dynamic)
def test_a_validation(self):
dynamic = a.dynamic
self.assertTrue(dynamic)
pools = dynamic.pools
self.assertTrue(pools)
self.assertEquals(a_data['dynamic']['pools']['one'], pools['one'].data)
self.assertEquals(a_data['dynamic']['pools']['two'], pools['two'].data)
rules = dynamic.rules
self.assertTrue(rules)
self.assertEquals(a_data['dynamic']['rules'][0], rules[0].data)
def test_simple_aaaa_weighted(self):
aaaa_data = {
'dynamic': {
'pools': {
'one': '2601:642:500:e210:62f8:1dff:feb8:9473',
'two': [
'2601:642:500:e210:62f8:1dff:feb8:9474',
'2601:642:500:e210:62f8:1dff:feb8:9475',
],
},
'rules': [{
'pools': {
100: 'one',
200: 'two',
}
}],
},
'ttl': 60,
'values': [
'2601:642:500:e210:62f8:1dff:feb8:9471',
'2601:642:500:e210:62f8:1dff:feb8:9472',
],
}
aaaa = AaaaRecord(self.zone, 'weighted', aaaa_data)
self.assertEquals('AAAA', aaaa._type)
self.assertEquals(aaaa_data['ttl'], aaaa.ttl)
self.assertEquals(aaaa_data['values'], aaaa.values)
dynamic = aaaa.dynamic
self.assertTrue(dynamic)
pools = dynamic.pools
self.assertTrue(pools)
self.assertEquals(aaaa_data['dynamic']['pools']['one'],
pools['one'].data)
self.assertEquals(aaaa_data['dynamic']['pools']['two'],
pools['two'].data)
rules = dynamic.rules
self.assertTrue(rules)
self.assertEquals(aaaa_data['dynamic']['rules'][0], rules[0].data)
def test_simple_cname_weighted(self):
cname_data = {
'dynamic': {
'pools': {
'one': 'one.cname.target.',
'two': 'two.cname.target.',
},
'rules': [{
'pools': {
100: 'one',
200: 'two',
}
}],
},
'ttl': 60,
'value': 'cname.target.',
}
cname = CnameRecord(self.zone, 'weighted', cname_data)
self.assertEquals('CNAME', cname._type)
self.assertEquals(cname_data['ttl'], cname.ttl)
self.assertEquals(cname_data['value'], cname.value)
dynamic = cname.dynamic
self.assertTrue(dynamic)
pools = dynamic.pools
self.assertTrue(pools)
self.assertEquals(cname_data['dynamic']['pools']['one'],
pools['one'].data)
self.assertEquals(cname_data['dynamic']['pools']['two'],
pools['two'].data)
rules = dynamic.rules
self.assertTrue(rules)
self.assertEquals(cname_data['dynamic']['rules'][0], rules[0].data)
def test_dynamic_validation(self):
# Missing pools
a_data = {
'dynamic': {
@@ -2255,3 +2344,48 @@ class TestDynamicRecords(TestCase):
Record.new(self.zone, 'bad', a_data)
self.assertEquals(['invalid pool weight "foo"'],
ctx.exception.reasons)
def test_dynamic_lenient(self):
# Missing pools
a_data = {
'dynamic': {
'rules': [{
'pools': {
1: 'one',
}
}],
},
'ttl': 60,
'type': 'A',
'values': [
'1.1.1.1',
'2.2.2.2',
],
}
a = Record.new(self.zone, 'bad', a_data, lenient=True)
self.assertEquals({
'pools': {},
'rules': a_data['dynamic']['rules'],
}, a._data()['dynamic'])
# Missing rule
a_data = {
'dynamic': {
'pools': {
'one': '1.1.1.1',
},
},
'ttl': 60,
'type': 'A',
'values': [
'1.1.1.1',
'2.2.2.2',
],
}
a = Record.new(self.zone, 'bad', a_data, lenient=True)
self.assertEquals({
'pools': {
'one': '1.1.1.1',
},
'rules': [],
}, a._data()['dynamic'])