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

Merge branch 'master' of https://github.com/github/octodns into AZProvider

Browse Source
This commit is contained in:
Heesu Hwang
2017-06-22 14:26:29 -07:00
parent 3c1e409e6f 0aa0878ae8
commit 4a9889bd59
17 changed files with 288 additions and 159 deletions
Download Patch File Download Diff File Expand all files Collapse all files
octodns/provider/cloudflare.py
+2 -5
View File
@@ -36,7 +36,7 @@ class CloudflareProvider(BaseProvider):
'''
SUPPORTS_GEO = False
# TODO: support SRV
UNSUPPORTED_TYPES = ('ALIAS', 'NAPTR', 'PTR', 'SOA', 'SRV', 'SSHFP')
SUPPORTS = set(('A', 'AAAA', 'CNAME', 'MX', 'NS', 'SPF', 'TXT'))
MIN_TTL = 120
TIMEOUT = 15
@@ -56,9 +56,6 @@ class CloudflareProvider(BaseProvider):
self._zones = None
self._zone_records = {}
def supports(self, record):
return record._type not in self.UNSUPPORTED_TYPES
def _request(self, method, path, params=None, data=None):
self.log.debug('_request: method=%s, path=%s', method, path)
@@ -167,7 +164,7 @@ class CloudflareProvider(BaseProvider):
for record in records:
name = zone.hostname_from_fqdn(record['name'])
_type = record['type']
if _type not in self.UNSUPPORTED_TYPES:
if _type in self.SUPPORTS:
values[name][record['type']].append(record)
for name, types in values.items():
octodns/provider/dnsimple.py
+2
View File
@@ -91,6 +91,8 @@ class DnsimpleProvider(BaseProvider):
account: 42
'''
SUPPORTS_GEO = False
SUPPORTS = set(('A', 'AAAA', 'ALIAS', 'CNAME', 'MX', 'NAPTR', 'NS', 'PTR',
'SPF', 'SRV', 'SSHFP', 'TXT'))
def __init__(self, id, token, account, *args, **kwargs):
self.log = logging.getLogger('DnsimpleProvider[{}]'.format(id))
octodns/provider/dyn.py
+2
View File
@@ -106,6 +106,7 @@ class DynProvider(BaseProvider):
than one account active at a time. See DynProvider._check_dyn_sess for some
related bits.
'''
RECORDS_TO_TYPE = {
'a_records': 'A',
'aaaa_records': 'AAAA',
@@ -121,6 +122,7 @@ class DynProvider(BaseProvider):
'txt_records': 'TXT',
}
TYPE_TO_RECORDS = {v: k for k, v in RECORDS_TO_TYPE.items()}
SUPPORTS = set(TYPE_TO_RECORDS.keys())
# https://help.dyn.com/predefined-geotm-regions-groups/
REGION_CODES = {
octodns/provider/ns1.py
+3 -3
View File
@@ -22,6 +22,9 @@ class Ns1Provider(BaseProvider):
api_key: env/NS1_API_KEY
'''
SUPPORTS_GEO = False
SUPPORTS = set(('A', 'AAAA', 'ALIAS', 'CNAME', 'MX', 'NAPTR', 'NS', 'PTR',
'SPF', 'SRV', 'TXT'))
ZONE_NOT_FOUND_MESSAGE = 'server error: zone not found'
def __init__(self, id, api_key, *args, **kwargs):
@@ -30,9 +33,6 @@ class Ns1Provider(BaseProvider):
super(Ns1Provider, self).__init__(id, *args, **kwargs)
self._client = NSONE(apiKey=api_key)
def supports(self, record):
return record._type != 'SSHFP'
def _data_for_A(self, _type, record):
return {
'ttl': record['ttl'],
octodns/provider/powerdns.py
+2
View File
@@ -14,6 +14,8 @@ from .base import BaseProvider
class PowerDnsBaseProvider(BaseProvider):
SUPPORTS_GEO = False
SUPPORTS = set(('A', 'AAAA', 'ALIAS', 'CNAME', 'MX', 'NAPTR', 'NS', 'PTR',
'SPF', 'SSHFP', 'SRV', 'TXT'))
TIMEOUT = 5
def __init__(self, id, host, api_key, port=8081, *args, **kwargs):
octodns/provider/route53.py
+143 -103
View File
@@ -16,27 +16,71 @@ from ..record import Record, Update
from .base import BaseProvider
octal_re = re.compile(r'\\(\d\d\d)')
def _octal_replace(s):
# See http://docs.aws.amazon.com/Route53/latest/DeveloperGuide/
# DomainNameFormat.html
return octal_re.sub(lambda m: chr(int(m.group(1), 8)), s)
class _Route53Record(object):
def __init__(self, fqdn, _type, ttl, record=None, values=None, geo=None,
health_check_id=None):
self.fqdn = fqdn
self._type = _type
self.ttl = ttl
# From here on things are a little ugly, it works, but would be nice to
# clean up someday.
if record:
values_for = getattr(self, '_values_for_{}'.format(self._type))
self.values = values_for(record)
@classmethod
def new(self, provider, record, creating):
ret = set()
if getattr(record, 'geo', False):
ret.add(_Route53GeoDefault(provider, record, creating))
for ident, geo in record.geo.items():
ret.add(_Route53GeoRecord(provider, record, ident, geo,
creating))
else:
self.values = values
self.geo = geo
self.health_check_id = health_check_id
self.is_geo_default = False
ret.add(_Route53Record(provider, record, creating))
return ret
@property
def _geo_code(self):
return getattr(self.geo, 'code', '')
def __init__(self, provider, record, creating):
self.fqdn = record.fqdn
self._type = record._type
self.ttl = record.ttl
values_for = getattr(self, '_values_for_{}'.format(self._type))
self.values = values_for(record)
def mod(self, action):
return {
'Action': action,
'ResourceRecordSet': {
'Name': self.fqdn,
'ResourceRecords': [{'Value': v} for v in self.values],
'TTL': self.ttl,
'Type': self._type,
}
}
# NOTE: we're using __hash__ and __cmp__ methods that consider
# _Route53Records equivalent if they have the same class, fqdn, and _type.
# Values are ignored. This is usful when computing diffs/changes.
def __hash__(self):
'sub-classes should never use this method'
return '{}:{}'.format(self.fqdn, self._type).__hash__()
def __cmp__(self, other):
'''sub-classes should call up to this and return its value if non-zero.
When it's zero they should compute their own __cmp__'''
if self.__class__ != other.__class__:
return cmp(self.__class__, other.__class__)
elif self.fqdn != other.fqdn:
return cmp(self.fqdn, other.fqdn)
elif self._type != other._type:
return cmp(self._type, other._type)
# We're ignoring ttl, it's not an actual differentiator
return 0
def __repr__(self):
return '_Route53Record<{} {} {} {}>'.format(self.fqdn, self._type,
self.ttl, self.values)
def _values_for_values(self, record):
return record.values
@@ -75,68 +119,91 @@ class _Route53Record(object):
v.target)
for v in record.values]
class _Route53GeoDefault(_Route53Record):
def mod(self, action):
return {
'Action': action,
'ResourceRecordSet': {
'Name': self.fqdn,
'GeoLocation': {
'CountryCode': '*'
},
'ResourceRecords': [{'Value': v} for v in self.values],
'SetIdentifier': 'default',
'TTL': self.ttl,
'Type': self._type,
}
}
def __hash__(self):
return '{}:{}:default'.format(self.fqdn, self._type).__hash__()
def __repr__(self):
return '_Route53GeoDefault<{} {} {} {}>'.format(self.fqdn, self._type,
self.ttl, self.values)
class _Route53GeoRecord(_Route53Record):
def __init__(self, provider, record, ident, geo, creating):
super(_Route53GeoRecord, self).__init__(provider, record, creating)
self.geo = geo
self.health_check_id = provider.get_health_check_id(record, ident,
geo, creating)
def mod(self, action):
geo = self.geo
rrset = {
'Name': self.fqdn,
'Type': self._type,
'TTL': self.ttl,
'ResourceRecords': [{'Value': v} for v in self.values],
}
if self.is_geo_default:
rrset['GeoLocation'] = {
'GeoLocation': {
'CountryCode': '*'
},
'ResourceRecords': [{'Value': v} for v in geo.values],
'SetIdentifier': geo.code,
'TTL': self.ttl,
'Type': self._type,
}
if self.health_check_id:
rrset['HealthCheckId'] = self.health_check_id
if geo.subdivision_code:
rrset['GeoLocation'] = {
'CountryCode': geo.country_code,
'SubdivisionCode': geo.subdivision_code
}
elif geo.country_code:
rrset['GeoLocation'] = {
'CountryCode': geo.country_code
}
else:
rrset['GeoLocation'] = {
'ContinentCode': geo.continent_code
}
rrset['SetIdentifier'] = 'default'
elif self.geo:
geo = self.geo
rrset['SetIdentifier'] = geo.code
if self.health_check_id:
rrset['HealthCheckId'] = self.health_check_id
if geo.subdivision_code:
rrset['GeoLocation'] = {
'CountryCode': geo.country_code,
'SubdivisionCode': geo.subdivision_code
}
elif geo.country_code:
rrset['GeoLocation'] = {
'CountryCode': geo.country_code
}
else:
rrset['GeoLocation'] = {
'ContinentCode': geo.continent_code
}
return {
'Action': action,
'ResourceRecordSet': rrset,
}
# NOTE: we're using __hash__ and __cmp__ methods that consider
# _Route53Records equivalent if they have the same fqdn, _type, and
# geo.ident. Values are ignored. This is usful when computing
# diffs/changes.
def __hash__(self):
return '{}:{}:{}'.format(self.fqdn, self._type,
self._geo_code).__hash__()
self.geo.code).__hash__()
def __cmp__(self, other):
return 0 if (self.fqdn == other.fqdn and
self._type == other._type and
self._geo_code == other._geo_code) else 1
ret = super(_Route53GeoRecord, self).__cmp__(other)
if ret != 0:
return ret
return cmp(self.geo.code, other.geo.code)
def __repr__(self):
return '_Route53Record<{} {:>5} {:8} {}>' \
.format(self.fqdn, self._type, self._geo_code, self.values)
octal_re = re.compile(r'\\(\d\d\d)')
def _octal_replace(s):
# See http://docs.aws.amazon.com/Route53/latest/DeveloperGuide/
# DomainNameFormat.html
return octal_re.sub(lambda m: chr(int(m.group(1), 8)), s)
return '_Route53GeoRecord<{} {} {} {} {}>'.format(self.fqdn,
self._type, self.ttl,
self.geo.code,
self.values)
class Route53Provider(BaseProvider):
@@ -153,6 +220,8 @@ class Route53Provider(BaseProvider):
In general the account used will need full permissions on Route53.
'''
SUPPORTS_GEO = True
SUPPORTS = set(('A', 'AAAA', 'CNAME', 'MX', 'NAPTR', 'NS', 'PTR', 'SPF',
'SRV', 'TXT'))
# This should be bumped when there are underlying changes made to the
# health check config.
@@ -172,9 +241,6 @@ class Route53Provider(BaseProvider):
self._r53_rrsets = {}
self._health_checks = None
def supports(self, record):
return record._type not in ('ALIAS', 'SSHFP')
@property
def r53_zones(self):
if self._r53_zones is None:
@@ -391,7 +457,7 @@ class Route53Provider(BaseProvider):
def _gen_mods(self, action, records):
'''
Turns `_Route53Record`s in to `change_resource_record_sets` `Changes`
Turns `_Route53*`s in to `change_resource_record_sets` `Changes`
'''
return [r.mod(action) for r in records]
@@ -420,14 +486,14 @@ class Route53Provider(BaseProvider):
# We've got a cached version use it
return self._health_checks
def _get_health_check_id(self, record, ident, geo, create):
def get_health_check_id(self, record, ident, geo, create):
# fqdn & the first value are special, we use them to match up health
# checks to their records. Route53 health checks check a single ip and
# we're going to assume that ips are interchangeable to avoid
# health-checking each one independently
fqdn = record.fqdn
first_value = geo.values[0]
self.log.debug('_get_health_check_id: fqdn=%s, type=%s, geo=%s, '
self.log.debug('get_health_check_id: fqdn=%s, type=%s, geo=%s, '
'first_value=%s', fqdn, record._type, ident,
first_value)
@@ -473,7 +539,7 @@ class Route53Provider(BaseProvider):
# store the new health check so that we'll be able to find it in the
# future
self._health_checks[id] = health_check
self.log.info('_get_health_check_id: created id=%s, host=%s, '
self.log.info('get_health_check_id: created id=%s, host=%s, '
'first_value=%s', id, host, first_value)
return id
@@ -482,8 +548,9 @@ class Route53Provider(BaseProvider):
# Find the health checks we're using for the new route53 records
in_use = set()
for r in new:
if r.health_check_id:
in_use.add(r.health_check_id)
hc_id = getattr(r, 'health_check_id', False)
if hc_id:
in_use.add(hc_id)
self.log.debug('_gc_health_checks: in_use=%s', in_use)
# Now we need to run through ALL the health checks looking for those
# that apply to this record, deleting any that do and are no longer in
@@ -502,23 +569,9 @@ class Route53Provider(BaseProvider):
def _gen_records(self, record, creating=False):
'''
Turns an octodns.Record into one or more `_Route53Record`s
Turns an octodns.Record into one or more `_Route53*`s
'''
records = set()
base = _Route53Record(record.fqdn, record._type, record.ttl,
record=record)
records.add(base)
if getattr(record, 'geo', False):
base.is_geo_default = True
for ident, geo in record.geo.items():
health_check_id = self._get_health_check_id(record, ident, geo,
creating)
records.add(_Route53Record(record.fqdn, record._type,
record.ttl, values=geo.values,
geo=geo,
health_check_id=health_check_id))
return records
return _Route53Record.new(self, record, creating)
def _mod_Create(self, change):
# New is the stuff that needs to be created
@@ -544,24 +597,11 @@ class Route53Provider(BaseProvider):
# things that haven't actually changed, but that's for another day.
# We can't use set math here b/c we won't be able to control which of
# the two objects will be in the result and we need to ensure it's the
# new one and we have to include some special handling when converting
# to/from a GEO enabled record
# new one.
upserts = set()
existing_records = {r: r for r in existing_records}
for new_record in new_records:
try:
existing_record = existing_records[new_record]
if new_record.is_geo_default != existing_record.is_geo_default:
# going from normal to geo or geo to normal, need a delete
# and create
deletes.add(existing_record)
creates.add(new_record)
else:
# just an update
upserts.add(new_record)
except KeyError:
# Completely new record, ignore
pass
if new_record in existing_records:
upserts.add(new_record)
return self._gen_mods('DELETE', deletes) + \
self._gen_mods('CREATE', creates) + \
octodns/provider/yaml.py
+12 -4
View File
@@ -26,16 +26,24 @@ class YamlProvider(BaseProvider):
# The ttl to use for records when not specified in the data
# (optional, default 3600)
default_ttl: 3600
# Whether or not to enforce sorting order on the yaml config
# (optional, default True)
enforce_order: True
'''
SUPPORTS_GEO = True
SUPPORTS = set(('A', 'AAAA', 'ALIAS', 'CNAME', 'MX', 'NAPTR', 'NS', 'PTR',
'SSHFP', 'SPF', 'SRV', 'TXT'))
def __init__(self, id, directory, default_ttl=3600, *args, **kwargs):
def __init__(self, id, directory, default_ttl=3600, enforce_order=True,
*args, **kwargs):
self.log = logging.getLogger('YamlProvider[{}]'.format(id))
self.log.debug('__init__: id=%s, directory=%s, default_ttl=%d', id,
directory, default_ttl)
self.log.debug('__init__: id=%s, directory=%s, default_ttl=%d, '
'enforce_order=%d', id, directory, default_ttl,
enforce_order)
super(YamlProvider, self).__init__(id, *args, **kwargs)
self.directory = directory
self.default_ttl = default_ttl
self.enforce_order = enforce_order
def populate(self, zone, target=False):
self.log.debug('populate: zone=%s, target=%s', zone.name, target)
@@ -47,7 +55,7 @@ class YamlProvider(BaseProvider):
before = len(zone.records)
filename = join(self.directory, '{}yaml'.format(zone.name))
with open(filename, 'r') as fh:
yaml_data = safe_load(fh)
yaml_data = safe_load(fh, enforce_order=self.enforce_order)
if yaml_data:
for name, data in yaml_data.items():
if not isinstance(data, list):
octodns/source/base.py
+4 -3
View File
@@ -16,6 +16,9 @@ class BaseSource(object):
if not hasattr(self, 'SUPPORTS_GEO'):
raise NotImplementedError('Abstract base class, SUPPORTS_GEO '
'property missing')
if not hasattr(self, 'SUPPORTS'):
raise NotImplementedError('Abstract base class, SUPPORTS '
'property missing')
def populate(self, zone, target=False):
'''
@@ -25,9 +28,7 @@ class BaseSource(object):
'missing')
def supports(self, record):
# Unless overriden and handled appropriaitely we'll assume that all
# record types are supported
return True
return record._type in self.SUPPORTS
def __repr__(self):
return self.__class__.__name__
octodns/source/tinydns.py
+1
View File
@@ -19,6 +19,7 @@ from .base import BaseSource
class TinyDnsBaseSource(BaseSource):
SUPPORTS_GEO = False
SUPPORTS = set(('A', 'CNAME', 'MX', 'NS'))
split_re = re.compile(r':+')
octodns/yaml.py
+4 -17
View File
@@ -5,25 +5,12 @@
from __future__ import absolute_import, division, print_function, \
unicode_literals
from natsort import natsort_keygen
from yaml import SafeDumper, SafeLoader, load, dump
from yaml.constructor import ConstructorError
import re
# zero-padded sort, simplified version of
# https://www.xormedia.com/natural-sort-order-with-zero-padding/
_pad_re = re.compile('\d+')
def _zero_pad(match):
return '{:04d}'.format(int(match.group(0)))
def _zero_padded_numbers(s):
try:
int(s)
except ValueError:
return _pad_re.sub(lambda d: _zero_pad(d), s)
_natsort_key = natsort_keygen()
# Found http://stackoverflow.com/a/21912744 which guided me on how to hook in
@@ -34,7 +21,7 @@ class SortEnforcingLoader(SafeLoader):
self.flatten_mapping(node)
ret = self.construct_pairs(node)
keys = [d[0] for d in ret]
if keys != sorted(keys, key=_zero_padded_numbers):
if keys != sorted(keys, key=_natsort_key):
raise ConstructorError(None, None, "keys out of order: {}"
.format(', '.join(keys)), node.start_mark)
return dict(ret)
@@ -59,7 +46,7 @@ class SortingDumper(SafeDumper):
def _representer(self, data):
data = data.items()
data.sort(key=lambda d: _zero_padded_numbers(d[0]))
data.sort(key=lambda d: _natsort_key(d[0]))
return self.represent_mapping(self.DEFAULT_MAPPING_TAG, data)
requirements.txt
+1
View File
@@ -10,6 +10,7 @@ futures==3.0.5
incf.countryutils==1.0
ipaddress==1.0.18
jmespath==0.9.0
natsort==5.0.3
nsone==0.9.10
python-dateutil==2.6.0
requests==2.13.0
setup.py
+1
View File
@@ -34,6 +34,7 @@ setup(
'futures>=3.0.5',
'incf.countryutils>=1.0',
'ipaddress>=1.0.18',
'natsort>=5.0.3',
'python-dateutil>=2.6.0',
'requests>=2.13.0'
],
tests/helpers.py
+1
View File
@@ -17,6 +17,7 @@ class SimpleSource(object):
class SimpleProvider(object):
SUPPORTS_GEO = False
SUPPORTS = set(('A',))
def __init__(self, id='test'):
pass
tests/test_octodns_provider_base.py
+11 -2
View File
@@ -16,6 +16,8 @@ from octodns.zone import Zone
class HelperProvider(BaseProvider):
log = getLogger('HelperProvider')
SUPPORTS = set(('A',))
def __init__(self, extra_changes, apply_disabled=False,
include_change_callback=None):
self.__extra_changes = extra_changes
@@ -58,10 +60,17 @@ class TestBaseProvider(TestCase):
zone = Zone('unit.tests.', [])
with self.assertRaises(NotImplementedError) as ctx:
HasSupportsGeo('hassupportesgeo').populate(zone)
self.assertEquals('Abstract base class, SUPPORTS property missing',
ctx.exception.message)
class HasSupports(HasSupportsGeo):
SUPPORTS = set(('A',))
with self.assertRaises(NotImplementedError) as ctx:
HasSupports('hassupportes').populate(zone)
self.assertEquals('Abstract base class, populate method missing',
ctx.exception.message)
class HasPopulate(HasSupportsGeo):
class HasPopulate(HasSupports):
def populate(self, zone, target=False):
zone.add_record(Record.new(zone, '', {
@@ -81,7 +90,7 @@ class TestBaseProvider(TestCase):
'value': '1.2.3.4'
}))
self.assertTrue(HasSupportsGeo('hassupportesgeo')
self.assertTrue(HasSupports('hassupportesgeo')
.supports(list(zone.records)[0]))
plan = HasPopulate('haspopulate').plan(zone)
tests/test_octodns_provider_route53.py
+84 -22
View File
@@ -11,8 +11,8 @@ from unittest import TestCase
from mock import patch
from octodns.record import Create, Delete, Record, Update
from octodns.provider.route53 import _Route53Record, Route53Provider, \
_octal_replace
from octodns.provider.route53 import Route53Provider, _Route53GeoDefault, \
_Route53GeoRecord, _Route53Record, _octal_replace
from octodns.zone import Zone
from helpers import GeoProvider
@@ -520,16 +520,6 @@ class TestRoute53Provider(TestCase):
change_resource_record_sets_params = {
'ChangeBatch': {
'Changes': [{
'Action': 'DELETE',
'ResourceRecordSet': {
'GeoLocation': {'ContinentCode': 'OC'},
'Name': 'simple.unit.tests.',
'ResourceRecords': [{'Value': '3.2.3.4'},
{'Value': '4.2.3.4'}],
'SetIdentifier': 'OC',
'TTL': 61,
'Type': 'A'}
}, {
'Action': 'DELETE',
'ResourceRecordSet': {
'GeoLocation': {'CountryCode': '*'},
@@ -539,6 +529,16 @@ class TestRoute53Provider(TestCase):
'SetIdentifier': 'default',
'TTL': 61,
'Type': 'A'}
}, {
'Action': 'DELETE',
'ResourceRecordSet': {
'GeoLocation': {'ContinentCode': 'OC'},
'Name': 'simple.unit.tests.',
'ResourceRecords': [{'Value': '3.2.3.4'},
{'Value': '4.2.3.4'}],
'SetIdentifier': 'OC',
'TTL': 61,
'Type': 'A'}
}, {
'Action': 'CREATE',
'ResourceRecordSet': {
@@ -694,8 +694,7 @@ class TestRoute53Provider(TestCase):
'AF': ['4.2.3.4'],
}
})
id = provider._get_health_check_id(record, 'AF', record.geo['AF'],
True)
id = provider.get_health_check_id(record, 'AF', record.geo['AF'], True)
self.assertEquals('42', id)
def test_health_check_create(self):
@@ -765,13 +764,12 @@ class TestRoute53Provider(TestCase):
})
# if not allowed to create returns none
id = provider._get_health_check_id(record, 'AF', record.geo['AF'],
False)
id = provider.get_health_check_id(record, 'AF', record.geo['AF'],
False)
self.assertFalse(id)
# when allowed to create we do
id = provider._get_health_check_id(record, 'AF', record.geo['AF'],
True)
id = provider.get_health_check_id(record, 'AF', record.geo['AF'], True)
self.assertEquals('42', id)
stubber.assert_no_pending_responses()
@@ -1106,10 +1104,6 @@ class TestRoute53Provider(TestCase):
self.assertEquals(0, len(extra))
stubber.assert_no_pending_responses()
def test_route_53_record(self):
# Just make sure it doesn't blow up
_Route53Record('foo.unit.tests.', 'A', 30).__repr__()
def _get_test_plan(self, max_changes):
provider = Route53Provider('test', 'abc', '123', max_changes)
@@ -1237,3 +1231,71 @@ class TestRoute53Provider(TestCase):
'TTL': 30,
'Type': 'TXT',
}))
class TestRoute53Records(TestCase):
def test_route53_record(self):
existing = Zone('unit.tests.', [])
record_a = Record.new(existing, '', {
'geo': {
'NA-US': ['2.2.2.2', '3.3.3.3'],
'OC': ['4.4.4.4', '5.5.5.5']
},
'ttl': 99,
'type': 'A',
'values': ['9.9.9.9']
})
a = _Route53Record(None, record_a, False)
self.assertEquals(a, a)
b = _Route53Record(None, Record.new(existing, '',
{'ttl': 32, 'type': 'A',
'values': ['8.8.8.8',
'1.1.1.1']}),
False)
self.assertEquals(b, b)
c = _Route53Record(None, Record.new(existing, 'other',
{'ttl': 99, 'type': 'A',
'values': ['9.9.9.9']}),
False)
self.assertEquals(c, c)
d = _Route53Record(None, Record.new(existing, '',
{'ttl': 42, 'type': 'CNAME',
'value': 'foo.bar.'}),
False)
self.assertEquals(d, d)
# Same fqdn & type is same record
self.assertEquals(a, b)
# Same name & different type is not the same
self.assertNotEquals(a, d)
# Different name & same type is not the same
self.assertNotEquals(a, c)
# Same everything, different class is not the same
e = _Route53GeoDefault(None, record_a, False)
self.assertNotEquals(a, e)
class DummyProvider(object):
def get_health_check_id(self, *args, **kwargs):
return None
provider = DummyProvider()
f = _Route53GeoRecord(provider, record_a, 'NA-US',
record_a.geo['NA-US'], False)
self.assertEquals(f, f)
g = _Route53GeoRecord(provider, record_a, 'OC',
record_a.geo['OC'], False)
self.assertEquals(g, g)
# Geo and non-geo are not the same, using Geo as primary to get it's
# __cmp__
self.assertNotEquals(f, a)
# Same everything, different geo's is not the same
self.assertNotEquals(f, g)
# Make sure it doesn't blow up
a.__repr__()
e.__repr__()
f.__repr__()
tests/test_octodns_provider_yaml.py
+6
View File
@@ -100,6 +100,12 @@ class TestYamlProvider(TestCase):
with self.assertRaises(ConstructorError):
source.populate(zone)
source = YamlProvider('test', join(dirname(__file__), 'config'),
enforce_order=False)
# no exception
source.populate(zone)
self.assertEqual(2, len(zone.records))
def test_subzone_handling(self):
source = YamlProvider('test', join(dirname(__file__), 'config'))
tests/test_octodns_yaml.py
+9
View File
@@ -59,3 +59,12 @@ class TestYaml(TestCase):
}, buf)
self.assertEquals("---\n'*.1.1': 42\n'*.2.1': 44\n'*.11.1': 43\n",
buf.getvalue())
# hex sorting isn't ideal, not treated as hex, this make sure we don't
# change the behavior
buf = StringIO()
safe_dump({
'45a03129': 42,
'45a0392a': 43,
}, buf)
self.assertEquals("---\n45a0392a: 43\n45a03129: 42\n", buf.getvalue())