RuhNetConsulting
/
octodns
mirror of https://github.com/octodns/octodns
1
0
Fork 0
Code Issues 0 Projects 0 Releases 52 Wiki Activity
Browse Source

WIP implementation of dyanmic pools & rules validation

pull/307/head
Ross McFarland 7 years ago
parent
e07165a2ab
commit
70c35aac26
No known key found for this signature in database GPG Key ID: 61C10C4FC8FE4A89
3 changed files with 356 additions and 11 deletions
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. +45
    -2
      octodns/record.py
  2. +8
    -7
      tests/config/dynamic.tests.yaml
  3. +303
    -2
      tests/test_octodns_record.py

+ 45
- 2
octodns/record.py View File

@ -395,12 +395,55 @@ class _DynamicMixin(object):
@classmethod
def validate(cls, name, data):
reasons = super(_DynamicMixin, cls).validate(name, data)
if 'dynamic' not in data:
return reasons
try:
pools = data['dynamic']['pools']
except KeyError:
pools = {}
for pool in sorted(pools.values()):
reasons.extend(cls._value_type.validate(pool))
if not pools:
reasons.append('missing pools')
else:
for pool in sorted(pools.values()):
reasons.extend(cls._value_type.validate(pool))
try:
rules = data['dynamic']['rules']
except KeyError:
rules = []
if not isinstance(rules, (list, tuple)):
reasons.append('rules must be a list')
elif not rules:
reasons.append('missing rules')
else:
for rule_num, rule in enumerate(rules):
rule_num += 1
try:
rule_pools = rule['pools']
except KeyError:
rule_pools = {}
if not rule_pools:
reasons.append('rule {} missing pools'.format(rule_num))
elif not isinstance(rule_pools, dict):
reasons.append('rule {} pools must be a dict'
.format(rule_num))
else:
for weight, pool in rule_pools.items():
try:
weight = int(weight)
if weight < 1 or weight > 255:
reasons.append('invalid pool weight "{}"'
.format(weight))
except ValueError:
reasons.append('invalid pool weight "{}"'
.format(weight))
if pool not in pools:
reasons.append('undefined pool "{}"'.format(pool))
return reasons
def __init__(self, zone, name, data, *args, **kwargs):


+ 8
- 7
tests/config/dynamic.tests.yaml View File

@ -11,11 +11,11 @@ a:
rules:
- geo: EU-UK
pools:
- iad
10: iad
- geo: EU
pools:
- ams
- iad
10: ams
10: iad
- geos:
- NA-US-CA
- NA-US-OR
@ -23,7 +23,8 @@ a:
pools:
25: iad
75: sea
- pool: iad
- pools:
10: iad
type: A
values:
- 2.2.2.2
@ -42,11 +43,11 @@ cname:
rules:
- geo: EU-UK
pools:
- iad
10: iad
- geo: EU
pools:
- ams
- iad
10: ams
10: iad
- geos:
- NA-US-CA
- NA-US-OR


+ 303
- 2
tests/test_octodns_record.py View File

@ -1926,6 +1926,51 @@ class TestDynamicRecords(TestCase):
self.assertEquals(a_data['dynamic'], a.dynamic)
def test_a_validation(self):
# Missing pools
a_data = {
'dynamic': {
'rules': [{
'pools': {
1: 'one',
}
}],
},
'ttl': 60,
'type': 'A',
'values': [
'1.1.1.1',
'2.2.2.2',
],
}
with self.assertRaises(ValidationError) as ctx:
Record.new(self.zone, 'bad', a_data)
self.assertEquals(['missing pools', 'undefined pool "one"'],
ctx.exception.reasons)
# Empty pools
a_data = {
'dynamic': {
'pools': {
},
'rules': [{
'pools': {
1: 'one',
}
}],
},
'ttl': 60,
'type': 'A',
'values': [
'1.1.1.1',
'2.2.2.2',
],
}
with self.assertRaises(ValidationError) as ctx:
Record.new(self.zone, 'bad', a_data)
self.assertEquals(['missing pools', 'undefined pool "one"'],
ctx.exception.reasons)
# Invalid addresses
a_data = {
'dynamic': {
'pools': {
@ -1937,8 +1982,8 @@ class TestDynamicRecords(TestCase):
},
'rules': [{
'pools': {
100: '5.5.5.5',
200: '6.6.6.6',
100: 'one',
200: 'two',
}
}],
},
@ -1954,3 +1999,259 @@ class TestDynamicRecords(TestCase):
self.assertEquals(['invalid IPv4 address "nor-is-this"',
'invalid IPv4 address "this-aint-right"'],
ctx.exception.reasons)
# missing value(s)
a_data = {
'dynamic': {
'pools': {
'one': [],
'two': [
'3.3.3.3',
'4.4.4.4',
],
},
'rules': [{
'pools': {
100: 'one',
200: 'two',
}
}],
},
'ttl': 60,
'type': 'A',
'values': [
'1.1.1.1',
'2.2.2.2',
],
}
with self.assertRaises(ValidationError) as ctx:
Record.new(self.zone, 'bad', a_data)
self.assertEquals(['missing value(s)'], ctx.exception.reasons)
# Empty value
a_data = {
'dynamic': {
'pools': {
'one': '',
'two': [
'3.3.3.3',
'blip',
],
},
'rules': [{
'pools': {
100: 'one',
200: 'two',
}
}],
},
'ttl': 60,
'type': 'A',
'values': [
'1.1.1.1',
'2.2.2.2',
],
}
with self.assertRaises(ValidationError) as ctx:
Record.new(self.zone, 'bad', a_data)
self.assertEquals(['invalid IPv4 address "blip"', 'empty value'],
ctx.exception.reasons)
# multiple problems
a_data = {
'dynamic': {
'pools': {
'one': '',
'two': [
'3.3.3.3',
'blip',
],
},
'rules': [{
'pools': {
100: 'one',
200: 'two',
}
}],
},
'ttl': 60,
'type': 'A',
'values': [
'1.1.1.1',
'2.2.2.2',
],
}
with self.assertRaises(ValidationError) as ctx:
Record.new(self.zone, 'bad', a_data)
self.assertEquals(['invalid IPv4 address "blip"', 'empty value'],
ctx.exception.reasons)
# missing rules
a_data = {
'dynamic': {
'pools': {
'one': '1.2.3.4',
},
},
'ttl': 60,
'type': 'A',
'values': [
'1.1.1.1',
'2.2.2.2',
],
}
with self.assertRaises(ValidationError) as ctx:
Record.new(self.zone, 'bad', a_data)
self.assertEquals(['missing rules'], ctx.exception.reasons)
# empty rules
a_data = {
'dynamic': {
'pools': {
'one': '1.2.3.4',
},
'rules': [],
},
'ttl': 60,
'type': 'A',
'values': [
'1.1.1.1',
'2.2.2.2',
],
}
with self.assertRaises(ValidationError) as ctx:
Record.new(self.zone, 'bad', a_data)
self.assertEquals(['missing rules'], ctx.exception.reasons)
# rules not a list/tuple
a_data = {
'dynamic': {
'pools': {
'one': '1.2.3.4',
},
'rules': {},
},
'ttl': 60,
'type': 'A',
'values': [
'1.1.1.1',
'2.2.2.2',
],
}
with self.assertRaises(ValidationError) as ctx:
Record.new(self.zone, 'bad', a_data)
self.assertEquals(['rules must be a list'], ctx.exception.reasons)
# rule without pools
a_data = {
'dynamic': {
'pools': {
'one': '1.2.3.4',
},
'rules': [{
}],
},
'ttl': 60,
'type': 'A',
'values': [
'1.1.1.1',
'2.2.2.2',
],
}
with self.assertRaises(ValidationError) as ctx:
Record.new(self.zone, 'bad', a_data)
self.assertEquals(['rule 1 missing pools'], ctx.exception.reasons)
# rule with non-dict pools
a_data = {
'dynamic': {
'pools': {
'one': '1.2.3.4',
},
'rules': [{
'pools': ['one'],
}],
},
'ttl': 60,
'type': 'A',
'values': [
'1.1.1.1',
'2.2.2.2',
],
}
with self.assertRaises(ValidationError) as ctx:
Record.new(self.zone, 'bad', a_data)
self.assertEquals(["rule 1 pools must be a dict"],
ctx.exception.reasons)
# rule references non-existant pool
a_data = {
'dynamic': {
'pools': {
'one': '1.2.3.4',
},
'rules': [{
'pools': {
10: 'non-existant'
}
}],
},
'ttl': 60,
'type': 'A',
'values': [
'1.1.1.1',
'2.2.2.2',
],
}
with self.assertRaises(ValidationError) as ctx:
Record.new(self.zone, 'bad', a_data)
self.assertEquals(["undefined pool \"non-existant\""],
ctx.exception.reasons)
# invalid int weight
a_data = {
'dynamic': {
'pools': {
'one': '1.2.3.4',
},
'rules': [{
'pools': {
256: 'one'
}
}],
},
'ttl': 60,
'type': 'A',
'values': [
'1.1.1.1',
'2.2.2.2',
],
}
with self.assertRaises(ValidationError) as ctx:
Record.new(self.zone, 'bad', a_data)
self.assertEquals(['invalid pool weight "256"'],
ctx.exception.reasons)
# invalid non-int weight
a_data = {
'dynamic': {
'pools': {
'one': '1.2.3.4',
},
'rules': [{
'pools': {
'foo': 'one'
}
}],
},
'ttl': 60,
'type': 'A',
'values': [
'1.1.1.1',
'2.2.2.2',
],
}
with self.assertRaises(ValidationError) as ctx:
Record.new(self.zone, 'bad', a_data)
self.assertEquals(['invalid pool weight "foo"'],
ctx.exception.reasons)

Write Preview
Loading…
Cancel
Save