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

Extract & shim EasyDNSProvider

pull/839/head
Ross McFarland 4 years ago
parent
8f1c713a61
commit
60c188400a
No known key found for this signature in database GPG Key ID: 943B179E15D3B22A
5 changed files with 24 additions and 1165 deletions
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. +1
    -0
      CHANGELOG.md
  2. +1
    -1
      README.md
  3. +17
    -434
      octodns/provider/easydns.py
  4. +0
    -296
      tests/fixtures/easydns-records.json
  5. +5
    -434
      tests/test_octodns_provider_easydns.py

+ 1
- 0
CHANGELOG.md View File

@ -12,6 +12,7 @@
* [DnsimpleProvider](https://github.com/octodns/octodns-dnsimple/) * [DnsimpleProvider](https://github.com/octodns/octodns-dnsimple/)
* [DnsMadeEasyProvider](https://github.com/octodns/octodns-dnsmadeeasy/) * [DnsMadeEasyProvider](https://github.com/octodns/octodns-dnsmadeeasy/)
* [DynProvider](https://github.com/octodns/octodns-dynprovider/) * [DynProvider](https://github.com/octodns/octodns-dynprovider/)
* [EasyDnsProvider](https://github.com/octodns/octodns-easydns/)
* [Ns1Provider](https://github.com/octodns/octodns-ns1/) * [Ns1Provider](https://github.com/octodns/octodns-ns1/)
* [PowerDnsProvider](https://github.com/octodns/octodns-powerdns/) * [PowerDnsProvider](https://github.com/octodns/octodns-powerdns/)
* [Route53Provider](https://github.com/octodns/octodns-route53/) also * [Route53Provider](https://github.com/octodns/octodns-route53/) also


+ 1
- 1
README.md View File

@ -200,7 +200,7 @@ The table below lists the providers octoDNS supports. We're currently in the pro
| [DnsMadeEasyProvider](https://github.com/octodns/octodns-dnsmadeeasy/) | [octodns_dnsmadeeasy](https://github.com/octodns/octodns-dnsmadeeasy/) | | | | | | [DnsMadeEasyProvider](https://github.com/octodns/octodns-dnsmadeeasy/) | [octodns_dnsmadeeasy](https://github.com/octodns/octodns-dnsmadeeasy/) | | | | |
| [DnsimpleProvider](https://github.com/octodns/octodns-dnsimple/) | [octodns_dnsimple](https://github.com/octodns/octodns-dnsimple/) | | | | | | [DnsimpleProvider](https://github.com/octodns/octodns-dnsimple/) | [octodns_dnsimple](https://github.com/octodns/octodns-dnsimple/) | | | | |
| [DynProvider](https://github.com/octodns/octodns-dyn/) (deprecated) | [octodns_dyn](https://github.com/octodns/octodns-dyn/) | | | | | | [DynProvider](https://github.com/octodns/octodns-dyn/) (deprecated) | [octodns_dyn](https://github.com/octodns/octodns-dyn/) | | | | |
| [EasyDNSProvider](/octodns/provider/easydns.py) | | | A, AAAA, CAA, CNAME, MX, NAPTR, NS, SRV, TXT | No | |
| [EasyDnsProvider](https://github.com/octodns/octodns-easydns/) | [octodns_easydns](https://github.com/octodns/octodns-easydns/) | | | | |
| [EtcHostsProvider](/octodns/provider/etc_hosts.py) | | | A, AAAA, ALIAS, CNAME | No | | | [EtcHostsProvider](/octodns/provider/etc_hosts.py) | | | A, AAAA, ALIAS, CNAME | No | |
| [EnvVarSource](/octodns/source/envvar.py) | | | TXT | No | read-only environment variable injection | | [EnvVarSource](/octodns/source/envvar.py) | | | TXT | No | read-only environment variable injection |
| [GandiProvider](/octodns/provider/gandi.py) | | | A, AAAA, ALIAS, CAA, CNAME, DNAME, MX, NS, PTR, SPF, SRV, SSHFP, TXT | No | | | [GandiProvider](/octodns/provider/gandi.py) | | | A, AAAA, ALIAS, CAA, CNAME, DNAME, MX, NS, PTR, SPF, SRV, SSHFP, TXT | No | |


+ 17
- 434
octodns/provider/easydns.py View File

@ -5,437 +5,20 @@
from __future__ import absolute_import, division, print_function, \ from __future__ import absolute_import, division, print_function, \
unicode_literals unicode_literals
from collections import defaultdict
from requests import Session
from time import sleep
import logging
import base64
from ..record import Record
from . import ProviderException
from .base import BaseProvider
class EasyDNSClientException(ProviderException):
pass
class EasyDNSClientBadRequest(EasyDNSClientException):
def __init__(self):
super(EasyDNSClientBadRequest, self).__init__('Bad request')
class EasyDNSClientNotFound(EasyDNSClientException):
def __init__(self):
super(EasyDNSClientNotFound, self).__init__('Not Found')
class EasyDNSClientUnauthorized(EasyDNSClientException):
def __init__(self):
super(EasyDNSClientUnauthorized, self).__init__('Unauthorized')
class EasyDNSClient(object):
# EasyDNS Sandbox API
SANDBOX = 'https://sandbox.rest.easydns.net'
# EasyDNS Live API
LIVE = 'https://rest.easydns.net'
# Default Currency CAD
default_currency = 'CAD'
# Domain Portfolio
domain_portfolio = 'myport'
def __init__(self, token, api_key, currency, portfolio, sandbox,
domain_create_sleep):
self.log = logging.getLogger(f'EasyDNSProvider[{id}]')
self.default_currency = currency
self.domain_portfolio = portfolio
self.domain_create_sleep = domain_create_sleep
auth_key = f'{token}:{api_key}'
auth_key = base64.b64encode(auth_key.encode("utf-8"))
auth_key = auth_key.decode('utf-8')
self.base_path = self.SANDBOX if sandbox else self.LIVE
sess = Session()
sess.headers.update({'Authorization': f'Basic {auth_key}'})
sess.headers.update({'accept': 'application/json'})
self._sess = sess
def _request(self, method, path, params=None, data=None):
url = f'{self.base_path}{path}'
resp = self._sess.request(method, url, params=params, json=data)
if resp.status_code == 400:
self.log.debug('Response code 400, path=%s', path)
if method == 'GET' and path[:8] == '/domain/':
raise EasyDNSClientNotFound()
raise EasyDNSClientBadRequest()
if resp.status_code == 401:
raise EasyDNSClientUnauthorized()
if resp.status_code == 403 or resp.status_code == 404:
raise EasyDNSClientNotFound()
resp.raise_for_status()
return resp
def domain(self, name):
path = f'/domain/{name}'
return self._request('GET', path).json()
def domain_create(self, name):
# EasyDNS allows for new domains to be created for the purpose of DNS
# only, or with domain registration. This function creates a DNS only
# record expectig the domain to be registered already
path = f'/domains/add/{name}'
domain_data = {'service': 'dns',
'term': 1,
'dns_only': 1,
'portfolio': self.domain_portfolio,
'currency': self.default_currency}
self._request('PUT', path, data=domain_data).json()
# EasyDNS creates default records for MX, A and CNAME for new domains,
# we need to delete those default record so we can sync with the source
# records, first we'll sleep for a second before gathering new records
# We also create default NS records, but they won't be deleted
sleep(self.domain_create_sleep)
records = self.records(name, True)
for record in records:
if record['host'] in ('', 'www') \
and record['type'] in ('A', 'MX', 'CNAME'):
self.record_delete(name, record['id'])
def records(self, zone_name, raw=False):
if raw:
path = f'/zones/records/all/{zone_name}'
else:
path = f'/zones/records/parsed/{zone_name}'
ret = []
resp = self._request('GET', path).json()
ret += resp['data']
for record in ret:
# change any apex record to empty string
if record['host'] == '@':
record['host'] = ''
# change any apex value to zone name
if record['rdata'] == '@':
record['rdata'] = f'{zone_name}.'
return ret
def record_create(self, zone_name, params):
path = f'/zones/records/add/{zone_name}/{params["type"]}'
# change empty name string to @, EasyDNS uses @ for apex record names
params['host'] = params['name']
if params['host'] == '':
params['host'] = '@'
self._request('PUT', path, data=params)
def record_delete(self, zone_name, record_id):
path = f'/zones/records/{zone_name}/{record_id}'
self._request('DELETE', path)
class EasyDNSProvider(BaseProvider):
'''
EasyDNS provider using API v3
easydns:
class: octodns.provider.easydns.EasyDNSProvider
# Your EasyDNS API token (required)
token: foo
# Your EasyDNS API Key (required)
api_key: bar
# Use SandBox or Live environment, optional, defaults to live
sandbox: False
# Currency to use for creating domains, default CAD
default_currency: CAD
# Domain Portfolio under which to create domains
portfolio: myport
'''
SUPPORTS_GEO = False
SUPPORTS_DYNAMIC = False
SUPPORTS = set(('A', 'AAAA', 'CAA', 'CNAME', 'MX', 'NS', 'TXT',
'SRV', 'NAPTR'))
def __init__(self, id, token, api_key, currency='CAD', portfolio='myport',
sandbox=False, domain_create_sleep=1, *args, **kwargs):
self.log = logging.getLogger(f'EasyDNSProvider[{id}]')
self.log.debug('__init__: id=%s, token=***', id)
super(EasyDNSProvider, self).__init__(id, *args, **kwargs)
self._client = EasyDNSClient(token, api_key, currency, portfolio,
sandbox, domain_create_sleep)
self._zone_records = {}
def _data_for_multiple(self, _type, records):
return {
'ttl': records[0]['ttl'],
'type': _type,
'values': [r['rdata'] for r in records]
}
_data_for_A = _data_for_multiple
_data_for_AAAA = _data_for_multiple
def _data_for_CAA(self, _type, records):
values = []
for record in records:
try:
flags, tag, value = record['rdata'].split(' ', 2)
except ValueError:
continue
values.append({
'flags': flags,
'tag': tag,
'value': value,
})
return {
'ttl': records[0]['ttl'],
'type': _type,
'values': values
}
def _data_for_NAPTR(self, _type, records):
values = []
for record in records:
try:
order, preference, flags, service, regexp, replacement = \
record['rdata'].split(' ', 5)
except ValueError:
continue
values.append({
'flags': flags[1:-1],
'order': order,
'preference': preference,
'regexp': regexp[1:-1],
'replacement': replacement,
'service': service[1:-1],
})
return {
'type': _type,
'ttl': records[0]['ttl'],
'values': values
}
def _data_for_CNAME(self, _type, records):
record = records[0]
return {
'ttl': record['ttl'],
'type': _type,
'value': str(record['rdata'])
}
def _data_for_MX(self, _type, records):
values = []
for record in records:
values.append({
'preference': record['prio'],
'exchange': str(record['rdata'])
})
return {
'ttl': records[0]['ttl'],
'type': _type,
'values': values
}
def _data_for_NS(self, _type, records):
values = []
for record in records:
data = str(record['rdata'])
values.append(data)
return {
'ttl': records[0]['ttl'],
'type': _type,
'values': values,
}
def _data_for_SRV(self, _type, records):
values = []
record = records[0]
for record in records:
try:
priority, weight, port, target = record['rdata'].split(' ', 3)
except ValueError:
rdata = record['rdata'].split(' ', 3)
priority = 0
weight = 0
port = 0
target = ''
if len(rdata) != 0 and rdata[0] != '':
priority = rdata[0]
if len(rdata) >= 2:
weight = rdata[1]
if len(rdata) >= 3:
port = rdata[2]
values.append({
'port': int(port),
'priority': int(priority),
'target': target,
'weight': int(weight)
})
return {
'type': _type,
'ttl': records[0]['ttl'],
'values': values
}
def _data_for_TXT(self, _type, records):
values = ['"' + value['rdata'].replace(';', '\\;') +
'"' for value in records]
return {
'ttl': records[0]['ttl'],
'type': _type,
'values': values
}
def zone_records(self, zone):
if zone.name not in self._zone_records:
try:
self._zone_records[zone.name] = \
self._client.records(zone.name[:-1])
except EasyDNSClientNotFound:
return []
return self._zone_records[zone.name]
def populate(self, zone, target=False, lenient=False):
self.log.debug('populate: name=%s, target=%s, lenient=%s', zone.name,
target, lenient)
values = defaultdict(lambda: defaultdict(list))
for record in self.zone_records(zone):
_type = record['type']
if _type not in self.SUPPORTS:
self.log.warning('populate: skipping unsupported %s record',
_type)
continue
values[record['host']][record['type']].append(record)
before = len(zone.records)
for name, types in values.items():
for _type, records in types.items():
data_for = getattr(self, f'_data_for_{_type}')
record = Record.new(zone, name, data_for(_type, records),
source=self, lenient=lenient)
zone.add_record(record, lenient=lenient)
exists = zone.name in self._zone_records
self.log.info('populate: found %s records, exists=%s',
len(zone.records) - before, exists)
return exists
def _params_for_multiple(self, record):
for value in record.values:
yield {
'rdata': value,
'name': record.name,
'ttl': record.ttl,
'type': record._type
}
_params_for_A = _params_for_multiple
_params_for_AAAA = _params_for_multiple
_params_for_NS = _params_for_multiple
def _params_for_CAA(self, record):
for value in record.values:
yield {
'rdata': f"{value.flags} {value.tag} {value.value}",
'name': record.name,
'ttl': record.ttl,
'type': record._type
}
def _params_for_NAPTR(self, record):
for value in record.values:
content = f'{value.order} {value.preference} "{value.flags}" ' \
f'"{value.service}" "{value.regexp}" {value.replacement}'
yield {
'rdata': content,
'name': record.name,
'ttl': record.ttl,
'type': record._type
}
def _params_for_single(self, record):
yield {
'rdata': record.value,
'name': record.name,
'ttl': record.ttl,
'type': record._type
}
_params_for_CNAME = _params_for_single
def _params_for_MX(self, record):
for value in record.values:
yield {
'rdata': value.exchange,
'name': record.name,
'prio': value.preference,
'ttl': record.ttl,
'type': record._type
}
def _params_for_SRV(self, record):
for value in record.values:
yield {
'rdata': f"{value.priority} {value.port} {value.weight} "
f"{value.target}",
'name': record.name,
'ttl': record.ttl,
'type': record._type,
}
def _params_for_TXT(self, record):
for value in record.values:
yield {
'rdata': '"' + value.replace('\\;', ';') + '"',
'name': record.name,
'ttl': record.ttl,
'type': record._type
}
def _apply_Create(self, change):
new = change.new
params_for = getattr(self, f'_params_for_{new._type}')
for params in params_for(new):
self._client.record_create(new.zone.name[:-1], params)
def _apply_Update(self, change):
self._apply_Delete(change)
self._apply_Create(change)
def _apply_Delete(self, change):
existing = change.existing
zone = existing.zone
for record in self.zone_records(zone):
self.log.debug('apply_Delete: zone=%s, type=%s, host=%s', zone,
record['type'], record['host'])
if existing.name == record['host'] and \
existing._type == record['type']:
self._client.record_delete(zone.name[:-1], record['id'])
def _apply(self, plan):
desired = plan.desired
changes = plan.changes
self.log.debug('_apply: zone=%s, len(changes)=%d', desired.name,
len(changes))
domain_name = desired.name[:-1]
try:
self._client.domain(domain_name)
except EasyDNSClientNotFound:
self.log.debug('_apply: no matching zone, creating domain')
self._client.domain_create(domain_name)
for change in changes:
class_name = change.__class__.__name__
getattr(self, f'_apply_{class_name}')(change)
# Clear out the cache if any
self._zone_records.pop(desired.name, None)
from logging import getLogger
logger = getLogger('EasyDns')
try:
logger.warn('octodns_easydns shimmed. Update your provider class to '
'octodns_easydns.EasyDnsProvider. '
'Shim will be removed in 1.0')
from octodns_easydns import EasyDnsProvider, EasyDNSProvider
EasyDnsProvider # pragma: no cover
EasyDNSProvider # pragma: no cover
except ModuleNotFoundError:
logger.exception('EasyDNSProvider has been moved into a seperate module, '
'octodns_easydns is now required. Provider class should '
'be updated to octodns_easydns.EasyDnsProvider. See '
'https://github.com/octodns/octodns/README.md#updating-'
'to-use-extracted-providers for more information.')
raise

+ 0
- 296
tests/fixtures/easydns-records.json View File

@ -1,296 +0,0 @@
{
"tm": 1000000000,
"data": [
{
"id": "12340001",
"domain": "unit.tests",
"host": "@",
"ttl": "3600",
"prio": "0",
"type": "SOA",
"rdata": "dns1.easydns.com. zone.easydns.com. 2020010101 3600 600 604800 0",
"geozone_id": "0",
"last_mod": "2020-01-01 01:01:01"
},
{
"id": "12340002",
"domain": "unit.tests",
"host": "@",
"ttl": "300",
"prio": "0",
"type": "A",
"rdata": "1.2.3.4",
"geozone_id": "0",
"last_mod": "2020-01-01 01:01:01"
},
{
"id": "12340003",
"domain": "unit.tests",
"host": "@",
"ttl": "300",
"prio": "0",
"type": "A",
"rdata": "1.2.3.5",
"geozone_id": "0",
"last_mod": "2020-01-01 01:01:01"
},
{
"id": "12340004",
"domain": "unit.tests",
"host": "@",
"ttl": "0",
"prio": null,
"type": "NS",
"rdata": "6.2.3.4.",
"geozone_id": "0",
"last_mod": "2020-01-01 01:01:01"
},
{
"id": "12340005",
"domain": "unit.tests",
"host": "@",
"ttl": "0",
"prio": null,
"type": "NS",
"rdata": "7.2.3.4.",
"geozone_id": "0",
"last_mod": "2020-01-01 01:01:01"
},
{
"id": "12340006",
"domain": "unit.tests",
"host": "@",
"ttl": "3600",
"prio": "0",
"type": "CAA",
"rdata": "0 issue ca.unit.tests",
"geozone_id": "0",
"last_mod": "2020-01-01 01:01:01"
},
{
"id": "12340007",
"domain": "unit.tests",
"host": "_srv._tcp",
"ttl": "600",
"prio": "12",
"type": "SRV",
"rdata": "12 20 30 foo-2.unit.tests.",
"geozone_id": "0",
"last_mod": "2020-01-01 01:01:01"
},
{
"id": "12340008",
"domain": "unit.tests",
"host": "_srv._tcp",
"ttl": "600",
"prio": "12",
"type": "SRV",
"rdata": "10 20 30 foo-1.unit.tests.",
"geozone_id": "0",
"last_mod": "2020-01-01 01:01:01"
},
{
"id": "12340009",
"domain": "unit.tests",
"host": "aaaa",
"ttl": "600",
"prio": "0",
"type": "AAAA",
"rdata": "2601:644:500:e210:62f8:1dff:feb8:947a",
"geozone_id": "0",
"last_mod": "2020-01-01 01:01:01"
},
{
"id": "12340010",
"domain": "unit.tests",
"host": "cname",
"ttl": "300",
"prio": null,
"type": "CNAME",
"rdata": "@",
"geozone_id": "0",
"last_mod": "2020-01-01 01:01:01"
},
{
"id": "12340012",
"domain": "unit.tests",
"host": "mx",
"ttl": "300",
"prio": "10",
"type": "MX",
"rdata": "smtp-4.unit.tests.",
"geozone_id": "0",
"last_mod": "2020-01-01 01:01:01"
},
{
"id": "12340013",
"domain": "unit.tests",
"host": "mx",
"ttl": "300",
"prio": "20",
"type": "MX",
"rdata": "smtp-2.unit.tests.",
"geozone_id": "0",
"last_mod": "2020-01-01 01:01:01"
},
{
"id": "12340014",
"domain": "unit.tests",
"host": "mx",
"ttl": "300",
"prio": "30",
"type": "MX",
"rdata": "smtp-3.unit.tests.",
"geozone_id": "0",
"last_mod": "2020-01-01 01:01:01"
},
{
"id": "12340015",
"domain": "unit.tests",
"host": "mx",
"ttl": "300",
"prio": "40",
"type": "MX",
"rdata": "smtp-1.unit.tests.",
"geozone_id": "0",
"last_mod": "2020-01-01 01:01:01"
},
{
"id": "12340016",
"domain": "unit.tests",
"host": "naptr",
"ttl": "600",
"prio": null,
"type": "NAPTR",
"rdata": "100 100 'U' 'SIP+D2U' '!^.*$!sip:info@bar.example.com!' .",
"geozone_id": "0",
"last_mod": "2020-01-01 01:01:01"
},
{
"id": "12340017",
"domain": "unit.tests",
"host": "naptr",
"ttl": "600",
"prio": null,
"type": "NAPTR",
"rdata": "10 100 'S' 'SIP+D2U' '!^.*$!sip:info@bar.example.com!' .",
"geozone_id": "0",
"last_mod": "2020-01-01 01:01:01"
},
{
"id": "12340018",
"domain": "unit.tests",
"host": "sub",
"ttl": "3600",
"prio": null,
"type": "NS",
"rdata": "6.2.3.4.",
"geozone_id": "0",
"last_mod": "2020-01-01 01:01:01"
},
{
"id": "12340019",
"domain": "unit.tests",
"host": "sub",
"ttl": "0",
"prio": null,
"type": "NS",
"rdata": "7.2.3.4.",
"geozone_id": "0",
"last_mod": "2020-01-01 01:01:01"
},
{
"id": "12340020",
"domain": "unit.tests",
"host": "www",
"ttl": "300",
"prio": "0",
"type": "A",
"rdata": "2.2.3.6",
"geozone_id": "0",
"last_mod": "2020-01-01 01:01:01"
},
{
"id": "12340021",
"domain": "unit.tests",
"host": "www.sub",
"ttl": "300",
"prio": "0",
"type": "A",
"rdata": "2.2.3.6",
"geozone_id": "0",
"last_mod": "2020-01-01 01:01:01"
},
{
"id": "12340022",
"domain": "unit.tests",
"host": "included",
"ttl": "3600",
"prio": null,
"type": "CNAME",
"rdata": "unit.tests.",
"geozone_id": "0",
"last_mod": "2020-01-01 01:01:01"
},
{
"id": "12340011",
"domain": "unit.tests",
"host": "txt",
"ttl": "600",
"prio": "0",
"type": "TXT",
"rdata": "Bah bah black sheep",
"geozone_id": "0",
"last_mod": "2020-01-01 01:01:01"
},
{
"id": "12340023",
"domain": "unit.tests",
"host": "txt",
"ttl": "600",
"prio": "0",
"type": "TXT",
"rdata": "have you any wool.",
"geozone_id": "0",
"last_mod": "2020-01-01 01:01:01"
},
{
"id": "12340024",
"domain": "unit.tests",
"host": "txt",
"ttl": "600",
"prio": "0",
"type": "TXT",
"rdata": "v=DKIM1;k=rsa;s=email;h=sha256;p=A\/kinda+of\/long\/string+with+numb3rs",
"geozone_id": "0",
"last_mod": "2020-01-01 01:01:01"
},
{
"id": "12340025",
"domain": "unit.tests",
"host": "_imap._tcp",
"ttl": "600",
"prio": "0",
"type": "SRV",
"rdata": "0 0 0 .",
"geozone_id": "0",
"last_mod": "2020-01-01 01:01:01"
},
{
"id": "12340026",
"domain": "unit.tests",
"host": "_pop3._tcp",
"ttl": "600",
"prio": "0",
"type": "SRV",
"rdata": "0 0 0 .",
"geozone_id": "0",
"last_mod": "2020-01-01 01:01:01"
}
],
"count": 26,
"total": 26,
"start": 0,
"max": 1000,
"status": 200
}

+ 5
- 434
tests/test_octodns_provider_easydns.py View File

@ -5,441 +5,12 @@
from __future__ import absolute_import, division, print_function, \ from __future__ import absolute_import, division, print_function, \
unicode_literals unicode_literals
import json
from mock import Mock, call
from os.path import dirname, join
from requests import HTTPError
from requests_mock import ANY, mock as requests_mock
from unittest import TestCase from unittest import TestCase
from octodns.record import Record
from octodns.provider.easydns import EasyDNSClientNotFound, \
EasyDNSProvider
from octodns.provider.yaml import YamlProvider
from octodns.zone import Zone
class TestEasyDnsShim(TestCase):
class TestEasyDNSProvider(TestCase):
expected = Zone('unit.tests.', [])
source = YamlProvider('test', join(dirname(__file__), 'config'))
source.populate(expected)
def test_populate(self):
provider = EasyDNSProvider('test', 'token', 'apikey')
# Bad auth
with requests_mock() as mock:
mock.get(ANY, status_code=401,
text='{"id":"unauthorized",'
'"message":"Unable to authenticate you."}')
with self.assertRaises(Exception) as ctx:
zone = Zone('unit.tests.', [])
provider.populate(zone)
self.assertEquals('Unauthorized', str(ctx.exception))
# Bad request
with requests_mock() as mock:
mock.get(ANY, status_code=400,
text='{"id":"invalid",'
'"message":"Bad request"}')
with self.assertRaises(Exception) as ctx:
zone = Zone('unit.tests.', [])
provider.populate(zone)
self.assertEquals('Bad request', str(ctx.exception))
# General error
with requests_mock() as mock:
mock.get(ANY, status_code=502, text='Things caught fire')
with self.assertRaises(HTTPError) as ctx:
zone = Zone('unit.tests.', [])
provider.populate(zone)
self.assertEquals(502, ctx.exception.response.status_code)
# Non-existent zone doesn't populate anything
with requests_mock() as mock:
mock.get(ANY, status_code=404,
text='{"id":"not_found","message":"The resource you '
'were accessing could not be found."}')
zone = Zone('unit.tests.', [])
provider.populate(zone)
self.assertEquals(set(), zone.records)
# No diffs == no changes
with requests_mock() as mock:
base = 'https://rest.easydns.net/zones/records/'
with open('tests/fixtures/easydns-records.json') as fh:
mock.get(f'{base}parsed/unit.tests', text=fh.read())
with open('tests/fixtures/easydns-records.json') as fh:
mock.get(f'{base}all/unit.tests', text=fh.read())
provider.populate(zone)
self.assertEquals(15, len(zone.records))
changes = self.expected.changes(zone, provider)
self.assertEquals(0, len(changes))
# 2nd populate makes no network calls/all from cache
again = Zone('unit.tests.', [])
provider.populate(again)
self.assertEquals(15, len(again.records))
# bust the cache
del provider._zone_records[zone.name]
def test_domain(self):
provider = EasyDNSProvider('test', 'token', 'apikey')
with requests_mock() as mock:
base = 'https://rest.easydns.net/'
mock.get(f'{base}domain/unit.tests', status_code=400,
text='{"id":"not_found","message":"The resource you '
'were accessing could not be found."}')
with self.assertRaises(Exception) as ctx:
provider._client.domain('unit.tests')
self.assertEquals('Not Found', str(ctx.exception))
def test_apply_not_found(self):
provider = EasyDNSProvider('test', 'token', 'apikey',
domain_create_sleep=0)
wanted = Zone('unit.tests.', [])
wanted.add_record(Record.new(wanted, 'test1', {
"name": "test1",
"ttl": 300,
"type": "A",
"value": "1.2.3.4",
}))
with requests_mock() as mock:
base = 'https://rest.easydns.net/'
mock.get(f'{base}domain/unit.tests', status_code=404,
text='{"id":"not_found","message":"The resource you '
'were accessing could not be found."}')
mock.put(f'{base}domains/add/unit.tests', status_code=200,
text='{"id":"OK","message":"Zone created."}')
mock.get(f'{base}zones/records/parsed/unit.tests',
status_code=404,
text='{"id":"not_found","message":"The resource you '
'were accessing could not be found."}')
mock.get(f'{base}zones/records/all/unit.tests', status_code=404,
text='{"id":"not_found","message":"The resource you '
'were accessing could not be found."}')
plan = provider.plan(wanted)
self.assertFalse(plan.exists)
self.assertEquals(1, len(plan.changes))
with self.assertRaises(Exception) as ctx:
provider.apply(plan)
self.assertEquals('Not Found', str(ctx.exception))
def test_domain_create(self):
provider = EasyDNSProvider('test', 'token', 'apikey',
domain_create_sleep=0)
domain_after_creation = {
"tm": 1000000000,
"data": [{
"id": "12341001",
"domain": "unit.tests",
"host": "@",
"ttl": "0",
"prio": "0",
"type": "SOA",
"rdata": "dns1.easydns.com. zone.easydns.com. "
"2020010101 3600 600 604800 0",
"geozone_id": "0",
"last_mod": "2020-01-01 01:01:01"
}, {
"id": "12341002",
"domain": "unit.tests",
"host": "@",
"ttl": "0",
"prio": "0",
"type": "NS",
"rdata": "LOCAL.",
"geozone_id": "0",
"last_mod": "2020-01-01 01:01:01"
}, {
"id": "12341003",
"domain": "unit.tests",
"host": "@",
"ttl": "0",
"prio": "0",
"type": "MX",
"rdata": "LOCAL.",
"geozone_id": "0",
"last_mod": "2020-01-01 01:01:01"
}],
"count": 3,
"total": 3,
"start": 0,
"max": 1000,
"status": 200
}
with requests_mock() as mock:
base = 'https://rest.easydns.net/'
mock.put(f'{base}domains/add/unit.tests',
status_code=201, text='{"id":"OK"}')
mock.get(f'{base}zones/records/all/unit.tests',
text=json.dumps(domain_after_creation))
mock.delete(ANY, text='{"id":"OK"}')
provider._client.domain_create('unit.tests')
def test_caa(self):
provider = EasyDNSProvider('test', 'token', 'apikey')
# Invalid rdata records
caa_record_invalid = [{
"domain": "unit.tests",
"host": "@",
"ttl": "3600",
"prio": "0",
"type": "CAA",
"rdata": "0",
}]
# Valid rdata records
caa_record_valid = [{
"domain": "unit.tests",
"host": "@",
"ttl": "3600",
"prio": "0",
"type": "CAA",
"rdata": "0 issue ca.unit.tests",
}]
provider._data_for_CAA('CAA', caa_record_invalid)
provider._data_for_CAA('CAA', caa_record_valid)
def test_naptr(self):
provider = EasyDNSProvider('test', 'token', 'apikey')
# Invalid rdata records
naptr_record_invalid = [{
"domain": "unit.tests",
"host": "naptr",
"ttl": "600",
"prio": "10",
"type": "NAPTR",
"rdata": "100",
}]
# Valid rdata records
naptr_record_valid = [{
"domain": "unit.tests",
"host": "naptr",
"ttl": "600",
"prio": "10",
"type": "NAPTR",
"rdata": "10 10 'U' 'SIP+D2U' '!^.*$!sip:info@bar.example.com!' .",
}]
provider._data_for_NAPTR('NAPTR', naptr_record_invalid)
provider._data_for_NAPTR('NAPTR', naptr_record_valid)
def test_srv(self):
provider = EasyDNSProvider('test', 'token', 'apikey')
# Invalid rdata records
srv_invalid = [{
"domain": "unit.tests",
"host": "_srv._tcp",
"ttl": "600",
"type": "SRV",
"rdata": "",
}]
srv_invalid2 = [{
"domain": "unit.tests",
"host": "_srv._tcp",
"ttl": "600",
"type": "SRV",
"rdata": "11",
}]
srv_invalid3 = [{
"domain": "unit.tests",
"host": "_srv._tcp",
"ttl": "600",
"type": "SRV",
"rdata": "12 30",
}]
srv_invalid4 = [{
"domain": "unit.tests",
"host": "_srv._tcp",
"ttl": "600",
"type": "SRV",
"rdata": "13 40 1234",
}]
# Valid rdata
srv_valid = [{
"domain": "unit.tests",
"host": "_srv._tcp",
"ttl": "600",
"type": "SRV",
"rdata": "100 20 5678 foo-2.unit.tests.",
}]
srv_invalid_content = provider._data_for_SRV('SRV', srv_invalid)
srv_invalid_content2 = provider._data_for_SRV('SRV', srv_invalid2)
srv_invalid_content3 = provider._data_for_SRV('SRV', srv_invalid3)
srv_invalid_content4 = provider._data_for_SRV('SRV', srv_invalid4)
srv_valid_content = provider._data_for_SRV('SRV', srv_valid)
self.assertEqual(srv_valid_content['values'][0]['priority'], 100)
self.assertEqual(srv_invalid_content['values'][0]['priority'], 0)
self.assertEqual(srv_invalid_content2['values'][0]['priority'], 11)
self.assertEqual(srv_invalid_content3['values'][0]['priority'], 12)
self.assertEqual(srv_invalid_content4['values'][0]['priority'], 13)
self.assertEqual(srv_valid_content['values'][0]['weight'], 20)
self.assertEqual(srv_invalid_content['values'][0]['weight'], 0)
self.assertEqual(srv_invalid_content2['values'][0]['weight'], 0)
self.assertEqual(srv_invalid_content3['values'][0]['weight'], 30)
self.assertEqual(srv_invalid_content4['values'][0]['weight'], 40)
self.assertEqual(srv_valid_content['values'][0]['port'], 5678)
self.assertEqual(srv_invalid_content['values'][0]['port'], 0)
self.assertEqual(srv_invalid_content2['values'][0]['port'], 0)
self.assertEqual(srv_invalid_content3['values'][0]['port'], 0)
self.assertEqual(srv_invalid_content4['values'][0]['port'], 1234)
self.assertEqual(srv_valid_content['values'][0]['target'],
'foo-2.unit.tests.')
self.assertEqual(srv_invalid_content['values'][0]['target'], '')
self.assertEqual(srv_invalid_content2['values'][0]['target'], '')
self.assertEqual(srv_invalid_content3['values'][0]['target'], '')
self.assertEqual(srv_invalid_content4['values'][0]['target'], '')
def test_apply(self):
provider = EasyDNSProvider('test', 'token', 'apikey')
resp = Mock()
resp.json = Mock()
provider._client._request = Mock(return_value=resp)
domain_after_creation = {
"tm": 1000000000,
"data": [{
"id": "12341001",
"domain": "unit.tests",
"host": "@",
"ttl": "0",
"prio": "0",
"type": "SOA",
"rdata": "dns1.easydns.com. zone.easydns.com. 2020010101"
" 3600 600 604800 0",
"geozone_id": "0",
"last_mod": "2020-01-01 01:01:01"
}, {
"id": "12341002",
"domain": "unit.tests",
"host": "@",
"ttl": "0",
"prio": "0",
"type": "NS",
"rdata": "LOCAL.",
"geozone_id": "0",
"last_mod": "2020-01-01 01:01:01"
}, {
"id": "12341003",
"domain": "unit.tests",
"host": "@",
"ttl": "0",
"prio": "0",
"type": "MX",
"rdata": "LOCAL.",
"geozone_id": "0",
"last_mod": "2020-01-01 01:01:01"
}],
"count": 3,
"total": 3,
"start": 0,
"max": 1000,
"status": 200
}
# non-existent domain, create everything
resp.json.side_effect = [
EasyDNSClientNotFound, # no zone in populate
domain_after_creation
]
plan = provider.plan(self.expected)
# No root NS, no ignored, no excluded, no unsupported
n = len(self.expected.records) - 9
self.assertEquals(n, len(plan.changes))
self.assertEquals(n, provider.apply(plan))
self.assertFalse(plan.exists)
self.assertEquals(25, provider._client._request.call_count)
provider._client._request.reset_mock()
# delete 1 and update 1
provider._client.records = Mock(return_value=[
{
"id": "12342001",
"domain": "unit.tests",
"host": "www",
"ttl": "300",
"prio": "0",
"type": "A",
"rdata": "2.2.3.9",
"geozone_id": "0",
"last_mod": "2020-01-01 01:01:01"
}, {
"id": "12342002",
"domain": "unit.tests",
"host": "www",
"ttl": "300",
"prio": "0",
"type": "A",
"rdata": "2.2.3.8",
"geozone_id": "0",
"last_mod": "2020-01-01 01:01:01"
}, {
"id": "12342003",
"domain": "unit.tests",
"host": "test1",
"ttl": "3600",
"prio": "0",
"type": "A",
"rdata": "1.2.3.4",
"geozone_id": "0",
"last_mod": "2020-01-01 01:01:01"
}
])
# Domain exists, we don't care about return
resp.json.side_effect = ['{}']
wanted = Zone('unit.tests.', [])
wanted.add_record(Record.new(wanted, 'test1', {
"name": "test1",
"ttl": 300,
"type": "A",
"value": "1.2.3.4",
}))
plan = provider.plan(wanted)
self.assertTrue(plan.exists)
self.assertEquals(2, len(plan.changes))
self.assertEquals(2, provider.apply(plan))
# recreate for update, and delete for the 2 parts of the other
provider._client._request.assert_has_calls([
call('PUT', '/zones/records/add/unit.tests/A', data={
'rdata': '1.2.3.4',
'name': 'test1',
'ttl': 300,
'type': 'A',
'host': 'test1',
}),
call('DELETE', '/zones/records/unit.tests/12342001'),
call('DELETE', '/zones/records/unit.tests/12342002'),
call('DELETE', '/zones/records/unit.tests/12342003')
], any_order=True)
def test_missing(self):
with self.assertRaises(ModuleNotFoundError):
from octodns.provider.easydns import EasyDnsProvider
EasyDnsProvider

Write Preview
Loading…
Cancel
Save