Browse Source

Further clean up/division of labor and tests

pull/307/head
Ross McFarland 7 years ago
parent
commit
c9b373f0ae
No known key found for this signature in database GPG Key ID: 61C10C4FC8FE4A89
2 changed files with 158 additions and 40 deletions
  1. +33
    -37
      octodns/record.py
  2. +125
    -3
      tests/test_octodns_record.py

+ 33
- 37
octodns/record.py View File

@ -254,35 +254,13 @@ class _ValuesMixin(object):
def validate(cls, name, data): def validate(cls, name, data):
reasons = super(_ValuesMixin, cls).validate(name, data) reasons = super(_ValuesMixin, cls).validate(name, data)
values = []
try: try:
values = data['values'] values = data['values']
if not values:
values = []
reasons.append('missing value(s)')
else:
# loop through copy of values
# remove invalid value from values
for value in list(values):
if value is None:
reasons.append('missing value(s)')
values.remove(value)
elif len(value) == 0:
reasons.append('empty value')
values.remove(value)
except KeyError: except KeyError:
try: try:
value = data['value']
if value is None:
reasons.append('missing value(s)')
values = []
elif len(value) == 0:
reasons.append('empty value')
values = []
else:
values = value
values = data['value']
except KeyError: except KeyError:
reasons.append('missing value(s)')
values = []
reasons.extend(cls._value_type.validate(values)) reasons.extend(cls._value_type.validate(values))
@ -391,10 +369,10 @@ class _ValueMixin(object):
value = None value = None
try: try:
value = data['value'] value = data['value']
if value is None:
reasons.append('missing value')
elif value == '':
if value == '':
reasons.append('empty value') reasons.append('empty value')
elif value is None:
reasons.append('missing value')
except KeyError: except KeyError:
reasons.append('missing value') reasons.append('missing value')
if value: if value:
@ -451,12 +429,19 @@ class Ipv4List(object):
def validate(cls, data): def validate(cls, data):
if not isinstance(data, (list, tuple)): if not isinstance(data, (list, tuple)):
data = (data,) data = (data,)
if len(data) == 0:
return ['missing value(s)']
reasons = [] reasons = []
for value in data: for value in data:
try:
IPv4Address(unicode(value))
except Exception:
reasons.append('invalid IPv4 address "{}"'.format(value))
if value is '':
reasons.append('empty value')
elif value is None:
reasons.append('missing value(s)')
else:
try:
IPv4Address(unicode(value))
except Exception:
reasons.append('invalid IPv4 address "{}"'.format(value))
return reasons return reasons
@ -466,12 +451,19 @@ class Ipv6List(object):
def validate(cls, data): def validate(cls, data):
if not isinstance(data, (list, tuple)): if not isinstance(data, (list, tuple)):
data = (data,) data = (data,)
if len(data) == 0:
return ['missing value(s)']
reasons = [] reasons = []
for value in data: for value in data:
try:
IPv6Address(unicode(value))
except Exception:
reasons.append('invalid IPv6 address "{}"'.format(value))
if value is '':
reasons.append('empty value')
elif value is None:
reasons.append('missing value(s)')
else:
try:
IPv6Address(unicode(value))
except Exception:
reasons.append('invalid IPv6 address "{}"'.format(value))
return reasons return reasons
@ -743,7 +735,9 @@ class _NsValue(object):
@classmethod @classmethod
def validate(cls, data): def validate(cls, data):
if not isinstance(data, (list, tuple)):
if not data:
return ['missing value(s)']
elif not isinstance(data, (list, tuple)):
data = (data,) data = (data,)
reasons = [] reasons = []
for value in data: for value in data:
@ -869,7 +863,9 @@ class _ChunkedValue(object):
@classmethod @classmethod
def validate(cls, data): def validate(cls, data):
if not isinstance(data, (list, tuple)):
if not data:
return ['missing value(s)']
elif not isinstance(data, (list, tuple)):
data = (data,) data = (data,)
reasons = [] reasons = []
for value in data: for value in data:


+ 125
- 3
tests/test_octodns_record.py View File

@ -934,7 +934,7 @@ class TestRecordValidation(TestCase):
self.assertEquals(['missing ttl', 'missing value(s)'], self.assertEquals(['missing ttl', 'missing value(s)'],
ctx.exception.reasons) ctx.exception.reasons)
# invalid ip address
# invalid ipv4 address
with self.assertRaises(ValidationError) as ctx: with self.assertRaises(ValidationError) as ctx:
Record.new(self.zone, '', { Record.new(self.zone, '', {
'type': 'A', 'type': 'A',
@ -944,7 +944,7 @@ class TestRecordValidation(TestCase):
self.assertEquals(['invalid IPv4 address "hello"'], self.assertEquals(['invalid IPv4 address "hello"'],
ctx.exception.reasons) ctx.exception.reasons)
# invalid ip addresses
# invalid ipv4 addresses
with self.assertRaises(ValidationError) as ctx: with self.assertRaises(ValidationError) as ctx:
Record.new(self.zone, '', { Record.new(self.zone, '', {
'type': 'A', 'type': 'A',
@ -956,7 +956,7 @@ class TestRecordValidation(TestCase):
'invalid IPv4 address "goodbye"' 'invalid IPv4 address "goodbye"'
], ctx.exception.reasons) ], ctx.exception.reasons)
# invalid & valid ip addresses, no ttl
# invalid & valid ipv4 addresses, no ttl
with self.assertRaises(ValidationError) as ctx: with self.assertRaises(ValidationError) as ctx:
Record.new(self.zone, '', { Record.new(self.zone, '', {
'type': 'A', 'type': 'A',
@ -967,6 +967,128 @@ class TestRecordValidation(TestCase):
'invalid IPv4 address "hello"', 'invalid IPv4 address "hello"',
], ctx.exception.reasons) ], ctx.exception.reasons)
def test_AAAA_validation(self):
# doesn't blow up
Record.new(self.zone, '', {
'type': 'AAAA',
'ttl': 600,
'value': '2601:644:500:e210:62f8:1dff:feb8:947a',
})
Record.new(self.zone, '', {
'type': 'AAAA',
'ttl': 600,
'values': [
'2601:644:500:e210:62f8:1dff:feb8:947a',
]
})
Record.new(self.zone, '', {
'type': 'AAAA',
'ttl': 600,
'values': [
'2601:644:500:e210:62f8:1dff:feb8:947a',
'2601:642:500:e210:62f8:1dff:feb8:947a',
]
})
# missing value(s), no value or value
with self.assertRaises(ValidationError) as ctx:
Record.new(self.zone, '', {
'type': 'AAAA',
'ttl': 600,
})
self.assertEquals(['missing value(s)'], ctx.exception.reasons)
# missing value(s), empty values
with self.assertRaises(ValidationError) as ctx:
Record.new(self.zone, 'www', {
'type': 'AAAA',
'ttl': 600,
'values': []
})
self.assertEquals(['missing value(s)'], ctx.exception.reasons)
# missing value(s), None values
with self.assertRaises(ValidationError) as ctx:
Record.new(self.zone, 'www', {
'type': 'AAAA',
'ttl': 600,
'values': None
})
self.assertEquals(['missing value(s)'], ctx.exception.reasons)
# missing value(s) and empty value
with self.assertRaises(ValidationError) as ctx:
Record.new(self.zone, 'www', {
'type': 'AAAA',
'ttl': 600,
'values': [None, '']
})
self.assertEquals(['missing value(s)',
'empty value'], ctx.exception.reasons)
# missing value(s), None value
with self.assertRaises(ValidationError) as ctx:
Record.new(self.zone, 'www', {
'type': 'AAAA',
'ttl': 600,
'value': None
})
self.assertEquals(['missing value(s)'], ctx.exception.reasons)
# empty value, empty string value
with self.assertRaises(ValidationError) as ctx:
Record.new(self.zone, 'www', {
'type': 'AAAA',
'ttl': 600,
'value': ''
})
self.assertEquals(['empty value'], ctx.exception.reasons)
# missing value(s) & ttl
with self.assertRaises(ValidationError) as ctx:
Record.new(self.zone, '', {
'type': 'AAAA',
})
self.assertEquals(['missing ttl', 'missing value(s)'],
ctx.exception.reasons)
# invalid IPv6 address
with self.assertRaises(ValidationError) as ctx:
Record.new(self.zone, '', {
'type': 'AAAA',
'ttl': 600,
'value': 'hello'
})
self.assertEquals(['invalid IPv6 address "hello"'],
ctx.exception.reasons)
# invalid IPv6 addresses
with self.assertRaises(ValidationError) as ctx:
Record.new(self.zone, '', {
'type': 'AAAA',
'ttl': 600,
'values': ['hello', 'goodbye']
})
self.assertEquals([
'invalid IPv6 address "hello"',
'invalid IPv6 address "goodbye"'
], ctx.exception.reasons)
# invalid & valid IPv6 addresses, no ttl
with self.assertRaises(ValidationError) as ctx:
Record.new(self.zone, '', {
'type': 'AAAA',
'values': [
'2601:644:500:e210:62f8:1dff:feb8:947a',
'hello',
'2601:642:500:e210:62f8:1dff:feb8:947a'
]
})
self.assertEquals([
'missing ttl',
'invalid IPv6 address "hello"',
], ctx.exception.reasons)
def test_geo(self): def test_geo(self):
Record.new(self.zone, '', { Record.new(self.zone, '', {
'geo': { 'geo': {


Loading…
Cancel
Save