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 DigitalOceanProvider

pull/836/head
Ross McFarland 4 years ago
parent
237610d3a9
commit
e2cbd955f3
No known key found for this signature in database GPG Key ID: 943B179E15D3B22A
6 changed files with 22 additions and 905 deletions
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. +1
    -0
      CHANGELOG.md
  2. +1
    -1
      README.md
  3. +15
    -342
      octodns/provider/digitalocean.py
  4. +0
    -188
      tests/fixtures/digitalocean-page-1.json
  5. +0
    -111
      tests/fixtures/digitalocean-page-2.json
  6. +5
    -263
      tests/test_octodns_provider_digitalocean.py

+ 1
- 0
CHANGELOG.md View File

@ -6,6 +6,7 @@
https://github.com/octodns/octodns/issues/622 &
https://github.com/octodns/octodns/pull/822 for more information. Providers
that have been extracted in this release include:
* [DigitalOceanProvider](https://github.com/octodns/octodns-digitalocean/)
* [DnsimpleProvider](https://github.com/octodns/octodns-dnsimple/)
* [Ns1Provider](https://github.com/octodns/octodns-ns1/)
* [PowerDnsProvider](https://github.com/octodns/octodns-powerdns/)


+ 1
- 1
README.md View File

@ -196,7 +196,7 @@ The table below lists the providers octoDNS supports. We're currently in the pro
| [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, LOC, MX, NS, PTR, SPF, SRV, TXT, URLFWD | No | CAA tags restricted |
| [ConstellixProvider](/octodns/provider/constellix.py) | | | A, AAAA, ALIAS (ANAME), CAA, CNAME, MX, NS, PTR, SPF, SRV, TXT | Yes | CAA tags restricted |
| [DigitalOceanProvider](/octodns/provider/digitalocean.py) | | | A, AAAA, CAA, CNAME, MX, NS, TXT, SRV | No | CAA tags restricted |
| [DigitalOceanProvider](https://github.com/octodns/octodns-digitalocean/) | [octodns_digitalocean](https://github.com/octodns/octodns-digitalocean/) | | | | |
| [DnsMadeEasyProvider](/octodns/provider/dnsmadeeasy.py) | | | A, AAAA, ALIAS (ANAME), CAA, CNAME, MX, NS, PTR, SPF, SRV, TXT | No | CAA tags restricted |
| [DnsimpleProvider](https://github.com/octodns/octodns-dnsimple/) | [octodns_dnsimple](https://github.com/octodns/octodns-dnsimple/) | | | | |
| [DynProvider](/octodns/provider/dyn.py) | | dyn | All | Both | |


+ 15
- 342
octodns/provider/digitalocean.py View File

@ -5,345 +5,18 @@
from __future__ import absolute_import, division, print_function, \
unicode_literals
from collections import defaultdict
from requests import Session
import logging
from ..record import Record
from . import ProviderException
from .base import BaseProvider
class DigitalOceanClientException(ProviderException):
pass
class DigitalOceanClientNotFound(DigitalOceanClientException):
def __init__(self):
super(DigitalOceanClientNotFound, self).__init__('Not Found')
class DigitalOceanClientUnauthorized(DigitalOceanClientException):
def __init__(self):
super(DigitalOceanClientUnauthorized, self).__init__('Unauthorized')
class DigitalOceanClient(object):
BASE = 'https://api.digitalocean.com/v2'
def __init__(self, token):
sess = Session()
sess.headers.update({'Authorization': f'Bearer {token}'})
self._sess = sess
def _request(self, method, path, params=None, data=None):
url = f'{self.BASE}{path}'
resp = self._sess.request(method, url, params=params, json=data)
if resp.status_code == 401:
raise DigitalOceanClientUnauthorized()
if resp.status_code == 404:
raise DigitalOceanClientNotFound()
resp.raise_for_status()
return resp
def domain(self, name):
path = f'/domains/{name}'
return self._request('GET', path).json()
def domain_create(self, name):
# Digitalocean requires an IP on zone creation
self._request('POST', '/domains', data={'name': name,
'ip_address': '192.0.2.1'})
# After the zone is created, immediately delete the record
records = self.records(name)
for record in records:
if record['name'] == '' and record['type'] == 'A':
self.record_delete(name, record['id'])
def records(self, zone_name):
path = f'/domains/{zone_name}/records'
ret = []
page = 1
while True:
data = self._request('GET', path, {'page': page}).json()
ret += data['domain_records']
links = data['links']
# https://developers.digitalocean.com/documentation/v2/#links
# pages exists if there is more than 1 page
# last doesn't exist if you're on the last page
try:
links['pages']['last']
page += 1
except KeyError:
break
for record in ret:
# change any apex record to empty string
if record['name'] == '@':
record['name'] = ''
# change any apex value to zone name
if record['data'] == '@':
record['data'] = zone_name
return ret
def record_create(self, zone_name, params):
path = f'/domains/{zone_name}/records'
# change empty name string to @, DO uses @ for apex record names
if params['name'] == '':
params['name'] = '@'
self._request('POST', path, data=params)
def record_delete(self, zone_name, record_id):
path = f'/domains/{zone_name}/records/{record_id}'
self._request('DELETE', path)
class DigitalOceanProvider(BaseProvider):
'''
DigitalOcean DNS provider using API v2
digitalocean:
class: octodns.provider.digitalocean.DigitalOceanProvider
# Your DigitalOcean API token (required)
token: foo
'''
SUPPORTS_GEO = False
SUPPORTS_DYNAMIC = False
SUPPORTS = set(('A', 'AAAA', 'CAA', 'CNAME', 'MX', 'NS', 'TXT', 'SRV'))
def __init__(self, id, token, *args, **kwargs):
self.log = logging.getLogger(f'DigitalOceanProvider[{id}]')
self.log.debug('__init__: id=%s, token=***', id)
super(DigitalOceanProvider, self).__init__(id, *args, **kwargs)
self._client = DigitalOceanClient(token)
self._zone_records = {}
def _data_for_multiple(self, _type, records):
return {
'ttl': records[0]['ttl'],
'type': _type,
'values': [r['data'] 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:
values.append({
'flags': record['flags'],
'tag': record['tag'],
'value': record['data'],
})
return {
'ttl': records[0]['ttl'],
'type': _type,
'values': values
}
def _data_for_CNAME(self, _type, records):
record = records[0]
return {
'ttl': record['ttl'],
'type': _type,
'value': f'{record["data"]}.'
}
def _data_for_MX(self, _type, records):
values = []
for record in records:
values.append({
'preference': record['priority'],
'exchange': f'{record["data"]}.'
})
return {
'ttl': records[0]['ttl'],
'type': _type,
'values': values
}
def _data_for_NS(self, _type, records):
values = []
for record in records:
values.append(f'{record["data"]}.')
return {
'ttl': records[0]['ttl'],
'type': _type,
'values': values,
}
def _data_for_SRV(self, _type, records):
values = []
for record in records:
target = f'{record["data"]}.' if record['data'] != "." else "."
values.append({
'port': record['port'],
'priority': record['priority'],
'target': target,
'weight': record['weight']
})
return {
'type': _type,
'ttl': records[0]['ttl'],
'values': values
}
def _data_for_TXT(self, _type, records):
values = [value['data'].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 DigitalOceanClientNotFound:
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['name']][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 {
'data': 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 {
'data': value.value,
'flags': value.flags,
'name': record.name,
'tag': value.tag,
'ttl': record.ttl,
'type': record._type
}
def _params_for_single(self, record):
yield {
'data': 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 {
'data': value.exchange,
'name': record.name,
'priority': value.preference,
'ttl': record.ttl,
'type': record._type
}
def _params_for_SRV(self, record):
for value in record.values:
yield {
'data': value.target,
'name': record.name,
'port': value.port,
'priority': value.priority,
'ttl': record.ttl,
'type': record._type,
'weight': value.weight
}
def _params_for_TXT(self, record):
# DigitalOcean doesn't want things escaped in values so we
# have to strip them here and add them when going the other way
for value in record.values:
yield {
'data': 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):
if existing.name == record['name'] 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 DigitalOceanClientNotFound:
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('DigitalOcean')
try:
logger.warn('octodns_digitalocean shimmed. Update your provider class to '
'octodns_digitalocean.DigitalOceanProvider. '
'Shim will be removed in 1.0')
from octodns_digitalocean import DigitalOceanProvider
DigitalOceanProvider # pragma: no cover
except ModuleNotFoundError:
logger.exception('DigitalOceanProvider has been moved into a seperate '
'module, octodns_digitalocean is now required. Provider '
'class should be updated to '
'octodns_digitalocean.DigitalOceanProvider')
raise

+ 0
- 188
tests/fixtures/digitalocean-page-1.json View File

@ -1,188 +0,0 @@
{
"domain_records": [{
"id": null,
"type": "SOA",
"name": "@",
"data": null,
"priority": null,
"port": null,
"ttl": null,
"weight": null,
"flags": null,
"tag": null
}, {
"id": 11189874,
"type": "NS",
"name": "@",
"data": "ns1.digitalocean.com",
"priority": null,
"port": null,
"ttl": 3600,
"weight": null,
"flags": null,
"tag": null
}, {
"id": 11189875,
"type": "NS",
"name": "@",
"data": "ns2.digitalocean.com",
"priority": null,
"port": null,
"ttl": 3600,
"weight": null,
"flags": null,
"tag": null
}, {
"id": 11189876,
"type": "NS",
"name": "@",
"data": "ns3.digitalocean.com",
"priority": null,
"port": null,
"ttl": 3600,
"weight": null,
"flags": null,
"tag": null
}, {
"id": 11189877,
"type": "NS",
"name": "under",
"data": "ns1.unit.tests",
"priority": null,
"port": null,
"ttl": 3600,
"weight": null,
"flags": null,
"tag": null
}, {
"id": 11189878,
"type": "NS",
"name": "under",
"data": "ns2.unit.tests",
"priority": null,
"port": null,
"ttl": 3600,
"weight": null,
"flags": null,
"tag": null
}, {
"id": 11189879,
"type": "SRV",
"name": "_srv._tcp",
"data": "foo-1.unit.tests",
"priority": 10,
"port": 30,
"ttl": 600,
"weight": 20,
"flags": null,
"tag": null
}, {
"id": 11189880,
"type": "SRV",
"name": "_srv._tcp",
"data": "foo-2.unit.tests",
"priority": 12,
"port": 30,
"ttl": 600,
"weight": 20,
"flags": null,
"tag": null
}, {
"id": 11189881,
"type": "TXT",
"name": "txt",
"data": "Bah bah black sheep",
"priority": null,
"port": null,
"ttl": 600,
"weight": null,
"flags": null,
"tag": null
}, {
"id": 11189882,
"type": "TXT",
"name": "txt",
"data": "have you any wool.",
"priority": null,
"port": null,
"ttl": 600,
"weight": null,
"flags": null,
"tag": null
}, {
"id": 11189883,
"type": "A",
"name": "@",
"data": "1.2.3.4",
"priority": null,
"port": null,
"ttl": 300,
"weight": null,
"flags": null,
"tag": null
}, {
"id": 11189884,
"type": "A",
"name": "@",
"data": "1.2.3.5",
"priority": null,
"port": null,
"ttl": 300,
"weight": null,
"flags": null,
"tag": null
}, {
"id": 11189885,
"type": "A",
"name": "www",
"data": "2.2.3.6",
"priority": null,
"port": null,
"ttl": 300,
"weight": null,
"flags": null,
"tag": null
}, {
"id": 11189886,
"type": "MX",
"name": "mx",
"data": "smtp-4.unit.tests",
"priority": 10,
"port": null,
"ttl": 300,
"weight": null,
"flags": null,
"tag": null
}, {
"id": 11189887,
"type": "MX",
"name": "mx",
"data": "smtp-2.unit.tests",
"priority": 20,
"port": null,
"ttl": 300,
"weight": null,
"flags": null,
"tag": null
}, {
"id": 11189888,
"type": "MX",
"name": "mx",
"data": "smtp-3.unit.tests",
"priority": 30,
"port": null,
"ttl": 300,
"weight": null,
"flags": null,
"tag": null
}],
"links": {
"pages": {
"last": "https://api.digitalocean.com/v2/domains/unit.tests/records?page=2",
"next": "https://api.digitalocean.com/v2/domains/unit.tests/records?page=2"
}
},
"meta": {
"total": 21
}
}

+ 0
- 111
tests/fixtures/digitalocean-page-2.json View File

@ -1,111 +0,0 @@
{
"domain_records": [{
"id": 11189889,
"type": "MX",
"name": "mx",
"data": "smtp-1.unit.tests",
"priority": 40,
"port": null,
"ttl": 300,
"weight": null,
"flags": null,
"tag": null
}, {
"id": 11189890,
"type": "AAAA",
"name": "aaaa",
"data": "2601:644:500:e210:62f8:1dff:feb8:947a",
"priority": null,
"port": null,
"ttl": 600,
"weight": null,
"flags": null,
"tag": null
}, {
"id": 11189891,
"type": "CNAME",
"name": "cname",
"data": "@",
"priority": null,
"port": null,
"ttl": 300,
"weight": null,
"flags": null,
"tag": null
}, {
"id": 11189892,
"type": "A",
"name": "www.sub",
"data": "2.2.3.6",
"priority": null,
"port": null,
"ttl": 300,
"weight": null,
"flags": null,
"tag": null
}, {
"id": 11189893,
"type": "TXT",
"name": "txt",
"data": "v=DKIM1;k=rsa;s=email;h=sha256;p=A/kinda+of/long/string+with+numb3rs",
"priority": null,
"port": null,
"ttl": 600,
"weight": null,
"flags": null,
"tag": null
}, {
"id": 11189894,
"type": "CAA",
"name": "@",
"data": "ca.unit.tests",
"priority": null,
"port": null,
"ttl": 3600,
"weight": null,
"flags": 0,
"tag": "issue"
}, {
"id": 11189895,
"type": "CNAME",
"name": "included",
"data": "@",
"priority": null,
"port": null,
"ttl": 3600,
"weight": null,
"flags": null,
"tag": null
}, {
"id": 11189896,
"type": "SRV",
"name": "_imap._tcp",
"data": ".",
"priority": 0,
"port": 0,
"ttl": 600,
"weight": 0,
"flags": null,
"tag": null
}, {
"id": 11189897,
"type": "SRV",
"name": "_pop3._tcp",
"data": ".",
"priority": 0,
"port": 0,
"ttl": 600,
"weight": 0,
"flags": null,
"tag": null
}],
"links": {
"pages": {
"first": "https://api.digitalocean.com/v2/domains/unit.tests/records?page=1",
"prev": "https://api.digitalocean.com/v2/domains/unit.tests/records?page=1"
}
},
"meta": {
"total": 21
}
}

+ 5
- 263
tests/test_octodns_provider_digitalocean.py View File

@ -2,273 +2,15 @@
#
#
from __future__ import absolute_import, division, print_function, \
unicode_literals
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 octodns.record import Record
from octodns.provider.digitalocean import DigitalOceanClientNotFound, \
DigitalOceanProvider
from octodns.provider.yaml import YamlProvider
from octodns.zone import Zone
class TestDigitalOceanProvider(TestCase):
expected = Zone('unit.tests.', [])
source = YamlProvider('test', join(dirname(__file__), 'config'))
source.populate(expected)
# Our test suite differs a bit, add our NS and remove the simple one
expected.add_record(Record.new(expected, 'under', {
'ttl': 3600,
'type': 'NS',
'values': [
'ns1.unit.tests.',
'ns2.unit.tests.',
]
}))
for record in list(expected.records):
if record.name == 'sub' and record._type == 'NS':
expected._remove_record(record)
break
def test_populate(self):
provider = DigitalOceanProvider('test', 'token')
# 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))
# 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://api.digitalocean.com/v2/domains/unit.tests/' \
'records?page='
with open('tests/fixtures/digitalocean-page-1.json') as fh:
mock.get(f'{base}1', text=fh.read())
with open('tests/fixtures/digitalocean-page-2.json') as fh:
mock.get(f'{base}2', text=fh.read())
zone = Zone('unit.tests.', [])
provider.populate(zone)
self.assertEquals(14, 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(14, len(again.records))
# bust the cache
del provider._zone_records[zone.name]
def test_apply(self):
provider = DigitalOceanProvider('test', 'token')
resp = Mock()
resp.json = Mock()
provider._client._request = Mock(return_value=resp)
domain_after_creation = {
"domain_records": [{
"id": 11189874,
"type": "NS",
"name": "@",
"data": "ns1.digitalocean.com",
"priority": None,
"port": None,
"ttl": 3600,
"weight": None,
"flags": None,
"tag": None
}, {
"id": 11189875,
"type": "NS",
"name": "@",
"data": "ns2.digitalocean.com",
"priority": None,
"port": None,
"ttl": 3600,
"weight": None,
"flags": None,
"tag": None
}, {
"id": 11189876,
"type": "NS",
"name": "@",
"data": "ns3.digitalocean.com",
"priority": None,
"port": None,
"ttl": 3600,
"weight": None,
"flags": None,
"tag": None
}, {
"id": 11189877,
"type": "A",
"name": "@",
"data": "192.0.2.1",
"priority": None,
"port": None,
"ttl": 3600,
"weight": None,
"flags": None,
"tag": None
}],
"links": {},
"meta": {
"total": 4
}
}
# non-existent domain, create everything
resp.json.side_effect = [
DigitalOceanClientNotFound, # no zone in populate
DigitalOceanClientNotFound, # no domain during apply
domain_after_creation
]
plan = provider.plan(self.expected)
# No root NS, no ignored, no excluded, no unsupported
n = len(self.expected.records) - 10
self.assertEquals(n, len(plan.changes))
self.assertEquals(n, provider.apply(plan))
self.assertFalse(plan.exists)
provider._client._request.assert_has_calls([
# created the domain
call('POST', '/domains', data={'ip_address': '192.0.2.1',
'name': 'unit.tests'}),
# get all records in newly created zone
call('GET', '/domains/unit.tests/records', {'page': 1}),
# delete the initial A record
call('DELETE', '/domains/unit.tests/records/11189877'),
# created at least some of the record with expected data
call('POST', '/domains/unit.tests/records', data={
'data': '1.2.3.4',
'name': '@',
'ttl': 300, 'type': 'A'}),
call('POST', '/domains/unit.tests/records', data={
'data': '1.2.3.5',
'name': '@',
'ttl': 300, 'type': 'A'}),
call('POST', '/domains/unit.tests/records', data={
'data': 'ca.unit.tests',
'flags': 0, 'name': '@',
'tag': 'issue',
'ttl': 3600, 'type': 'CAA'}),
call('POST', '/domains/unit.tests/records', data={
'name': '_imap._tcp',
'weight': 0,
'data': '.',
'priority': 0,
'ttl': 600,
'type': 'SRV',
'port': 0
}),
call('POST', '/domains/unit.tests/records', data={
'name': '_pop3._tcp',
'weight': 0,
'data': '.',
'priority': 0,
'ttl': 600,
'type': 'SRV',
'port': 0
}),
call('POST', '/domains/unit.tests/records', data={
'name': '_srv._tcp',
'weight': 20,
'data': 'foo-1.unit.tests.',
'priority': 10,
'ttl': 600,
'type': 'SRV',
'port': 30
}),
])
self.assertEquals(26, provider._client._request.call_count)
provider._client._request.reset_mock()
# delete 1 and update 1
provider._client.records = Mock(return_value=[
{
'id': 11189897,
'name': 'www',
'data': '1.2.3.4',
'ttl': 300,
'type': 'A',
},
{
'id': 11189898,
'name': 'www',
'data': '2.2.3.4',
'ttl': 300,
'type': 'A',
},
{
'id': 11189899,
'name': 'ttl',
'data': '3.2.3.4',
'ttl': 600,
'type': 'A',
}
])
# Domain exists, we don't care about return
resp.json.side_effect = ['{}']
wanted = Zone('unit.tests.', [])
wanted.add_record(Record.new(wanted, 'ttl', {
'ttl': 300,
'type': 'A',
'value': '3.2.3.4'
}))
class TestDigitalOceanShim(TestCase):
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('POST', '/domains/unit.tests/records', data={
'data': '3.2.3.4',
'type': 'A',
'name': 'ttl',
'ttl': 300
}),
call('DELETE', '/domains/unit.tests/records/11189899'),
call('DELETE', '/domains/unit.tests/records/11189897'),
call('DELETE', '/domains/unit.tests/records/11189898')
], any_order=True)
def test_missing(self):
with self.assertRaises(ModuleNotFoundError):
from octodns.provider.digitalocean import DigitalOceanProvider
DigitalOceanProvider

Write Preview
Loading…
Cancel
Save