From 446f66f562b3752c9ffa807444cf3bcf48b44d99 Mon Sep 17 00:00:00 2001 From: Ross McFarland Date: Mon, 3 Dec 2018 15:51:30 -0800 Subject: [PATCH] Implement _DynamicMixin's value objects, unit test them --- octodns/record.py | 97 +++++++++++++++++++++++- tests/test_octodns_record.py | 138 ++++++++++++++++++++++++++++++++++- 2 files changed, 230 insertions(+), 5 deletions(-) diff --git a/octodns/record.py b/octodns/record.py index c19f22f..caac7a0 100644 --- a/octodns/record.py +++ b/octodns/record.py @@ -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) + + 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: - self.dynamic = dict(data['dynamic']) + rules = list(data['dynamic']['rules']) except: - self.dynamic = {} + 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) - # TODO: implement all this + 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): diff --git a/tests/test_octodns_record.py b/tests/test_octodns_record.py index 7a71d4c..5c7b7f2 100644 --- a/tests/test_octodns_record.py +++ b/tests/test_octodns_record.py @@ -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'])