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

add selectel as octodns provider

pull/116/head
Nikita Belecky 8 years ago
parent
8a13ccab46
commit
18dcfe6d8e
3 changed files with 624 additions and 0 deletions
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. +1
    -0
      README.md
  2. +278
    -0
      octodns/provider/selectel.py
  3. +345
    -0
      tests/test_octodns_provider_selectel.py

+ 1
- 0
README.md View File

@ -155,6 +155,7 @@ The above command pulled the existing data out of Route53 and placed the results
| [DynProvider](/octodns/provider/dyn.py) | All | Yes | |
| [Ns1Provider](/octodns/provider/ns1.py) | All | No | |
| [OVH](/octodns/provider/ovh.py) | A, AAAA, CNAME, MX, NAPTR, NS, PTR, SPF, SRV, SSHFP, TXT | No | |
| [Selectel](/octodns/provider/selectel.py) | A, AAAA, CNAME, MX, NS, SRV, TXT | No | |
| [PowerDnsProvider](/octodns/provider/powerdns.py) | All | No | |
| [Route53](/octodns/provider/route53.py) | A, AAAA, CAA, CNAME, MX, NAPTR, NS, PTR, SPF, SRV, TXT | Yes | |
| [TinyDNSSource](/octodns/source/tinydns.py) | A, CNAME, MX, NS, PTR | No | read-only |


+ 278
- 0
octodns/provider/selectel.py View File

@ -0,0 +1,278 @@
#
#
#
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
from .base import BaseProvider
class SelectelAuthenticationRequired(Exception):
def __init__(self, msg):
Exception.__init__(self,
'Authorization failed. Invalid or empty token.')
class SelectelProvider(BaseProvider):
SUPPORTS_GEO = False
SUPPORTS = set(('A', 'AAAA', 'CNAME', 'MX', 'NS', 'TXT', 'SRV'))
MIN_TTL = 60
MAX_TTL = 604800
API_URL = 'https://api.selectel.ru/domains/v1'
def __init__(self, id, token, *args, **kwargs):
self.log = getLogger('SelectelProvider[{}]'.format(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()
def _request(self, method, path, params=None, data=None):
self.log.debug('_request: method=%s, path=%s', method, path)
url = '{}{}'.format(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 {}
if resp.json():
return resp.json()
self.log.debug('_request: empty response')
return {}
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, '_apply_{}'.format(class_name).lower())(zone_name,
change)
def _apply_create(self, zone_name, change):
new = change.new
params_for = getattr(self, '_params_for_{}'.format(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_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': ['{}.'.format(r['content']) for r in records],
}
def _data_for_MX(self, _type, records):
values = []
for record in records:
values.append({
'preference': record['priority'],
'exchange': '{}.'.format(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': '{}.'.format(only['content'])
}
def _data_for_TXT(self, _type, records):
return {
'ttl': records[0]['ttl'],
'type': _type,
'values': [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': '{}.'.format(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, '_data_for_{}'.format(_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 = '/'
resp = self._request('GET', path)
domain_dict = {}
for domain in resp:
domain_dict[domain['name']] = domain
return domain_dict
def zone_records(self, zone):
path = '/{}/records/'.format(zone.name[:-1])
resp = self._request('GET', path)
self._zone_records[zone.name] = resp
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 = '/{}/records/'.format(domain_id)
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']
path = '/{}/records/'.format(domain_id)
records = self._request('GET', path)
for record in records:
full_domain = domain
if zone:
full_domain = '{}.{}'.format(zone, domain)
if record['type'] == _type and record['name'] == full_domain:
path = '/{}/records/{}'.format(domain_id, record['id'])
return self._request('DELETE', path)
self.log.debug('Delete record failed (Record not found)')

+ 345
- 0
tests/test_octodns_provider_selectel.py View File

@ -0,0 +1,345 @@
#
#
#
from __future__ import absolute_import, division, print_function, \
unicode_literals
from unittest import TestCase
import requests_mock
from octodns.provider.selectel import SelectelProvider
from octodns.record import Record
from octodns.zone import Zone
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('{}/unit.tests/records/'.format(self.API_URL),
json=self.api_record)
fake_http.get('{}/'.format(self.API_URL), json=self.domain)
provider = SelectelProvider(3123, 'secret_token')
provider.populate(zone)
self.assertEquals(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('{}/unit.tests/records/'.format(self.API_URL),
json=more_record)
fake_http.get('{}/'.format(self.API_URL), json=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(3123, 'secret_token')
provider.populate(zone)
self.assertNotEqual(self.expected, zone.records)
@requests_mock.Mocker()
def test_apply(self, fake_http):
fake_http.get('{}/unit.tests/records/'.format(self.API_URL),
json=list())
fake_http.get('{}/'.format(self.API_URL), json=self.domain)
fake_http.post('{}/100000/records/'.format(self.API_URL), 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.assertEquals(8, len(plan.changes))
self.assertEquals(8, provider.apply(plan))
@requests_mock.Mocker()
def test_domain_list(self, fake_http):
fake_http.get('{}/'.format(self.API_URL), json=self.domain)
expected = {'unit.tests': self.domain[0]}
provider = SelectelProvider(123, 'test_token')
result = provider.domain_list()
self.assertEquals(result, expected)
@requests_mock.Mocker()
def test_authentication_fail(self, fake_http):
fake_http.get('{}/'.format(self.API_URL), status_code=401)
with self.assertRaises(Exception) as ctx:
SelectelProvider(123, 'fail_token')
self.assertEquals(ctx.exception.message,
'Authorization failed. Invalid or empty token.')
@requests_mock.Mocker()
def test_not_exist_domain(self, fake_http):
fake_http.get('{}/'.format(self.API_URL), status_code=404, json='')
fake_http.post('{}/'.format(self.API_URL),
json={"name": "unit.tests",
"create_date": 1507154178,
"id": 100000})
fake_http.get('{}/unit.tests/records/'.format(self.API_URL),
json=list())
fake_http.post('{}/100000/records/'.format(self.API_URL),
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.assertEquals(8, len(plan.changes))
self.assertEquals(8, provider.apply(plan))
@requests_mock.Mocker()
def test_delete_no_exist_record(self, fake_http):
fake_http.get('{}/'.format(self.API_URL), json=self.domain)
fake_http.get('{}/100000/records/'.format(self.API_URL), json=list())
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('{}/unit.tests/records/'.format(self.API_URL),
json=exist_record)
fake_http.get('{}/'.format(self.API_URL), json=self.domain)
fake_http.get('{}/100000/records/'.format(self.API_URL),
json=exist_record)
fake_http.post('{}/100000/records/'.format(self.API_URL),
json=list())
fake_http.delete('{}/100000/records/100001'.format(self.API_URL),
text="")
fake_http.delete('{}/100000/records/100002'.format(self.API_URL),
text="")
provider = SelectelProvider(123, 'test_token')
zone = Zone('unit.tests.', [])
for record in self.expected:
zone.add_record(record)
plan = provider.plan(zone)
self.assertEquals(8, len(plan.changes))
self.assertEquals(8, provider.apply(plan))

Write Preview
Loading…
Cancel
Save