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

Merge pull request #635 from ucc/loc_record_support 

Add LOC record support to OctoDNS for AXFR source, YAML, Cloudflare and PowerDNS providers
pull/679/head
Ross McFarland 5 years ago
committed by GitHub
parent
ed1a3f7f99 cac0a85a42
commit
a63fb06b05
No known key found for this signature in database GPG Key ID: 4AEE18F83AFDEB23
24 changed files with 1114 additions and 104 deletions
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. +2
    -2
      README.md
  2. +1
    -0
      docs/records.md
  3. +73
    -3
      octodns/provider/cloudflare.py
  4. +50
    -2
      octodns/provider/powerdns.py
  5. +1
    -1
      octodns/provider/yaml.py
  6. +190
    -0
      octodns/record/__init__.py
  7. +29
    -2
      octodns/source/axfr.py
  8. +28
    -0
      tests/config/unit.tests.yaml
  9. +3
    -53
      tests/fixtures/cloudflare-dns_records-page-2.json
  10. +128
    -0
      tests/fixtures/cloudflare-dns_records-page-3.json
  11. +16
    -0
      tests/fixtures/powerdns-full-data.json
  12. +7
    -7
      tests/test_octodns_manager.py
  13. +78
    -11
      tests/test_octodns_provider_cloudflare.py
  14. +1
    -1
      tests/test_octodns_provider_constellix.py
  15. +1
    -1
      tests/test_octodns_provider_digitalocean.py
  16. +2
    -2
      tests/test_octodns_provider_dnsimple.py
  17. +1
    -1
      tests/test_octodns_provider_dnsmadeeasy.py
  18. +1
    -1
      tests/test_octodns_provider_easydns.py
  19. +2
    -2
      tests/test_octodns_provider_gandi.py
  20. +3
    -3
      tests/test_octodns_provider_powerdns.py
  21. +5
    -4
      tests/test_octodns_provider_yaml.py
  22. +484
    -4
      tests/test_octodns_record.py
  23. +4
    -4
      tests/test_octodns_source_axfr.py
  24. +4
    -0
      tests/zones/unit.tests.tst

+ 2
- 2
README.md View File

@ -186,7 +186,7 @@ The above command pulled the existing data out of Route53 and placed the results
|--|--|--|--|--|
| [AzureProvider](/octodns/provider/azuredns.py) | azure-mgmt-dns | A, AAAA, CAA, CNAME, MX, NS, PTR, SRV, TXT | No | |
| [Akamai](/octodns/provider/edgedns.py) | edgegrid-python | A, AAAA, CNAME, MX, NAPTR, NS, PTR, SPF, SRV, SSHFP, TXT | No | |
| [CloudflareProvider](/octodns/provider/cloudflare.py) | | A, AAAA, ALIAS, CAA, CNAME, MX, NS, PTR, SPF, SRV, TXT | No | CAA tags restricted |
| [CloudflareProvider](/octodns/provider/cloudflare.py) | | A, AAAA, ALIAS, CAA, CNAME, LOC, MX, NS, PTR, SPF, SRV, TXT | No | CAA tags restricted |
| [ConstellixProvider](/octodns/provider/constellix.py) | | A, AAAA, ALIAS (ANAME), CAA, CNAME, MX, NS, PTR, SPF, SRV, TXT | No | CAA tags restricted |
| [DigitalOceanProvider](/octodns/provider/digitalocean.py) | | A, AAAA, CAA, CNAME, MX, NS, TXT, SRV | No | CAA tags restricted |
| [DnsMadeEasyProvider](/octodns/provider/dnsmadeeasy.py) | | A, AAAA, ALIAS (ANAME), CAA, CNAME, MX, NS, PTR, SPF, SRV, TXT | No | CAA tags restricted |
@ -206,7 +206,7 @@ The above command pulled the existing data out of Route53 and placed the results
| [Selectel](/octodns/provider/selectel.py) | | A, AAAA, CNAME, MX, NS, SPF, SRV, TXT | No | |
| [Transip](/octodns/provider/transip.py) | transip | A, AAAA, CNAME, MX, SRV, SPF, TXT, SSHFP, CAA | No | |
| [UltraDns](/octodns/provider/ultra.py) | | A, AAAA, CAA, CNAME, MX, NS, PTR, SPF, SRV, TXT | No | |
| [AxfrSource](/octodns/source/axfr.py) | | A, AAAA, CAA, CNAME, MX, NS, PTR, SPF, SRV, TXT | No | read-only |
| [AxfrSource](/octodns/source/axfr.py) | | A, AAAA, CAA, CNAME, LOC, MX, NS, PTR, SPF, SRV, TXT | No | read-only |
| [ZoneFileSource](/octodns/source/axfr.py) | | A, AAAA, CAA, CNAME, MX, NS, PTR, SPF, SRV, TXT | No | read-only |
| [TinyDnsFileSource](/octodns/source/tinydns.py) | | A, CNAME, MX, NS, PTR | No | read-only |
| [YamlProvider](/octodns/provider/yaml.py) | | All | Yes | config |


+ 1
- 0
docs/records.md View File

@ -10,6 +10,7 @@ OctoDNS supports the following record types:
* `CAA`
* `CNAME`
* `DNAME`
* `LOC`
* `MX`
* `NAPTR`
* `NS`


+ 73
- 3
octodns/provider/cloudflare.py View File

@ -75,8 +75,8 @@ class CloudflareProvider(BaseProvider):
'''
SUPPORTS_GEO = False
SUPPORTS_DYNAMIC = False
SUPPORTS = set(('ALIAS', 'A', 'AAAA', 'CAA', 'CNAME', 'MX', 'NS', 'PTR',
'SRV', 'SPF', 'TXT'))
SUPPORTS = set(('ALIAS', 'A', 'AAAA', 'CAA', 'CNAME', 'LOC', 'MX', 'NS',
'PTR', 'SRV', 'SPF', 'TXT'))
MIN_TTL = 120
TIMEOUT = 15
@ -133,6 +133,7 @@ class CloudflareProvider(BaseProvider):
timeout=self.TIMEOUT)
self.log.debug('_request: status=%d', resp.status_code)
if resp.status_code == 400:
self.log.debug('_request: data=%s', data)
raise CloudflareError(resp.json())
if resp.status_code == 403:
raise CloudflareAuthenticationError(resp.json())
@ -142,6 +143,11 @@ class CloudflareProvider(BaseProvider):
resp.raise_for_status()
return resp.json()
def _change_keyer(self, change):
key = change.__class__.__name__
order = {'Delete': 0, 'Create': 1, 'Update': 2}
return order[key]
@property
def zones(self):
if self._zones is None:
@ -216,6 +222,30 @@ class CloudflareProvider(BaseProvider):
_data_for_ALIAS = _data_for_CNAME
_data_for_PTR = _data_for_CNAME
def _data_for_LOC(self, _type, records):
values = []
for record in records:
r = record['data']
values.append({
'lat_degrees': int(r['lat_degrees']),
'lat_minutes': int(r['lat_minutes']),
'lat_seconds': float(r['lat_seconds']),
'lat_direction': r['lat_direction'],
'long_degrees': int(r['long_degrees']),
'long_minutes': int(r['long_minutes']),
'long_seconds': float(r['long_seconds']),
'long_direction': r['long_direction'],
'altitude': float(r['altitude']),
'size': float(r['size']),
'precision_horz': float(r['precision_horz']),
'precision_vert': float(r['precision_vert']),
})
return {
'ttl': records[0]['ttl'],
'type': _type,
'values': values
}
def _data_for_MX(self, _type, records):
values = []
for r in records:
@ -384,6 +414,25 @@ class CloudflareProvider(BaseProvider):
_contents_for_PTR = _contents_for_CNAME
def _contents_for_LOC(self, record):
for value in record.values:
yield {
'data': {
'lat_degrees': value.lat_degrees,
'lat_minutes': value.lat_minutes,
'lat_seconds': value.lat_seconds,
'lat_direction': value.lat_direction,
'long_degrees': value.long_degrees,
'long_minutes': value.long_minutes,
'long_seconds': value.long_seconds,
'long_direction': value.long_direction,
'altitude': value.altitude,
'size': value.size,
'precision_horz': value.precision_horz,
'precision_vert': value.precision_vert,
}
}
def _contents_for_MX(self, record):
for value in record.values:
yield {
@ -456,7 +505,7 @@ class CloudflareProvider(BaseProvider):
# new records cleanly. In general when there are multiple records for a
# name & type each will have a distinct/consistent `content` that can
# serve as a unique identifier.
# BUT... there are exceptions. MX, CAA, and SRV don't have a simple
# BUT... there are exceptions. MX, CAA, LOC and SRV don't have a simple
# content as things are currently implemented so we need to handle
# those explicitly and create unique/hashable strings for them.
_type = data['type']
@ -468,6 +517,22 @@ class CloudflareProvider(BaseProvider):
elif _type == 'SRV':
data = data['data']
return '{port} {priority} {target} {weight}'.format(**data)
elif _type == 'LOC':
data = data['data']
loc = (
'{lat_degrees}',
'{lat_minutes}',
'{lat_seconds}',
'{lat_direction}',
'{long_degrees}',
'{long_minutes}',
'{long_seconds}',
'{long_direction}',
'{altitude}',
'{size}',
'{precision_horz}',
'{precision_vert}')
return ' '.join(loc).format(**data)
return data['content']
def _apply_Create(self, change):
@ -616,6 +681,11 @@ class CloudflareProvider(BaseProvider):
self.zones[name] = zone_id
self._zone_records[name] = {}
# Force the operation order to be Delete() -> Create() -> Update()
# This will help avoid problems in updating a CNAME record into an
# A record and vice-versa
changes.sort(key=self._change_keyer)
for change in changes:
class_name = change.__class__.__name__
getattr(self, '_apply_{}'.format(class_name))(change)


+ 50
- 2
octodns/provider/powerdns.py View File

@ -16,8 +16,8 @@ from .base import BaseProvider
class PowerDnsBaseProvider(BaseProvider):
SUPPORTS_GEO = False
SUPPORTS_DYNAMIC = False
SUPPORTS = set(('A', 'AAAA', 'ALIAS', 'CAA', 'CNAME', 'MX', 'NAPTR', 'NS',
'PTR', 'SPF', 'SSHFP', 'SRV', 'TXT'))
SUPPORTS = set(('A', 'AAAA', 'ALIAS', 'CAA', 'CNAME', 'LOC', 'MX', 'NAPTR',
'NS', 'PTR', 'SPF', 'SSHFP', 'SRV', 'TXT'))
TIMEOUT = 5
def __init__(self, id, host, api_key, port=8081,
@ -103,6 +103,33 @@ class PowerDnsBaseProvider(BaseProvider):
_data_for_SPF = _data_for_quoted
_data_for_TXT = _data_for_quoted
def _data_for_LOC(self, rrset):
values = []
for record in rrset['records']:
lat_degrees, lat_minutes, lat_seconds, lat_direction, \
long_degrees, long_minutes, long_seconds, long_direction, \
altitude, size, precision_horz, precision_vert = \
record['content'].replace('m', '').split(' ', 11)
values.append({
'lat_degrees': int(lat_degrees),
'lat_minutes': int(lat_minutes),
'lat_seconds': float(lat_seconds),
'lat_direction': lat_direction,
'long_degrees': int(long_degrees),
'long_minutes': int(long_minutes),
'long_seconds': float(long_seconds),
'long_direction': long_direction,
'altitude': float(altitude),
'size': float(size),
'precision_horz': float(precision_horz),
'precision_vert': float(precision_vert),
})
return {
'ttl': rrset['ttl'],
'type': rrset['type'],
'values': values
}
def _data_for_MX(self, rrset):
values = []
for record in rrset['records']:
@ -286,6 +313,27 @@ class PowerDnsBaseProvider(BaseProvider):
_records_for_SPF = _records_for_quoted
_records_for_TXT = _records_for_quoted
def _records_for_LOC(self, record):
return [{
'content':
'%d %d %0.3f %s %d %d %.3f %s %0.2fm %0.2fm %0.2fm %0.2fm' %
(
int(v.lat_degrees),
int(v.lat_minutes),
float(v.lat_seconds),
v.lat_direction,
int(v.long_degrees),
int(v.long_minutes),
float(v.long_seconds),
v.long_direction,
float(v.altitude),
float(v.size),
float(v.precision_horz),
float(v.precision_vert)
),
'disabled': False
} for v in record.values]
def _records_for_MX(self, record):
return [{
'content': '{} {}'.format(v.preference, v.exchange),


+ 1
- 1
octodns/provider/yaml.py View File

@ -104,7 +104,7 @@ class YamlProvider(BaseProvider):
'''
SUPPORTS_GEO = True
SUPPORTS_DYNAMIC = True
SUPPORTS = set(('A', 'AAAA', 'ALIAS', 'CAA', 'CNAME', 'DNAME', 'MX',
SUPPORTS = set(('A', 'AAAA', 'ALIAS', 'CAA', 'CNAME', 'DNAME', 'LOC', 'MX',
'NAPTR', 'NS', 'PTR', 'SSHFP', 'SPF', 'SRV', 'TXT'))
def __init__(self, id, directory, default_ttl=3600, enforce_order=True,


+ 190
- 0
octodns/record/__init__.py View File

@ -97,6 +97,7 @@ class Record(EqualityTupleMixin):
'CAA': CaaRecord,
'CNAME': CnameRecord,
'DNAME': DnameRecord,
'LOC': LocRecord,
'MX': MxRecord,
'NAPTR': NaptrRecord,
'NS': NsRecord,
@ -879,6 +880,195 @@ class DnameRecord(_DynamicMixin, _ValueMixin, Record):
_value_type = DnameValue
class LocValue(EqualityTupleMixin):
# TODO: work out how to do defaults per RFC
@classmethod
def validate(cls, data, _type):
int_keys = [
'lat_degrees',
'lat_minutes',
'long_degrees',
'long_minutes',
]
float_keys = [
'lat_seconds',
'long_seconds',
'altitude',
'size',
'precision_horz',
'precision_vert',
]
direction_keys = [
'lat_direction',
'long_direction',
]
if not isinstance(data, (list, tuple)):
data = (data,)
reasons = []
for value in data:
for key in int_keys:
try:
int(value[key])
if (
(
key == 'lat_degrees' and
not 0 <= int(value[key]) <= 90
) or (
key == 'long_degrees' and
not 0 <= int(value[key]) <= 180
) or (
key in ['lat_minutes', 'long_minutes'] and
not 0 <= int(value[key]) <= 59
)
):
reasons.append('invalid value for {} "{}"'
.format(key, value[key]))
except KeyError:
reasons.append('missing {}'.format(key))
except ValueError:
reasons.append('invalid {} "{}"'
.format(key, value[key]))
for key in float_keys:
try:
float(value[key])
if (
(
key in ['lat_seconds', 'long_seconds'] and
not 0 <= float(value[key]) <= 59.999
) or (
key == 'altitude' and
not -100000.00 <= float(value[key]) <= 42849672.95
) or (
key in ['size',
'precision_horz',
'precision_vert'] and
not 0 <= float(value[key]) <= 90000000.00
)
):
reasons.append('invalid value for {} "{}"'
.format(key, value[key]))
except KeyError:
reasons.append('missing {}'.format(key))
except ValueError:
reasons.append('invalid {} "{}"'
.format(key, value[key]))
for key in direction_keys:
try:
str(value[key])
if (
key == 'lat_direction' and
value[key] not in ['N', 'S']
):
reasons.append('invalid direction for {} "{}"'
.format(key, value[key]))
if (
key == 'long_direction' and
value[key] not in ['E', 'W']
):
reasons.append('invalid direction for {} "{}"'
.format(key, value[key]))
except KeyError:
reasons.append('missing {}'.format(key))
return reasons
@classmethod
def process(cls, values):
return [LocValue(v) for v in values]
def __init__(self, value):
self.lat_degrees = int(value['lat_degrees'])
self.lat_minutes = int(value['lat_minutes'])
self.lat_seconds = float(value['lat_seconds'])
self.lat_direction = value['lat_direction'].upper()
self.long_degrees = int(value['long_degrees'])
self.long_minutes = int(value['long_minutes'])
self.long_seconds = float(value['long_seconds'])
self.long_direction = value['long_direction'].upper()
self.altitude = float(value['altitude'])
self.size = float(value['size'])
self.precision_horz = float(value['precision_horz'])
self.precision_vert = float(value['precision_vert'])
@property
def data(self):
return {
'lat_degrees': self.lat_degrees,
'lat_minutes': self.lat_minutes,
'lat_seconds': self.lat_seconds,
'lat_direction': self.lat_direction,
'long_degrees': self.long_degrees,
'long_minutes': self.long_minutes,
'long_seconds': self.long_seconds,
'long_direction': self.long_direction,
'altitude': self.altitude,
'size': self.size,
'precision_horz': self.precision_horz,
'precision_vert': self.precision_vert,
}
def __hash__(self):
return hash((
self.lat_degrees,
self.lat_minutes,
self.lat_seconds,
self.lat_direction,
self.long_degrees,
self.long_minutes,
self.long_seconds,
self.long_direction,
self.altitude,
self.size,
self.precision_horz,
self.precision_vert,
))
def _equality_tuple(self):
return (
self.lat_degrees,
self.lat_minutes,
self.lat_seconds,
self.lat_direction,
self.long_degrees,
self.long_minutes,
self.long_seconds,
self.long_direction,
self.altitude,
self.size,
self.precision_horz,
self.precision_vert,
)
def __repr__(self):
loc_format = "'{0} {1} {2:.3f} {3} " + \
"{4} {5} {6:.3f} {7} " + \
"{8:.2f}m {9:.2f}m {10:.2f}m {11:.2f}m'"
return loc_format.format(
self.lat_degrees,
self.lat_minutes,
self.lat_seconds,
self.lat_direction,
self.long_degrees,
self.long_minutes,
self.long_seconds,
self.long_direction,
self.altitude,
self.size,
self.precision_horz,
self.precision_vert,
)
class LocRecord(_ValuesMixin, Record):
_type = 'LOC'
_value_type = LocValue
class MxValue(EqualityTupleMixin):
@classmethod


+ 29
- 2
octodns/source/axfr.py View File

@ -26,8 +26,8 @@ class AxfrBaseSource(BaseSource):
SUPPORTS_GEO = False
SUPPORTS_DYNAMIC = False
SUPPORTS = set(('A', 'AAAA', 'CAA', 'CNAME', 'MX', 'NS', 'PTR', 'SPF',
'SRV', 'TXT'))
SUPPORTS = set(('A', 'AAAA', 'CAA', 'CNAME', 'LOC', 'MX', 'NS', 'PTR',
'SPF', 'SRV', 'TXT'))
def __init__(self, id):
super(AxfrBaseSource, self).__init__(id)
@ -58,6 +58,33 @@ class AxfrBaseSource(BaseSource):
'values': values
}
def _data_for_LOC(self, _type, records):
values = []
for record in records:
lat_degrees, lat_minutes, lat_seconds, lat_direction, \
long_degrees, long_minutes, long_seconds, long_direction, \
altitude, size, precision_horz, precision_vert = \
record['value'].replace('m', '').split(' ', 11)
values.append({
'lat_degrees': lat_degrees,
'lat_minutes': lat_minutes,
'lat_seconds': lat_seconds,
'lat_direction': lat_direction,
'long_degrees': long_degrees,
'long_minutes': long_minutes,
'long_seconds': long_seconds,
'long_direction': long_direction,
'altitude': altitude,
'size': size,
'precision_horz': precision_horz,
'precision_vert': precision_vert,
})
return {
'ttl': records[0]['ttl'],
'type': _type,
'values': values
}
def _data_for_MX(self, _type, records):
values = []
for record in records:


+ 28
- 0
tests/config/unit.tests.yaml View File

@ -77,6 +77,34 @@ included:
- test
type: CNAME
value: unit.tests.
loc:
ttl: 300
type: LOC
values:
- altitude: 20
lat_degrees: 31
lat_direction: S
lat_minutes: 58
lat_seconds: 52.1
long_degrees: 115
long_direction: E
long_minutes: 49
long_seconds: 11.7
precision_horz: 10
precision_vert: 2
size: 10
- altitude: 20
lat_degrees: 53
lat_direction: N
lat_minutes: 13
lat_seconds: 10
long_degrees: 2
long_direction: W
long_minutes: 18
long_seconds: 26
precision_horz: 1000
precision_vert: 2
size: 10
mx:
ttl: 300
type: MX


+ 3
- 53
tests/fixtures/cloudflare-dns_records-page-2.json View File

@ -173,64 +173,14 @@
"meta": {
"auto_added": false
}
},
{
"id": "fc12ab34cd5611334422ab3322997656",
"type": "SRV",
"name": "_srv._tcp.unit.tests",
"data": {
"service": "_srv",
"proto": "_tcp",
"name": "unit.tests",
"priority": 12,
"weight": 20,
"port": 30,
"target": "foo-2.unit.tests"
},
"proxiable": true,
"proxied": false,
"ttl": 600,
"locked": false,
"zone_id": "ff12ab34cd5611334422ab3322997650",
"zone_name": "unit.tests",
"modified_on": "2017-03-11T18:01:43.940682Z",
"created_on": "2017-03-11T18:01:43.940682Z",
"meta": {
"auto_added": false
}
},
{
"id": "fc12ab34cd5611334422ab3322997656",
"type": "SRV",
"name": "_srv._tcp.unit.tests",
"data": {
"service": "_srv",
"proto": "_tcp",
"name": "unit.tests",
"priority": 10,
"weight": 20,
"port": 30,
"target": "foo-1.unit.tests"
},
"proxiable": true,
"proxied": false,
"ttl": 600,
"locked": false,
"zone_id": "ff12ab34cd5611334422ab3322997650",
"zone_name": "unit.tests",
"modified_on": "2017-03-11T18:01:43.940682Z",
"created_on": "2017-03-11T18:01:43.940682Z",
"meta": {
"auto_added": false
}
}
],
"result_info": {
"page": 2,
"per_page": 11,
"total_pages": 2,
"per_page": 10,
"total_pages": 3,
"count": 10,
"total_count": 20
"total_count": 24
},
"success": true,
"errors": [],


+ 128
- 0
tests/fixtures/cloudflare-dns_records-page-3.json View File

@ -0,0 +1,128 @@
{
"result": [
{
"id": "fc12ab34cd5611334422ab3322997656",
"type": "SRV",
"name": "_srv._tcp.unit.tests",
"data": {
"service": "_srv",
"proto": "_tcp",
"name": "unit.tests",
"priority": 12,
"weight": 20,
"port": 30,
"target": "foo-2.unit.tests"
},
"proxiable": true,
"proxied": false,
"ttl": 600,
"locked": false,
"zone_id": "ff12ab34cd5611334422ab3322997650",
"zone_name": "unit.tests",
"modified_on": "2017-03-11T18:01:43.940682Z",
"created_on": "2017-03-11T18:01:43.940682Z",
"meta": {
"auto_added": false
}
},
{
"id": "fc12ab34cd5611334422ab3322997656",
"type": "SRV",
"name": "_srv._tcp.unit.tests",
"data": {
"service": "_srv",
"proto": "_tcp",
"name": "unit.tests",
"priority": 10,
"weight": 20,
"port": 30,
"target": "foo-1.unit.tests"
},
"proxiable": true,
"proxied": false,
"ttl": 600,
"locked": false,
"zone_id": "ff12ab34cd5611334422ab3322997650",
"zone_name": "unit.tests",
"modified_on": "2017-03-11T18:01:43.940682Z",
"created_on": "2017-03-11T18:01:43.940682Z",
"meta": {
"auto_added": false
}
},
{
"id": "372e67954025e0ba6aaa6d586b9e0b59",
"type": "LOC",
"name": "loc.unit.tests",
"content": "IN LOC 31 58 52.1 S 115 49 11.7 E 20m 10m 10m 2m",
"proxiable": true,
"proxied": false,
"ttl": 300,
"locked": false,
"zone_id": "ff12ab34cd5611334422ab3322997650",
"zone_name": "unit.tests",
"created_on": "2020-01-28T05:20:00.12345Z",
"modified_on": "2020-01-28T05:20:00.12345Z",
"data": {
"lat_degrees": 31,
"lat_minutes": 58,
"lat_seconds": 52.1,
"lat_direction": "S",
"long_degrees": 115,
"long_minutes": 49,
"long_seconds": 11.7,
"long_direction": "E",
"altitude": 20,
"size": 10,
"precision_horz": 10,
"precision_vert": 2
},
"meta": {
"auto_added": true,
"source": "primary"
}
},
{
"id": "372e67954025e0ba6aaa6d586b9e0b59",
"type": "LOC",
"name": "loc.unit.tests",
"content": "IN LOC 53 14 10 N 2 18 26 W 20m 10m 1000m 2m",
"proxiable": true,
"proxied": false,
"ttl": 300,
"locked": false,
"zone_id": "ff12ab34cd5611334422ab3322997650",
"zone_name": "unit.tests",
"created_on": "2020-01-28T05:20:00.12345Z",
"modified_on": "2020-01-28T05:20:00.12345Z",
"data": {
"lat_degrees": 53,
"lat_minutes": 13,
"lat_seconds": 10,
"lat_direction": "N",
"long_degrees": 2,
"long_minutes": 18,
"long_seconds": 26,
"long_direction": "W",
"altitude": 20,
"size": 10,
"precision_horz": 1000,
"precision_vert": 2
},
"meta": {
"auto_added": true,
"source": "primary"
}
}
],
"result_info": {
"page": 3,
"per_page": 10,
"total_pages": 3,
"count": 4,
"total_count": 24
},
"success": true,
"errors": [],
"messages": []
}

+ 16
- 0
tests/fixtures/powerdns-full-data.json View File

@ -32,6 +32,22 @@
"ttl": 300,
"type": "MX"
},
{
"comments": [],
"name": "loc.unit.tests.",
"records": [
{
"content": "31 58 52.100 S 115 49 11.700 E 20.00m 10.00m 10.00m 2.00m",
"disabled": false
},
{
"content": "53 13 10.000 N 2 18 26.000 W 20.00m 10.00m 1000.00m 2.00m",
"disabled": false
}
],
"ttl": 300,
"type": "LOC"
},
{
"comments": [],
"name": "sub.unit.tests.",


+ 7
- 7
tests/test_octodns_manager.py View File

@ -118,12 +118,12 @@ class TestManager(TestCase):
environ['YAML_TMP_DIR'] = tmpdir.dirname
tc = Manager(get_config_filename('simple.yaml')) \
.sync(dry_run=False)
self.assertEquals(22, tc)
self.assertEquals(23, tc)
# try with just one of the zones
tc = Manager(get_config_filename('simple.yaml')) \
.sync(dry_run=False, eligible_zones=['unit.tests.'])
self.assertEquals(16, tc)
self.assertEquals(17, tc)
# the subzone, with 2 targets
tc = Manager(get_config_filename('simple.yaml')) \
@ -138,18 +138,18 @@ class TestManager(TestCase):
# Again with force
tc = Manager(get_config_filename('simple.yaml')) \
.sync(dry_run=False, force=True)
self.assertEquals(22, tc)
self.assertEquals(23, tc)
# Again with max_workers = 1
tc = Manager(get_config_filename('simple.yaml'), max_workers=1) \
.sync(dry_run=False, force=True)
self.assertEquals(22, tc)
self.assertEquals(23, tc)
# Include meta
tc = Manager(get_config_filename('simple.yaml'), max_workers=1,
include_meta=True) \
.sync(dry_run=False, force=True)
self.assertEquals(26, tc)
self.assertEquals(27, tc)
def test_eligible_sources(self):
with TemporaryDirectory() as tmpdir:
@ -215,13 +215,13 @@ class TestManager(TestCase):
fh.write('---\n{}')
changes = manager.compare(['in'], ['dump'], 'unit.tests.')
self.assertEquals(16, len(changes))
self.assertEquals(17, len(changes))
# Compound sources with varying support
changes = manager.compare(['in', 'nosshfp'],
['dump'],
'unit.tests.')
self.assertEquals(15, len(changes))
self.assertEquals(16, len(changes))
with self.assertRaises(ManagerException) as ctx:
manager.compare(['nope'], ['dump'], 'unit.tests.')


+ 78
- 11
tests/test_octodns_provider_cloudflare.py View File

@ -177,10 +177,14 @@ class TestCloudflareProvider(TestCase):
'page-2.json') as fh:
mock.get('{}?page=2'.format(base), status_code=200,
text=fh.read())
with open('tests/fixtures/cloudflare-dns_records-'
'page-3.json') as fh:
mock.get('{}?page=3'.format(base), status_code=200,
text=fh.read())
zone = Zone('unit.tests.', [])
provider.populate(zone)
self.assertEquals(13, len(zone.records))
self.assertEquals(14, len(zone.records))
changes = self.expected.changes(zone, provider)
@ -189,7 +193,7 @@ class TestCloudflareProvider(TestCase):
# re-populating the same zone/records comes out of cache, no calls
again = Zone('unit.tests.', [])
provider.populate(again)
self.assertEquals(13, len(again.records))
self.assertEquals(14, len(again.records))
def test_apply(self):
provider = CloudflareProvider('test', 'email', 'token', retry_period=0)
@ -203,12 +207,12 @@ class TestCloudflareProvider(TestCase):
'id': 42,
}
}, # zone create
] + [None] * 22 # individual record creates
] + [None] * 24 # individual record creates
# non-existent zone, create everything
plan = provider.plan(self.expected)
self.assertEquals(13, len(plan.changes))
self.assertEquals(13, provider.apply(plan))
self.assertEquals(14, len(plan.changes))
self.assertEquals(14, provider.apply(plan))
self.assertFalse(plan.exists)
provider._request.assert_has_calls([
@ -234,7 +238,7 @@ class TestCloudflareProvider(TestCase):
}),
], True)
# expected number of total calls
self.assertEquals(23, provider._request.call_count)
self.assertEquals(25, provider._request.call_count)
provider._request.reset_mock()
@ -336,6 +340,10 @@ class TestCloudflareProvider(TestCase):
self.assertTrue(plan.exists)
# creates a the new value and then deletes all the old
provider._request.assert_has_calls([
call('DELETE', '/zones/ff12ab34cd5611334422ab3322997650/'
'dns_records/fc12ab34cd5611334422ab3322997653'),
call('DELETE', '/zones/ff12ab34cd5611334422ab3322997650/'
'dns_records/fc12ab34cd5611334422ab3322997654'),
call('PUT', '/zones/42/dns_records/'
'fc12ab34cd5611334422ab3322997655', data={
'content': '3.2.3.4',
@ -343,11 +351,7 @@ class TestCloudflareProvider(TestCase):
'name': 'ttl.unit.tests',
'proxied': False,
'ttl': 300
}),
call('DELETE', '/zones/ff12ab34cd5611334422ab3322997650/'
'dns_records/fc12ab34cd5611334422ab3322997653'),
call('DELETE', '/zones/ff12ab34cd5611334422ab3322997650/'
'dns_records/fc12ab34cd5611334422ab3322997654')
})
])
def test_update_add_swap(self):
@ -566,6 +570,52 @@ class TestCloudflareProvider(TestCase):
'content': 'foo.bar.com.'
}, list(ptr_record_contents)[0])
def test_loc(self):
self.maxDiff = None
provider = CloudflareProvider('test', 'email', 'token')
zone = Zone('unit.tests.', [])
# LOC record
loc_record = Record.new(zone, 'example', {
'ttl': 300,
'type': 'LOC',
'value': {
'lat_degrees': 31,
'lat_minutes': 58,
'lat_seconds': 52.1,
'lat_direction': 'S',
'long_degrees': 115,
'long_minutes': 49,
'long_seconds': 11.7,
'long_direction': 'E',
'altitude': 20,
'size': 10,
'precision_horz': 10,
'precision_vert': 2,
}
})
loc_record_contents = provider._gen_data(loc_record)
self.assertEquals({
'name': 'example.unit.tests',
'ttl': 300,
'type': 'LOC',
'data': {
'lat_degrees': 31,
'lat_minutes': 58,
'lat_seconds': 52.1,
'lat_direction': 'S',
'long_degrees': 115,
'long_minutes': 49,
'long_seconds': 11.7,
'long_direction': 'E',
'altitude': 20,
'size': 10,
'precision_horz': 10,
'precision_vert': 2,
}
}, list(loc_record_contents)[0])
def test_srv(self):
provider = CloudflareProvider('test', 'email', 'token')
@ -697,6 +747,23 @@ class TestCloudflareProvider(TestCase):
},
'type': 'SRV',
}),
('31 58 52.1 S 115 49 11.7 E 20 10 10 2', {
'data': {
'lat_degrees': 31,
'lat_minutes': 58,
'lat_seconds': 52.1,
'lat_direction': 'S',
'long_degrees': 115,
'long_minutes': 49,
'long_seconds': 11.7,
'long_direction': 'E',
'altitude': 20,
'size': 10,
'precision_horz': 10,
'precision_vert': 2,
},
'type': 'LOC',
}),
):
self.assertEqual(expected, provider._gen_key(data))


+ 1
- 1
tests/test_octodns_provider_constellix.py View File

@ -132,7 +132,7 @@ class TestConstellixProvider(TestCase):
plan = provider.plan(self.expected)
# No root NS, no ignored, no excluded, no unsupported
n = len(self.expected.records) - 6
n = len(self.expected.records) - 7
self.assertEquals(n, len(plan.changes))
self.assertEquals(n, provider.apply(plan))


+ 1
- 1
tests/test_octodns_provider_digitalocean.py View File

@ -163,7 +163,7 @@ class TestDigitalOceanProvider(TestCase):
plan = provider.plan(self.expected)
# No root NS, no ignored, no excluded, no unsupported
n = len(self.expected.records) - 8
n = len(self.expected.records) - 9
self.assertEquals(n, len(plan.changes))
self.assertEquals(n, provider.apply(plan))
self.assertFalse(plan.exists)


+ 2
- 2
tests/test_octodns_provider_dnsimple.py View File

@ -136,8 +136,8 @@ class TestDnsimpleProvider(TestCase):
]
plan = provider.plan(self.expected)
# No root NS, no ignored, no excluded
n = len(self.expected.records) - 4
# No root NS, no ignored, no excluded, no unsupported
n = len(self.expected.records) - 5
self.assertEquals(n, len(plan.changes))
self.assertEquals(n, provider.apply(plan))
self.assertFalse(plan.exists)


+ 1
- 1
tests/test_octodns_provider_dnsmadeeasy.py View File

@ -134,7 +134,7 @@ class TestDnsMadeEasyProvider(TestCase):
plan = provider.plan(self.expected)
# No root NS, no ignored, no excluded, no unsupported
n = len(self.expected.records) - 6
n = len(self.expected.records) - 7
self.assertEquals(n, len(plan.changes))
self.assertEquals(n, provider.apply(plan))


+ 1
- 1
tests/test_octodns_provider_easydns.py View File

@ -374,7 +374,7 @@ class TestEasyDNSProvider(TestCase):
plan = provider.plan(self.expected)
# No root NS, no ignored, no excluded, no unsupported
n = len(self.expected.records) - 7
n = len(self.expected.records) - 8
self.assertEquals(n, len(plan.changes))
self.assertEquals(n, provider.apply(plan))
self.assertFalse(plan.exists)


+ 2
- 2
tests/test_octodns_provider_gandi.py View File

@ -192,8 +192,8 @@ class TestGandiProvider(TestCase):
]
plan = provider.plan(self.expected)
# No root NS, no ignored, no excluded
n = len(self.expected.records) - 4
# No root NS, no ignored, no excluded, no LOC
n = len(self.expected.records) - 5
self.assertEquals(n, len(plan.changes))
self.assertEquals(n, provider.apply(plan))
self.assertFalse(plan.exists)


+ 3
- 3
tests/test_octodns_provider_powerdns.py View File

@ -186,7 +186,7 @@ class TestPowerDnsProvider(TestCase):
source = YamlProvider('test', join(dirname(__file__), 'config'))
source.populate(expected)
expected_n = len(expected.records) - 3
self.assertEquals(16, expected_n)
self.assertEquals(17, expected_n)
# No diffs == no changes
with requests_mock() as mock:
@ -194,7 +194,7 @@ class TestPowerDnsProvider(TestCase):
zone = Zone('unit.tests.', [])
provider.populate(zone)
self.assertEquals(16, len(zone.records))
self.assertEquals(17, len(zone.records))
changes = expected.changes(zone, provider)
self.assertEquals(0, len(changes))
@ -291,7 +291,7 @@ class TestPowerDnsProvider(TestCase):
expected = Zone('unit.tests.', [])
source = YamlProvider('test', join(dirname(__file__), 'config'))
source.populate(expected)
self.assertEquals(19, len(expected.records))
self.assertEquals(20, len(expected.records))
# A small change to a single record
with requests_mock() as mock:


+ 5
- 4
tests/test_octodns_provider_yaml.py View File

@ -35,7 +35,7 @@ class TestYamlProvider(TestCase):
# without it we see everything
source.populate(zone)
self.assertEquals(19, len(zone.records))
self.assertEquals(20, len(zone.records))
source.populate(dynamic_zone)
self.assertEquals(5, len(dynamic_zone.records))
@ -58,12 +58,12 @@ class TestYamlProvider(TestCase):
# We add everything
plan = target.plan(zone)
self.assertEquals(16, len([c for c in plan.changes
self.assertEquals(17, len([c for c in plan.changes
if isinstance(c, Create)]))
self.assertFalse(isfile(yaml_file))
# Now actually do it
self.assertEquals(16, target.apply(plan))
self.assertEquals(17, target.apply(plan))
self.assertTrue(isfile(yaml_file))
# Dynamic plan
@ -87,7 +87,7 @@ class TestYamlProvider(TestCase):
# A 2nd sync should still create everything
plan = target.plan(zone)
self.assertEquals(16, len([c for c in plan.changes
self.assertEquals(17, len([c for c in plan.changes
if isinstance(c, Create)]))
with open(yaml_file) as fh:
@ -106,6 +106,7 @@ class TestYamlProvider(TestCase):
self.assertTrue('values' in data.pop('naptr'))
self.assertTrue('values' in data.pop('sub'))
self.assertTrue('values' in data.pop('txt'))
self.assertTrue('values' in data.pop('loc'))
# these are stored as singular 'value'
self.assertTrue('value' in data.pop('aaaa'))
self.assertTrue('value' in data.pop('cname'))


+ 484
- 4
tests/test_octodns_record.py View File

@ -9,10 +9,11 @@ from six import text_type
from unittest import TestCase
from octodns.record import ARecord, AaaaRecord, AliasRecord, CaaRecord, \
CaaValue, CnameRecord, DnameRecord, Create, Delete, GeoValue, MxRecord, \
MxValue, NaptrRecord, NaptrValue, NsRecord, PtrRecord, Record, \
SshfpRecord, SshfpValue, SpfRecord, SrvRecord, SrvValue, TxtRecord, \
Update, ValidationError, _Dynamic, _DynamicPool, _DynamicRule
CaaValue, CnameRecord, DnameRecord, Create, Delete, GeoValue, LocRecord, \
LocValue, MxRecord, MxValue, NaptrRecord, NaptrValue, NsRecord, \
PtrRecord, Record, SshfpRecord, SshfpValue, SpfRecord, SrvRecord, \
SrvValue, TxtRecord, Update, ValidationError, _Dynamic, _DynamicPool, \
_DynamicRule
from octodns.zone import Zone
from helpers import DynamicProvider, GeoProvider, SimpleProvider
@ -379,6 +380,98 @@ class TestRecord(TestCase):
self.assertSingleValue(DnameRecord, 'target.foo.com.',
'other.foo.com.')
def test_loc(self):
a_values = [{
'lat_degrees': 31,
'lat_minutes': 58,
'lat_seconds': 52.1,
'lat_direction': 'S',
'long_degrees': 115,
'long_minutes': 49,
'long_seconds': 11.7,
'long_direction': 'E',
'altitude': 20,
'size': 10,
'precision_horz': 10,
'precision_vert': 2,
}]
a_data = {'ttl': 30, 'values': a_values}
a = LocRecord(self.zone, 'a', a_data)
self.assertEquals('a', a.name)
self.assertEquals('a.unit.tests.', a.fqdn)
self.assertEquals(30, a.ttl)
self.assertEquals(a_values[0]['lat_degrees'], a.values[0].lat_degrees)
self.assertEquals(a_values[0]['lat_minutes'], a.values[0].lat_minutes)
self.assertEquals(a_values[0]['lat_seconds'], a.values[0].lat_seconds)
self.assertEquals(a_values[0]['lat_direction'],
a.values[0].lat_direction)
self.assertEquals(a_values[0]['long_degrees'],
a.values[0].long_degrees)
self.assertEquals(a_values[0]['long_minutes'],
a.values[0].long_minutes)
self.assertEquals(a_values[0]['long_seconds'],
a.values[0].long_seconds)
self.assertEquals(a_values[0]['long_direction'],
a.values[0].long_direction)
self.assertEquals(a_values[0]['altitude'], a.values[0].altitude)
self.assertEquals(a_values[0]['size'], a.values[0].size)
self.assertEquals(a_values[0]['precision_horz'],
a.values[0].precision_horz)
self.assertEquals(a_values[0]['precision_vert'],
a.values[0].precision_vert)
b_value = {
'lat_degrees': 32,
'lat_minutes': 7,
'lat_seconds': 19,
'lat_direction': 'S',
'long_degrees': 116,
'long_minutes': 2,
'long_seconds': 25,
'long_direction': 'E',
'altitude': 10,
'size': 1,
'precision_horz': 10000,
'precision_vert': 10,
}
b_data = {'ttl': 30, 'value': b_value}
b = LocRecord(self.zone, 'b', b_data)
self.assertEquals(b_value['lat_degrees'], b.values[0].lat_degrees)
self.assertEquals(b_value['lat_minutes'], b.values[0].lat_minutes)
self.assertEquals(b_value['lat_seconds'], b.values[0].lat_seconds)
self.assertEquals(b_value['lat_direction'], b.values[0].lat_direction)
self.assertEquals(b_value['long_degrees'], b.values[0].long_degrees)
self.assertEquals(b_value['long_minutes'], b.values[0].long_minutes)
self.assertEquals(b_value['long_seconds'], b.values[0].long_seconds)
self.assertEquals(b_value['long_direction'],
b.values[0].long_direction)
self.assertEquals(b_value['altitude'], b.values[0].altitude)
self.assertEquals(b_value['size'], b.values[0].size)
self.assertEquals(b_value['precision_horz'],
b.values[0].precision_horz)
self.assertEquals(b_value['precision_vert'],
b.values[0].precision_vert)
self.assertEquals(b_data, b.data)
target = SimpleProvider()
# No changes with self
self.assertFalse(a.changes(a, target))
# Diff in lat_direction causes change
other = LocRecord(self.zone, 'a', {'ttl': 30, 'values': a_values})
other.values[0].lat_direction = 'N'
change = a.changes(other, target)
self.assertEqual(change.existing, a)
self.assertEqual(change.new, other)
# Diff in altitude causes change
other.values[0].altitude = a.values[0].altitude
other.values[0].altitude = -10
change = a.changes(other, target)
self.assertEqual(change.existing, a)
self.assertEqual(change.new, other)
# __repr__ doesn't blow up
a.__repr__()
def test_mx(self):
a_values = [{
'preference': 10,
@ -1127,6 +1220,93 @@ class TestRecord(TestCase):
self.assertTrue(d >= d)
self.assertTrue(d <= d)
def test_loc_value(self):
a = LocValue({
'lat_degrees': 31,
'lat_minutes': 58,
'lat_seconds': 52.1,
'lat_direction': 'S',
'long_degrees': 115,
'long_minutes': 49,
'long_seconds': 11.7,
'long_direction': 'E',
'altitude': 20,
'size': 10,
'precision_horz': 10,
'precision_vert': 2,
})
b = LocValue({
'lat_degrees': 32,
'lat_minutes': 7,
'lat_seconds': 19,
'lat_direction': 'S',
'long_degrees': 116,
'long_minutes': 2,
'long_seconds': 25,
'long_direction': 'E',
'altitude': 10,
'size': 1,
'precision_horz': 10000,
'precision_vert': 10,
})
c = LocValue({
'lat_degrees': 53,
'lat_minutes': 14,
'lat_seconds': 10,
'lat_direction': 'N',
'long_degrees': 2,
'long_minutes': 18,
'long_seconds': 26,
'long_direction': 'W',
'altitude': 10,
'size': 1,
'precision_horz': 1000,
'precision_vert': 10,
})
self.assertEqual(a, a)
self.assertEqual(b, b)
self.assertEqual(c, c)
self.assertNotEqual(a, b)
self.assertNotEqual(a, c)
self.assertNotEqual(b, a)
self.assertNotEqual(b, c)
self.assertNotEqual(c, a)
self.assertNotEqual(c, b)
self.assertTrue(a < b)
self.assertTrue(a < c)
self.assertTrue(b > a)
self.assertTrue(b < c)
self.assertTrue(c > a)
self.assertTrue(c > b)
self.assertTrue(a <= b)
self.assertTrue(a <= c)
self.assertTrue(a <= a)
self.assertTrue(a >= a)
self.assertTrue(b >= a)
self.assertTrue(b <= c)
self.assertTrue(b >= b)
self.assertTrue(b <= b)
self.assertTrue(c >= a)
self.assertTrue(c >= b)
self.assertTrue(c >= c)
self.assertTrue(c <= c)
# Hash
values = set()
values.add(a)
self.assertTrue(a in values)
self.assertFalse(b in values)
values.add(b)
self.assertTrue(b in values)
def test_mx_value(self):
a = MxValue({'preference': 0, 'priority': 'a', 'exchange': 'v',
'value': '1'})
@ -1960,6 +2140,306 @@ class TestRecordValidation(TestCase):
self.assertEquals(['DNAME value "foo.bar.com" missing trailing .'],
ctx.exception.reasons)
def test_LOC(self):
# doesn't blow up
Record.new(self.zone, '', {
'type': 'LOC',
'ttl': 600,
'value': {
'lat_degrees': 31,
'lat_minutes': 58,
'lat_seconds': 52.1,
'lat_direction': 'S',
'long_degrees': 115,
'long_minutes': 49,
'long_seconds': 11.7,
'long_direction': 'E',
'altitude': 20,
'size': 10,
'precision_horz': 10,
'precision_vert': 2,
}
})
# missing int key
with self.assertRaises(ValidationError) as ctx:
Record.new(self.zone, '', {
'type': 'LOC',
'ttl': 600,
'value': {
'lat_minutes': 58,
'lat_seconds': 52.1,
'lat_direction': 'S',
'long_degrees': 115,
'long_minutes': 49,
'long_seconds': 11.7,
'long_direction': 'E',
'altitude': 20,
'size': 10,
'precision_horz': 10,
'precision_vert': 2,
}
})
self.assertEquals(['missing lat_degrees'], ctx.exception.reasons)
# missing float key
with self.assertRaises(ValidationError) as ctx:
Record.new(self.zone, '', {
'type': 'LOC',
'ttl': 600,
'value': {
'lat_degrees': 31,
'lat_minutes': 58,
'lat_direction': 'S',
'long_degrees': 115,
'long_minutes': 49,
'long_seconds': 11.7,
'long_direction': 'E',
'altitude': 20,
'size': 10,
'precision_horz': 10,
'precision_vert': 2,
}
})
self.assertEquals(['missing lat_seconds'], ctx.exception.reasons)
# missing text key
with self.assertRaises(ValidationError) as ctx:
Record.new(self.zone, '', {
'type': 'LOC',
'ttl': 600,
'value': {
'lat_degrees': 31,
'lat_minutes': 58,
'lat_seconds': 52.1,
'long_degrees': 115,
'long_minutes': 49,
'long_seconds': 11.7,
'long_direction': 'E',
'altitude': 20,
'size': 10,
'precision_horz': 10,
'precision_vert': 2,
}
})
self.assertEquals(['missing lat_direction'], ctx.exception.reasons)
# invalid direction
with self.assertRaises(ValidationError) as ctx:
Record.new(self.zone, '', {
'type': 'LOC',
'ttl': 600,
'value': {
'lat_degrees': 31,
'lat_minutes': 58,
'lat_seconds': 52.1,
'lat_direction': 'U',
'long_degrees': 115,
'long_minutes': 49,
'long_seconds': 11.7,
'long_direction': 'E',
'altitude': 20,
'size': 10,
'precision_horz': 10,
'precision_vert': 2,
}
})
self.assertEquals(['invalid direction for lat_direction "U"'],
ctx.exception.reasons)
with self.assertRaises(ValidationError) as ctx:
Record.new(self.zone, '', {
'type': 'LOC',
'ttl': 600,
'value': {
'lat_degrees': 31,
'lat_minutes': 58,
'lat_seconds': 52.1,
'lat_direction': 'S',
'long_degrees': 115,
'long_minutes': 49,
'long_seconds': 11.7,
'long_direction': 'N',
'altitude': 20,
'size': 10,
'precision_horz': 10,
'precision_vert': 2,
}
})
self.assertEquals(['invalid direction for long_direction "N"'],
ctx.exception.reasons)
# invalid degrees
with self.assertRaises(ValidationError) as ctx:
Record.new(self.zone, '', {
'type': 'LOC',
'ttl': 600,
'value': {
'lat_degrees': 360,
'lat_minutes': 58,
'lat_seconds': 52.1,
'lat_direction': 'S',
'long_degrees': 115,
'long_minutes': 49,
'long_seconds': 11.7,
'long_direction': 'E',
'altitude': 20,
'size': 10,
'precision_horz': 10,
'precision_vert': 2,
}
})
self.assertEquals(['invalid value for lat_degrees "360"'],
ctx.exception.reasons)
with self.assertRaises(ValidationError) as ctx:
Record.new(self.zone, '', {
'type': 'LOC',
'ttl': 600,
'value': {
'lat_degrees': 'nope',
'lat_minutes': 58,
'lat_seconds': 52.1,
'lat_direction': 'S',
'long_degrees': 115,
'long_minutes': 49,
'long_seconds': 11.7,
'long_direction': 'E',
'altitude': 20,
'size': 10,
'precision_horz': 10,
'precision_vert': 2,
}
})
self.assertEquals(['invalid lat_degrees "nope"'],
ctx.exception.reasons)
# invalid minutes
with self.assertRaises(ValidationError) as ctx:
Record.new(self.zone, '', {
'type': 'LOC',
'ttl': 600,
'value': {
'lat_degrees': 31,
'lat_minutes': 60,
'lat_seconds': 52.1,
'lat_direction': 'S',
'long_degrees': 115,
'long_minutes': 49,
'long_seconds': 11.7,
'long_direction': 'E',
'altitude': 20,
'size': 10,
'precision_horz': 10,
'precision_vert': 2,
}
})
self.assertEquals(['invalid value for lat_minutes "60"'],
ctx.exception.reasons)
# invalid seconds
with self.assertRaises(ValidationError) as ctx:
Record.new(self.zone, '', {
'type': 'LOC',
'ttl': 600,
'value': {
'lat_degrees': 31,
'lat_minutes': 58,
'lat_seconds': 60,
'lat_direction': 'S',
'long_degrees': 115,
'long_minutes': 49,
'long_seconds': 11.7,
'long_direction': 'E',
'altitude': 20,
'size': 10,
'precision_horz': 10,
'precision_vert': 2,
}
})
self.assertEquals(['invalid value for lat_seconds "60"'],
ctx.exception.reasons)
with self.assertRaises(ValidationError) as ctx:
Record.new(self.zone, '', {
'type': 'LOC',
'ttl': 600,
'value': {
'lat_degrees': 31,
'lat_minutes': 58,
'lat_seconds': 'nope',
'lat_direction': 'S',
'long_degrees': 115,
'long_minutes': 49,
'long_seconds': 11.7,
'long_direction': 'E',
'altitude': 20,
'size': 10,
'precision_horz': 10,
'precision_vert': 2,
}
})
self.assertEquals(['invalid lat_seconds "nope"'],
ctx.exception.reasons)
# invalid altitude
with self.assertRaises(ValidationError) as ctx:
Record.new(self.zone, '', {
'type': 'LOC',
'ttl': 600,
'value': {
'lat_degrees': 31,
'lat_minutes': 58,
'lat_seconds': 52.1,
'lat_direction': 'S',
'long_degrees': 115,
'long_minutes': 49,
'long_seconds': 11.7,
'long_direction': 'E',
'altitude': -666666,
'size': 10,
'precision_horz': 10,
'precision_vert': 2,
}
})
self.assertEquals(['invalid value for altitude "-666666"'],
ctx.exception.reasons)
# invalid size
with self.assertRaises(ValidationError) as ctx:
Record.new(self.zone, '', {
'type': 'LOC',
'ttl': 600,
'value': {
'lat_degrees': 31,
'lat_minutes': 58,
'lat_seconds': 52.1,
'lat_direction': 'S',
'long_degrees': 115,
'long_minutes': 49,
'long_seconds': 11.7,
'long_direction': 'E',
'altitude': 20,
'size': 99999999.99,
'precision_horz': 10,
'precision_vert': 2,
}
})
self.assertEquals(['invalid value for size "99999999.99"'],
ctx.exception.reasons)
def test_MX(self):
# doesn't blow up
Record.new(self.zone, '', {


+ 4
- 4
tests/test_octodns_source_axfr.py View File

@ -36,7 +36,7 @@ class TestAxfrSource(TestCase):
]
self.source.populate(got)
self.assertEquals(12, len(got.records))
self.assertEquals(13, len(got.records))
with self.assertRaises(AxfrSourceZoneTransferFailed) as ctx:
zone = Zone('unit.tests.', [])
@ -73,18 +73,18 @@ class TestZoneFileSource(TestCase):
# Load zonefiles without a specified file extension
valid = Zone('unit.tests.', [])
source.populate(valid)
self.assertEquals(12, len(valid.records))
self.assertEquals(13, len(valid.records))
def test_populate(self):
# Valid zone file in directory
valid = Zone('unit.tests.', [])
self.source.populate(valid)
self.assertEquals(12, len(valid.records))
self.assertEquals(13, len(valid.records))
# 2nd populate does not read file again
again = Zone('unit.tests.', [])
self.source.populate(again)
self.assertEquals(12, len(again.records))
self.assertEquals(13, len(again.records))
# bust the cache
del self.source._zone_records[valid.name]


+ 4
- 0
tests/zones/unit.tests.tst View File

@ -32,6 +32,10 @@ mx 300 IN MX 20 smtp-2.unit.tests.
mx 300 IN MX 30 smtp-3.unit.tests.
mx 300 IN MX 40 smtp-1.unit.tests.
; LOC Records
loc 300 IN LOC 31 58 52.1 S 115 49 11.7 E 20m 10m 10m 2m
loc 300 IN LOC 53 14 10 N 2 18 26 W 20m 10m 1000m 2m
; A Records
@ 300 IN A 1.2.3.4
@ 300 IN A 1.2.3.5


Write Preview
Loading…
Cancel
Save