mirror of
https://github.com/github/octodns.git
synced 2024-05-11 05:55:00 +00:00
Extract SelectelProvider from octoDNS core
This commit is contained in:
Ross McFarland
@@ -27,6 +27,7 @@
|
||||
* [RackspaceProvider](https://github.com/octodns/octodns-rackspace/)
|
||||
* [Route53Provider](https://github.com/octodns/octodns-route53/) also
|
||||
AwsAcmMangingProcessor
|
||||
* [SelectelProvider](https://github.com/octodns/octodns-selectel/)
|
||||
* NS1 provider has received improvements to the dynamic record implementation.
|
||||
As a result, if octoDNS is downgraded from this version, any dynamic records
|
||||
created or updated using this version will show an update.
|
||||
|
||||
@@ -212,7 +212,8 @@ The table below lists the providers octoDNS supports. We're currently in the pro
|
||||
| [OvhProvider](https://github.com/octodns/octodns-ovh/) | [octodns_ovh](https://github.com/octodns/octodns-ovh/) | | | | |
|
||||
| [PowerDnsProvider](https://github.com/octodns/octodns-powerdns/) | [octodns_powerdns](https://github.com/octodns/octodns-powerdns/) | | | | |
|
||||
| [RackspaceProvider](https://github.com/octodns/octodns-rackspace/) | [octodns_rackspace](https://github.com/octodns/octodns-rackspace/) | | | | |
|
||||
| [Route53](https://github.com/octodns/octodns-route53) | [octodns_route53](https://github.com/octodns/octodns-route53) | | | | |
|
||||
| [Route53Provider](https://github.com/octodns/octodns-route53) | [octodns_route53](https://github.com/octodns/octodns-route53) | | | | |
|
||||
| [SelectelProvider](https://github.com/octodns/octodns-selectel/) | [octodns_selectel](https://github.com/octodns/octodns-selectel/) | | | | |
|
||||
| [Selectel](/octodns/provider/selectel.py) | | | A, AAAA, CNAME, MX, NS, SPF, SRV, TXT | No | |
|
||||
| [Transip](/octodns/provider/transip.py) | | transip | A, AAAA, CNAME, MX, NS, SRV, SPF, TXT, SSHFP, CAA | No | |
|
||||
| [UltraDns](/octodns/provider/ultra.py) | | | A, AAAA, CAA, CNAME, MX, NS, PTR, SPF, SRV, TXT | No | |
|
||||
|
||||
@@ -5,306 +5,17 @@
|
||||
from __future__ import absolute_import, division, print_function, \
|
||||
unicode_literals
|
||||
|
||||
from collections import defaultdict
|
||||
|
||||
from logging import getLogger
|
||||
|
||||
from requests import Session
|
||||
|
||||
from ..record import Record, Update
|
||||
from . import ProviderException
|
||||
from .base import BaseProvider
|
||||
|
||||
|
||||
def escape_semicolon(s):
|
||||
assert s
|
||||
return s.replace(';', '\\;')
|
||||
|
||||
|
||||
class SelectelAuthenticationRequired(ProviderException):
|
||||
def __init__(self, msg):
|
||||
message = 'Authorization failed. Invalid or empty token.'
|
||||
super(SelectelAuthenticationRequired, self).__init__(message)
|
||||
|
||||
|
||||
class SelectelProvider(BaseProvider):
|
||||
SUPPORTS_GEO = False
|
||||
|
||||
SUPPORTS = set(('A', 'AAAA', 'CNAME', 'MX', 'NS', 'TXT', 'SPF', 'SRV'))
|
||||
|
||||
MIN_TTL = 60
|
||||
|
||||
PAGINATION_LIMIT = 50
|
||||
|
||||
API_URL = 'https://api.selectel.ru/domains/v1'
|
||||
|
||||
def __init__(self, id, token, *args, **kwargs):
|
||||
self.log = getLogger(f'SelectelProvider[{id}]')
|
||||
self.log.debug('__init__: id=%s', id)
|
||||
super(SelectelProvider, self).__init__(id, *args, **kwargs)
|
||||
|
||||
self._sess = Session()
|
||||
self._sess.headers.update({
|
||||
'X-Token': token,
|
||||
'Content-Type': 'application/json',
|
||||
})
|
||||
self._zone_records = {}
|
||||
self._domain_list = self.domain_list()
|
||||
self._zones = None
|
||||
|
||||
def _request(self, method, path, params=None, data=None):
|
||||
self.log.debug('_request: method=%s, path=%s', method, path)
|
||||
|
||||
url = f'{self.API_URL}{path}'
|
||||
resp = self._sess.request(method, url, params=params, json=data)
|
||||
|
||||
self.log.debug('_request: status=%s', resp.status_code)
|
||||
if resp.status_code == 401:
|
||||
raise SelectelAuthenticationRequired(resp.text)
|
||||
elif resp.status_code == 404:
|
||||
return {}
|
||||
resp.raise_for_status()
|
||||
if method == 'DELETE':
|
||||
return {}
|
||||
return resp.json()
|
||||
|
||||
def _get_total_count(self, path):
|
||||
url = f'{self.API_URL}{path}'
|
||||
resp = self._sess.request('HEAD', url)
|
||||
return int(resp.headers['X-Total-Count'])
|
||||
|
||||
def _request_with_pagination(self, path, total_count):
|
||||
result = []
|
||||
for offset in range(0, total_count, self.PAGINATION_LIMIT):
|
||||
result += self._request('GET', path,
|
||||
params={'limit': self.PAGINATION_LIMIT,
|
||||
'offset': offset})
|
||||
return result
|
||||
|
||||
def _include_change(self, change):
|
||||
if isinstance(change, Update):
|
||||
existing = change.existing.data
|
||||
new = change.new.data
|
||||
new['ttl'] = max(self.MIN_TTL, new['ttl'])
|
||||
if new == existing:
|
||||
self.log.debug('_include_changes: new=%s, found existing=%s',
|
||||
new, existing)
|
||||
return False
|
||||
return True
|
||||
|
||||
def _apply(self, plan):
|
||||
desired = plan.desired
|
||||
changes = plan.changes
|
||||
self.log.debug('_apply: zone=%s, len(changes)=%d', desired.name,
|
||||
len(changes))
|
||||
|
||||
zone_name = desired.name[:-1]
|
||||
for change in changes:
|
||||
class_name = change.__class__.__name__
|
||||
getattr(self, f'_apply_{class_name}'.lower())(zone_name, change)
|
||||
|
||||
def _apply_create(self, zone_name, change):
|
||||
new = change.new
|
||||
params_for = getattr(self, f'_params_for_{new._type}')
|
||||
for params in params_for(new):
|
||||
self.create_record(zone_name, params)
|
||||
|
||||
def _apply_update(self, zone_name, change):
|
||||
self._apply_delete(zone_name, change)
|
||||
self._apply_create(zone_name, change)
|
||||
|
||||
def _apply_delete(self, zone_name, change):
|
||||
existing = change.existing
|
||||
self.delete_record(zone_name, existing._type, existing.name)
|
||||
|
||||
def _params_for_multiple(self, record):
|
||||
for value in record.values:
|
||||
yield {
|
||||
'content': value,
|
||||
'name': record.fqdn,
|
||||
'ttl': max(self.MIN_TTL, record.ttl),
|
||||
'type': record._type,
|
||||
}
|
||||
|
||||
def _params_for_single(self, record):
|
||||
yield {
|
||||
'content': record.value,
|
||||
'name': record.fqdn,
|
||||
'ttl': max(self.MIN_TTL, record.ttl),
|
||||
'type': record._type
|
||||
}
|
||||
|
||||
def _params_for_MX(self, record):
|
||||
for value in record.values:
|
||||
yield {
|
||||
'content': value.exchange,
|
||||
'name': record.fqdn,
|
||||
'ttl': max(self.MIN_TTL, record.ttl),
|
||||
'type': record._type,
|
||||
'priority': value.preference
|
||||
}
|
||||
|
||||
def _params_for_SRV(self, record):
|
||||
for value in record.values:
|
||||
yield {
|
||||
'name': record.fqdn,
|
||||
'target': value.target,
|
||||
'ttl': max(self.MIN_TTL, record.ttl),
|
||||
'type': record._type,
|
||||
'port': value.port,
|
||||
'weight': value.weight,
|
||||
'priority': value.priority
|
||||
}
|
||||
|
||||
_params_for_A = _params_for_multiple
|
||||
_params_for_AAAA = _params_for_multiple
|
||||
_params_for_NS = _params_for_multiple
|
||||
_params_for_TXT = _params_for_multiple
|
||||
_params_for_SPF = _params_for_multiple
|
||||
|
||||
_params_for_CNAME = _params_for_single
|
||||
|
||||
def _data_for_A(self, _type, records):
|
||||
return {
|
||||
'ttl': records[0]['ttl'],
|
||||
'type': _type,
|
||||
'values': [r['content'] for r in records],
|
||||
}
|
||||
|
||||
_data_for_AAAA = _data_for_A
|
||||
|
||||
def _data_for_NS(self, _type, records):
|
||||
return {
|
||||
'ttl': records[0]['ttl'],
|
||||
'type': _type,
|
||||
'values': [f'{r["content"]}.' for r in records],
|
||||
}
|
||||
|
||||
def _data_for_MX(self, _type, records):
|
||||
values = []
|
||||
for record in records:
|
||||
values.append({
|
||||
'preference': record['priority'],
|
||||
'exchange': f'{record["content"]}.',
|
||||
})
|
||||
return {
|
||||
'ttl': records[0]['ttl'],
|
||||
'type': _type,
|
||||
'values': values,
|
||||
}
|
||||
|
||||
def _data_for_CNAME(self, _type, records):
|
||||
only = records[0]
|
||||
return {
|
||||
'ttl': only['ttl'],
|
||||
'type': _type,
|
||||
'value': f'{only["content"]}.',
|
||||
}
|
||||
|
||||
def _data_for_TXT(self, _type, records):
|
||||
return {
|
||||
'ttl': records[0]['ttl'],
|
||||
'type': _type,
|
||||
'values': [escape_semicolon(r['content']) for r in records]
|
||||
}
|
||||
|
||||
def _data_for_SRV(self, _type, records):
|
||||
values = []
|
||||
for record in records:
|
||||
values.append({
|
||||
'priority': record['priority'],
|
||||
'weight': record['weight'],
|
||||
'port': record['port'],
|
||||
'target': f'{record["target"]}.',
|
||||
})
|
||||
|
||||
return {
|
||||
'type': _type,
|
||||
'ttl': records[0]['ttl'],
|
||||
'values': values,
|
||||
}
|
||||
|
||||
def populate(self, zone, target=False, lenient=False):
|
||||
self.log.debug('populate: name=%s, target=%s, lenient=%s',
|
||||
zone.name, target, lenient)
|
||||
before = len(zone.records)
|
||||
records = self.zone_records(zone)
|
||||
if records:
|
||||
values = defaultdict(lambda: defaultdict(list))
|
||||
for record in records:
|
||||
name = zone.hostname_from_fqdn(record['name'])
|
||||
_type = record['type']
|
||||
if _type in self.SUPPORTS:
|
||||
values[name][record['type']].append(record)
|
||||
for name, types in values.items():
|
||||
for _type, records in types.items():
|
||||
data_for = getattr(self, f'_data_for_{_type}')
|
||||
data = data_for(_type, records)
|
||||
record = Record.new(zone, name, data, source=self,
|
||||
lenient=lenient)
|
||||
zone.add_record(record)
|
||||
self.log.info('populate: found %s records',
|
||||
len(zone.records) - before)
|
||||
|
||||
def domain_list(self):
|
||||
path = '/'
|
||||
domains = {}
|
||||
domains_list = []
|
||||
|
||||
total_count = self._get_total_count(path)
|
||||
domains_list = self._request_with_pagination(path, total_count)
|
||||
|
||||
for domain in domains_list:
|
||||
domains[domain['name']] = domain
|
||||
return domains
|
||||
|
||||
def zone_records(self, zone):
|
||||
path = f'/{zone.name[:-1]}/records/'
|
||||
zone_records = []
|
||||
|
||||
total_count = self._get_total_count(path)
|
||||
zone_records = self._request_with_pagination(path, total_count)
|
||||
|
||||
self._zone_records[zone.name] = zone_records
|
||||
return self._zone_records[zone.name]
|
||||
|
||||
def create_domain(self, name, zone=""):
|
||||
path = '/'
|
||||
|
||||
data = {
|
||||
'name': name,
|
||||
'bind_zone': zone,
|
||||
}
|
||||
|
||||
resp = self._request('POST', path, data=data)
|
||||
self._domain_list[name] = resp
|
||||
return resp
|
||||
|
||||
def create_record(self, zone_name, data):
|
||||
self.log.debug('Create record. Zone: %s, data %s', zone_name, data)
|
||||
if zone_name in self._domain_list.keys():
|
||||
domain_id = self._domain_list[zone_name]['id']
|
||||
else:
|
||||
domain_id = self.create_domain(zone_name)['id']
|
||||
|
||||
path = f'/{domain_id}/records/'
|
||||
return self._request('POST', path, data=data)
|
||||
|
||||
def delete_record(self, domain, _type, zone):
|
||||
self.log.debug('Delete record. Domain: %s, Type: %s', domain, _type)
|
||||
|
||||
domain_id = self._domain_list[domain]['id']
|
||||
records = self._zone_records.get(f'{domain}.', False)
|
||||
if not records:
|
||||
path = f'/{domain_id}/records/'
|
||||
records = self._request('GET', path)
|
||||
|
||||
for record in records:
|
||||
full_domain = domain
|
||||
if zone:
|
||||
full_domain = f'{zone}{domain}'
|
||||
if record['type'] == _type and record['name'] == full_domain:
|
||||
path = f'/{domain_id}/records/{record["id"]}'
|
||||
return self._request('DELETE', path)
|
||||
|
||||
self.log.debug('Delete record failed (Record not found)')
|
||||
logger = getLogger('Selectel')
|
||||
try:
|
||||
logger.warning('octodns_selectel shimmed. Update your provider class to '
|
||||
'octodns_selectel.SelectelProvider. '
|
||||
'Shim will be removed in 1.0')
|
||||
from octodns_selectel import SelectelProvider
|
||||
SelectelProvider # pragma: no cover
|
||||
except ModuleNotFoundError:
|
||||
logger.exception('SelectelProvider has been moved into a seperate module, '
|
||||
'octodns_selectel is now required. Provider class should '
|
||||
'be updated to octodns_selectel.SelectelProvider')
|
||||
raise
|
||||
|
||||
@@ -7,387 +7,10 @@ from __future__ import absolute_import, division, print_function, \
|
||||
|
||||
from unittest import TestCase
|
||||
|
||||
import requests_mock
|
||||
|
||||
from octodns.provider.selectel import SelectelProvider
|
||||
from octodns.record import Record, Update
|
||||
from octodns.zone import Zone
|
||||
class TestSelectelShim(TestCase):
|
||||
|
||||
|
||||
class TestSelectelProvider(TestCase):
|
||||
API_URL = 'https://api.selectel.ru/domains/v1'
|
||||
|
||||
api_record = []
|
||||
|
||||
zone = Zone('unit.tests.', [])
|
||||
expected = set()
|
||||
|
||||
domain = [{"name": "unit.tests", "id": 100000}]
|
||||
|
||||
# A, subdomain=''
|
||||
api_record.append({
|
||||
'type': 'A',
|
||||
'ttl': 100,
|
||||
'content': '1.2.3.4',
|
||||
'name': 'unit.tests',
|
||||
'id': 1
|
||||
})
|
||||
expected.add(Record.new(zone, '', {
|
||||
'ttl': 100,
|
||||
'type': 'A',
|
||||
'value': '1.2.3.4',
|
||||
}))
|
||||
|
||||
# A, subdomain='sub'
|
||||
api_record.append({
|
||||
'type': 'A',
|
||||
'ttl': 200,
|
||||
'content': '1.2.3.4',
|
||||
'name': 'sub.unit.tests',
|
||||
'id': 2
|
||||
})
|
||||
expected.add(Record.new(zone, 'sub', {
|
||||
'ttl': 200,
|
||||
'type': 'A',
|
||||
'value': '1.2.3.4',
|
||||
}))
|
||||
|
||||
# CNAME
|
||||
api_record.append({
|
||||
'type': 'CNAME',
|
||||
'ttl': 300,
|
||||
'content': 'unit.tests',
|
||||
'name': 'www2.unit.tests',
|
||||
'id': 3
|
||||
})
|
||||
expected.add(Record.new(zone, 'www2', {
|
||||
'ttl': 300,
|
||||
'type': 'CNAME',
|
||||
'value': 'unit.tests.',
|
||||
}))
|
||||
|
||||
# MX
|
||||
api_record.append({
|
||||
'type': 'MX',
|
||||
'ttl': 400,
|
||||
'content': 'mx1.unit.tests',
|
||||
'priority': 10,
|
||||
'name': 'unit.tests',
|
||||
'id': 4
|
||||
})
|
||||
expected.add(Record.new(zone, '', {
|
||||
'ttl': 400,
|
||||
'type': 'MX',
|
||||
'values': [{
|
||||
'preference': 10,
|
||||
'exchange': 'mx1.unit.tests.',
|
||||
}]
|
||||
}))
|
||||
|
||||
# NS
|
||||
api_record.append({
|
||||
'type': 'NS',
|
||||
'ttl': 600,
|
||||
'content': 'ns1.unit.tests',
|
||||
'name': 'unit.tests.',
|
||||
'id': 6
|
||||
})
|
||||
api_record.append({
|
||||
'type': 'NS',
|
||||
'ttl': 600,
|
||||
'content': 'ns2.unit.tests',
|
||||
'name': 'unit.tests',
|
||||
'id': 7
|
||||
})
|
||||
expected.add(Record.new(zone, '', {
|
||||
'ttl': 600,
|
||||
'type': 'NS',
|
||||
'values': ['ns1.unit.tests.', 'ns2.unit.tests.'],
|
||||
}))
|
||||
|
||||
# NS with sub
|
||||
api_record.append({
|
||||
'type': 'NS',
|
||||
'ttl': 700,
|
||||
'content': 'ns3.unit.tests',
|
||||
'name': 'www3.unit.tests',
|
||||
'id': 8
|
||||
})
|
||||
api_record.append({
|
||||
'type': 'NS',
|
||||
'ttl': 700,
|
||||
'content': 'ns4.unit.tests',
|
||||
'name': 'www3.unit.tests',
|
||||
'id': 9
|
||||
})
|
||||
expected.add(Record.new(zone, 'www3', {
|
||||
'ttl': 700,
|
||||
'type': 'NS',
|
||||
'values': ['ns3.unit.tests.', 'ns4.unit.tests.'],
|
||||
}))
|
||||
|
||||
# SRV
|
||||
api_record.append({
|
||||
'type': 'SRV',
|
||||
'ttl': 800,
|
||||
'target': 'foo-1.unit.tests',
|
||||
'weight': 20,
|
||||
'priority': 10,
|
||||
'port': 30,
|
||||
'id': 10,
|
||||
'name': '_srv._tcp.unit.tests'
|
||||
})
|
||||
api_record.append({
|
||||
'type': 'SRV',
|
||||
'ttl': 800,
|
||||
'target': 'foo-2.unit.tests',
|
||||
'name': '_srv._tcp.unit.tests',
|
||||
'weight': 50,
|
||||
'priority': 40,
|
||||
'port': 60,
|
||||
'id': 11
|
||||
})
|
||||
expected.add(Record.new(zone, '_srv._tcp', {
|
||||
'ttl': 800,
|
||||
'type': 'SRV',
|
||||
'values': [{
|
||||
'priority': 10,
|
||||
'weight': 20,
|
||||
'port': 30,
|
||||
'target': 'foo-1.unit.tests.',
|
||||
}, {
|
||||
'priority': 40,
|
||||
'weight': 50,
|
||||
'port': 60,
|
||||
'target': 'foo-2.unit.tests.',
|
||||
}]
|
||||
}))
|
||||
|
||||
# AAAA
|
||||
aaaa_record = {
|
||||
'type': 'AAAA',
|
||||
'ttl': 200,
|
||||
'content': '1:1ec:1::1',
|
||||
'name': 'unit.tests',
|
||||
'id': 15
|
||||
}
|
||||
api_record.append(aaaa_record)
|
||||
expected.add(Record.new(zone, '', {
|
||||
'ttl': 200,
|
||||
'type': 'AAAA',
|
||||
'value': '1:1ec:1::1',
|
||||
}))
|
||||
|
||||
# TXT
|
||||
api_record.append({
|
||||
'type': 'TXT',
|
||||
'ttl': 300,
|
||||
'content': 'little text',
|
||||
'name': 'text.unit.tests',
|
||||
'id': 16
|
||||
})
|
||||
expected.add(Record.new(zone, 'text', {
|
||||
'ttl': 200,
|
||||
'type': 'TXT',
|
||||
'value': 'little text',
|
||||
}))
|
||||
|
||||
@requests_mock.Mocker()
|
||||
def test_populate(self, fake_http):
|
||||
zone = Zone('unit.tests.', [])
|
||||
fake_http.get(f'{self.API_URL}/unit.tests/records/',
|
||||
json=self.api_record)
|
||||
fake_http.get(f'{self.API_URL}/', json=self.domain)
|
||||
fake_http.head(f'{self.API_URL}/unit.tests/records/',
|
||||
headers={'X-Total-Count': str(len(self.api_record))})
|
||||
fake_http.head(f'{self.API_URL}/',
|
||||
headers={'X-Total-Count': str(len(self.domain))})
|
||||
|
||||
provider = SelectelProvider(123, 'secret_token')
|
||||
provider.populate(zone)
|
||||
|
||||
self.assertEqual(self.expected, zone.records)
|
||||
|
||||
@requests_mock.Mocker()
|
||||
def test_populate_invalid_record(self, fake_http):
|
||||
more_record = self.api_record
|
||||
more_record.append({"name": "unit.tests",
|
||||
"id": 100001,
|
||||
"content": "support.unit.tests.",
|
||||
"ttl": 300, "ns": "ns1.unit.tests",
|
||||
"type": "SOA",
|
||||
"email": "support@unit.tests"})
|
||||
|
||||
zone = Zone('unit.tests.', [])
|
||||
fake_http.get(f'{self.API_URL}/unit.tests/records/',
|
||||
json=more_record)
|
||||
fake_http.get(f'{self.API_URL}/', json=self.domain)
|
||||
fake_http.head(f'{self.API_URL}/unit.tests/records/',
|
||||
headers={'X-Total-Count': str(len(self.api_record))})
|
||||
fake_http.head(f'{self.API_URL}/',
|
||||
headers={'X-Total-Count': str(len(self.domain))})
|
||||
|
||||
zone.add_record(Record.new(self.zone, 'unsup', {
|
||||
'ttl': 200,
|
||||
'type': 'NAPTR',
|
||||
'value': {
|
||||
'order': 40,
|
||||
'preference': 70,
|
||||
'flags': 'U',
|
||||
'service': 'SIP+D2U',
|
||||
'regexp': '!^.*$!sip:info@bar.example.com!',
|
||||
'replacement': '.',
|
||||
}
|
||||
}))
|
||||
|
||||
provider = SelectelProvider(123, 'secret_token')
|
||||
provider.populate(zone)
|
||||
|
||||
self.assertNotEqual(self.expected, zone.records)
|
||||
|
||||
@requests_mock.Mocker()
|
||||
def test_apply(self, fake_http):
|
||||
|
||||
fake_http.get(f'{self.API_URL}/unit.tests/records/', json=list())
|
||||
fake_http.get(f'{self.API_URL}/', json=self.domain)
|
||||
fake_http.head(f'{self.API_URL}/unit.tests/records/',
|
||||
headers={'X-Total-Count': '0'})
|
||||
fake_http.head(f'{self.API_URL}/',
|
||||
headers={'X-Total-Count': str(len(self.domain))})
|
||||
fake_http.post(f'{self.API_URL}/100000/records/', json=list())
|
||||
|
||||
provider = SelectelProvider(123, 'test_token')
|
||||
|
||||
zone = Zone('unit.tests.', [])
|
||||
|
||||
for record in self.expected:
|
||||
zone.add_record(record)
|
||||
|
||||
plan = provider.plan(zone)
|
||||
self.assertEqual(8, len(plan.changes))
|
||||
self.assertEqual(8, provider.apply(plan))
|
||||
|
||||
@requests_mock.Mocker()
|
||||
def test_domain_list(self, fake_http):
|
||||
fake_http.get(f'{self.API_URL}/', json=self.domain)
|
||||
fake_http.head(f'{self.API_URL}/',
|
||||
headers={'X-Total-Count': str(len(self.domain))})
|
||||
|
||||
expected = {'unit.tests': self.domain[0]}
|
||||
provider = SelectelProvider(123, 'test_token')
|
||||
|
||||
result = provider.domain_list()
|
||||
self.assertEqual(result, expected)
|
||||
|
||||
@requests_mock.Mocker()
|
||||
def test_authentication_fail(self, fake_http):
|
||||
fake_http.get(f'{self.API_URL}/', status_code=401)
|
||||
fake_http.head(f'{self.API_URL}/',
|
||||
headers={'X-Total-Count': str(len(self.domain))})
|
||||
|
||||
with self.assertRaises(Exception) as ctx:
|
||||
SelectelProvider(123, 'fail_token')
|
||||
self.assertEqual(str(ctx.exception),
|
||||
'Authorization failed. Invalid or empty token.')
|
||||
|
||||
@requests_mock.Mocker()
|
||||
def test_not_exist_domain(self, fake_http):
|
||||
fake_http.get(f'{self.API_URL}/', status_code=404, json='')
|
||||
fake_http.head(f'{self.API_URL}/',
|
||||
headers={'X-Total-Count': str(len(self.domain))})
|
||||
|
||||
fake_http.post(f'{self.API_URL}/', json={"name": "unit.tests",
|
||||
"create_date": 1507154178,
|
||||
"id": 100000})
|
||||
fake_http.get(f'{self.API_URL}/unit.tests/records/', json=list())
|
||||
fake_http.head(f'{self.API_URL}/unit.tests/records/',
|
||||
headers={'X-Total-Count': str(len(self.api_record))})
|
||||
fake_http.post(f'{self.API_URL}/100000/records/', json=list())
|
||||
|
||||
provider = SelectelProvider(123, 'test_token')
|
||||
|
||||
zone = Zone('unit.tests.', [])
|
||||
|
||||
for record in self.expected:
|
||||
zone.add_record(record)
|
||||
|
||||
plan = provider.plan(zone)
|
||||
self.assertEqual(8, len(plan.changes))
|
||||
self.assertEqual(8, provider.apply(plan))
|
||||
|
||||
@requests_mock.Mocker()
|
||||
def test_delete_no_exist_record(self, fake_http):
|
||||
fake_http.get(f'{self.API_URL}/', json=self.domain)
|
||||
fake_http.get(f'{self.API_URL}/100000/records/', json=list())
|
||||
fake_http.head(f'{self.API_URL}/',
|
||||
headers={'X-Total-Count': str(len(self.domain))})
|
||||
fake_http.head(f'{self.API_URL}/unit.tests/records/',
|
||||
headers={'X-Total-Count': '0'})
|
||||
|
||||
provider = SelectelProvider(123, 'test_token')
|
||||
|
||||
zone = Zone('unit.tests.', [])
|
||||
|
||||
provider.delete_record('unit.tests', 'NS', zone)
|
||||
|
||||
@requests_mock.Mocker()
|
||||
def test_change_record(self, fake_http):
|
||||
exist_record = [self.aaaa_record,
|
||||
{"content": "6.6.5.7",
|
||||
"ttl": 100,
|
||||
"type": "A",
|
||||
"id": 100001,
|
||||
"name": "delete.unit.tests"},
|
||||
{"content": "9.8.2.1",
|
||||
"ttl": 100,
|
||||
"type": "A",
|
||||
"id": 100002,
|
||||
"name": "unit.tests"}] # exist
|
||||
fake_http.get(f'{self.API_URL}/unit.tests/records/', json=exist_record)
|
||||
fake_http.get(f'{self.API_URL}/', json=self.domain)
|
||||
fake_http.get(f'{self.API_URL}/100000/records/', json=exist_record)
|
||||
fake_http.head(f'{self.API_URL}/unit.tests/records/',
|
||||
headers={'X-Total-Count': str(len(exist_record))})
|
||||
fake_http.head(f'{self.API_URL}/',
|
||||
headers={'X-Total-Count': str(len(self.domain))})
|
||||
fake_http.head(f'{self.API_URL}/100000/records/',
|
||||
headers={'X-Total-Count': str(len(exist_record))})
|
||||
fake_http.post(f'{self.API_URL}/100000/records/',
|
||||
json=list())
|
||||
fake_http.delete(f'{self.API_URL}/100000/records/100001', text="")
|
||||
fake_http.delete(f'{self.API_URL}/100000/records/100002', text="")
|
||||
|
||||
provider = SelectelProvider(123, 'test_token')
|
||||
|
||||
zone = Zone('unit.tests.', [])
|
||||
|
||||
for record in self.expected:
|
||||
zone.add_record(record)
|
||||
|
||||
plan = provider.plan(zone)
|
||||
self.assertEqual(8, len(plan.changes))
|
||||
self.assertEqual(8, provider.apply(plan))
|
||||
|
||||
@requests_mock.Mocker()
|
||||
def test_include_change_returns_false(self, fake_http):
|
||||
fake_http.get(f'{self.API_URL}/', json=self.domain)
|
||||
fake_http.head(f'{self.API_URL}/',
|
||||
headers={'X-Total-Count': str(len(self.domain))})
|
||||
provider = SelectelProvider(123, 'test_token')
|
||||
zone = Zone('unit.tests.', [])
|
||||
|
||||
exist_record = Record.new(zone, '', {
|
||||
'ttl': 60,
|
||||
'type': 'A',
|
||||
'values': ['1.1.1.1', '2.2.2.2']
|
||||
})
|
||||
new = Record.new(zone, '', {
|
||||
'ttl': 10,
|
||||
'type': 'A',
|
||||
'values': ['1.1.1.1', '2.2.2.2']
|
||||
})
|
||||
change = Update(exist_record, new)
|
||||
|
||||
include_change = provider._include_change(change)
|
||||
|
||||
self.assertFalse(include_change)
|
||||
def test_missing(self):
|
||||
with self.assertRaises(ModuleNotFoundError):
|
||||
from octodns.provider.selectel import SelectelProvider
|
||||
SelectelProvider
|
||||
|
||||
Reference in New Issue
Block a user