mirror of
https://github.com/github/octodns.git
synced 2024-05-11 05:55:00 +00:00
Revert deletion of transip code
This commit is contained in:
Maikel Poot
@ -36,9 +36,6 @@
|
||||
previous versions of octoDNS are discouraged and may result in undefined
|
||||
behavior and broken records. See https://github.com/octodns/octodns/pull/749
|
||||
for related discussion.
|
||||
* TransipProvider removed as it currently relies on `suds` which is broken in
|
||||
new python versions and hasn't seen a release since 2010. May return with
|
||||
https://github.com/octodns/octodns/pull/762
|
||||
|
||||
#### Stuff
|
||||
|
||||
|
||||
@ -212,6 +212,7 @@ The above command pulled the existing data out of Route53 and placed the results
|
||||
| [Rackspace](/octodns/provider/rackspace.py) | | A, AAAA, ALIAS, CNAME, MX, NS, PTR, SPF, TXT | No | |
|
||||
| [Route53](/octodns/provider/route53.py) | boto3 | A, AAAA, CAA, CNAME, MX, NAPTR, NS, PTR, SPF, SRV, TXT | Both | CNAME health checks don't support a Host header |
|
||||
| [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 | |
|
||||
| [AxfrSource](/octodns/source/axfr.py) | | A, AAAA, CAA, CNAME, LOC, MX, NS, PTR, SPF, SRV, TXT | No | read-only |
|
||||
| [ZoneFileSource](/octodns/source/axfr.py) | | A, AAAA, CAA, CNAME, MX, NS, PTR, SPF, SRV, TXT | No | read-only |
|
||||
|
||||
354
octodns/provider/transip.py
Normal file
354
octodns/provider/transip.py
Normal file
@ -0,0 +1,354 @@
|
||||
#
|
||||
#
|
||||
#
|
||||
|
||||
from __future__ import absolute_import, division, print_function, \
|
||||
unicode_literals
|
||||
|
||||
from suds import WebFault
|
||||
|
||||
from collections import defaultdict
|
||||
from . import ProviderException
|
||||
from .base import BaseProvider
|
||||
from logging import getLogger
|
||||
from ..record import Record
|
||||
from transip.service.domain import DomainService
|
||||
from transip.service.objects import DnsEntry
|
||||
|
||||
|
||||
class TransipException(ProviderException):
|
||||
pass
|
||||
|
||||
|
||||
class TransipConfigException(TransipException):
|
||||
pass
|
||||
|
||||
|
||||
class TransipNewZoneException(TransipException):
|
||||
pass
|
||||
|
||||
|
||||
class TransipProvider(BaseProvider):
|
||||
'''
|
||||
Transip DNS provider
|
||||
|
||||
transip:
|
||||
class: octodns.provider.transip.TransipProvider
|
||||
# Your Transip account name (required)
|
||||
account: yourname
|
||||
# Path to a private key file (required if key is not used)
|
||||
key_file: /path/to/file
|
||||
# The api key as string (required if key_file is not used)
|
||||
key: |
|
||||
\'''
|
||||
-----BEGIN PRIVATE KEY-----
|
||||
...
|
||||
-----END PRIVATE KEY-----
|
||||
\'''
|
||||
# if both `key_file` and `key` are presented `key_file` is used
|
||||
|
||||
'''
|
||||
SUPPORTS_GEO = False
|
||||
SUPPORTS_DYNAMIC = False
|
||||
SUPPORTS = set(('A', 'AAAA', 'CNAME', 'MX', 'NS', 'SRV', 'SPF', 'TXT',
|
||||
'SSHFP', 'CAA'))
|
||||
# unsupported by OctoDNS: 'TLSA'
|
||||
MIN_TTL = 120
|
||||
TIMEOUT = 15
|
||||
ROOT_RECORD = '@'
|
||||
|
||||
def __init__(self, id, account, key=None, key_file=None, *args, **kwargs):
|
||||
self.log = getLogger('TransipProvider[{}]'.format(id))
|
||||
self.log.debug('__init__: id=%s, account=%s, token=***', id,
|
||||
account)
|
||||
super(TransipProvider, self).__init__(id, *args, **kwargs)
|
||||
|
||||
if key_file is not None:
|
||||
self._client = self._domain_service(account,
|
||||
private_key_file=key_file)
|
||||
elif key is not None:
|
||||
self._client = self._domain_service(account, private_key=key)
|
||||
else:
|
||||
raise TransipConfigException(
|
||||
'Missing `key` or `key_file` parameter in config'
|
||||
)
|
||||
|
||||
self._currentZone = {}
|
||||
|
||||
def _domain_service(self, *args, **kwargs):
|
||||
'This exists only for mocking purposes'
|
||||
return DomainService(*args, **kwargs)
|
||||
|
||||
def populate(self, zone, target=False, lenient=False):
|
||||
|
||||
exists = False
|
||||
self._currentZone = zone
|
||||
self.log.debug('populate: name=%s, target=%s, lenient=%s', zone.name,
|
||||
target, lenient)
|
||||
|
||||
before = len(zone.records)
|
||||
try:
|
||||
zoneInfo = self._client.get_info(zone.name[:-1])
|
||||
except WebFault as e:
|
||||
if e.fault.faultcode == '102' and target is False:
|
||||
# Zone not found in account, and not a target so just
|
||||
# leave an empty zone.
|
||||
return exists
|
||||
elif e.fault.faultcode == '102' and target is True:
|
||||
self.log.warning('populate: Transip can\'t create new zones')
|
||||
raise TransipNewZoneException(
|
||||
('populate: ({}) Transip used ' +
|
||||
'as target for non-existing zone: {}').format(
|
||||
e.fault.faultcode, zone.name))
|
||||
else:
|
||||
self.log.error('populate: (%s) %s ', e.fault.faultcode,
|
||||
e.fault.faultstring)
|
||||
raise e
|
||||
|
||||
self.log.debug('populate: found %s records for zone %s',
|
||||
len(zoneInfo.dnsEntries), zone.name)
|
||||
exists = True
|
||||
if zoneInfo.dnsEntries:
|
||||
values = defaultdict(lambda: defaultdict(list))
|
||||
for record in zoneInfo.dnsEntries:
|
||||
name = zone.hostname_from_fqdn(record['name'])
|
||||
if name == self.ROOT_RECORD:
|
||||
name = ''
|
||||
|
||||
if record['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))
|
||||
record = Record.new(zone, name, data_for(_type, records),
|
||||
source=self, lenient=lenient)
|
||||
zone.add_record(record, lenient=lenient)
|
||||
self.log.info('populate: found %s records, exists = %s',
|
||||
len(zone.records) - before, exists)
|
||||
|
||||
self._currentZone = {}
|
||||
return exists
|
||||
|
||||
def _apply(self, plan):
|
||||
desired = plan.desired
|
||||
changes = plan.changes
|
||||
self.log.debug('apply: zone=%s, changes=%d', desired.name,
|
||||
len(changes))
|
||||
|
||||
self._currentZone = plan.desired
|
||||
try:
|
||||
self._client.get_info(plan.desired.name[:-1])
|
||||
except WebFault as e:
|
||||
self.log.exception('_apply: get_info failed')
|
||||
raise e
|
||||
|
||||
_dns_entries = []
|
||||
for record in plan.desired.records:
|
||||
entries_for = getattr(self, '_entries_for_{}'.format(record._type))
|
||||
|
||||
# Root records have '@' as name
|
||||
name = record.name
|
||||
if name == '':
|
||||
name = self.ROOT_RECORD
|
||||
|
||||
_dns_entries.extend(entries_for(name, record))
|
||||
|
||||
try:
|
||||
self._client.set_dns_entries(plan.desired.name[:-1], _dns_entries)
|
||||
except WebFault as e:
|
||||
self.log.warning(('_apply: Set DNS returned ' +
|
||||
'one or more errors: {}').format(
|
||||
e.fault.faultstring))
|
||||
raise TransipException(200, e.fault.faultstring)
|
||||
|
||||
self._currentZone = {}
|
||||
|
||||
def _entries_for_multiple(self, name, record):
|
||||
_entries = []
|
||||
|
||||
for value in record.values:
|
||||
_entries.append(DnsEntry(name, record.ttl, record._type, value))
|
||||
|
||||
return _entries
|
||||
|
||||
def _entries_for_single(self, name, record):
|
||||
|
||||
return [DnsEntry(name, record.ttl, record._type, record.value)]
|
||||
|
||||
_entries_for_A = _entries_for_multiple
|
||||
_entries_for_AAAA = _entries_for_multiple
|
||||
_entries_for_NS = _entries_for_multiple
|
||||
_entries_for_SPF = _entries_for_multiple
|
||||
_entries_for_CNAME = _entries_for_single
|
||||
|
||||
def _entries_for_MX(self, name, record):
|
||||
_entries = []
|
||||
|
||||
for value in record.values:
|
||||
content = "{} {}".format(value.preference, value.exchange)
|
||||
_entries.append(DnsEntry(name, record.ttl, record._type, content))
|
||||
|
||||
return _entries
|
||||
|
||||
def _entries_for_SRV(self, name, record):
|
||||
_entries = []
|
||||
|
||||
for value in record.values:
|
||||
content = "{} {} {} {}".format(value.priority, value.weight,
|
||||
value.port, value.target)
|
||||
_entries.append(DnsEntry(name, record.ttl, record._type, content))
|
||||
|
||||
return _entries
|
||||
|
||||
def _entries_for_SSHFP(self, name, record):
|
||||
_entries = []
|
||||
|
||||
for value in record.values:
|
||||
content = "{} {} {}".format(value.algorithm,
|
||||
value.fingerprint_type,
|
||||
value.fingerprint)
|
||||
_entries.append(DnsEntry(name, record.ttl, record._type, content))
|
||||
|
||||
return _entries
|
||||
|
||||
def _entries_for_CAA(self, name, record):
|
||||
_entries = []
|
||||
|
||||
for value in record.values:
|
||||
content = "{} {} {}".format(value.flags, value.tag,
|
||||
value.value)
|
||||
_entries.append(DnsEntry(name, record.ttl, record._type, content))
|
||||
|
||||
return _entries
|
||||
|
||||
def _entries_for_TXT(self, name, record):
|
||||
_entries = []
|
||||
|
||||
for value in record.values:
|
||||
value = value.replace('\\;', ';')
|
||||
_entries.append(DnsEntry(name, record.ttl, record._type, value))
|
||||
|
||||
return _entries
|
||||
|
||||
def _parse_to_fqdn(self, value):
|
||||
|
||||
# Enforce switch from suds.sax.text.Text to string
|
||||
value = str(value)
|
||||
|
||||
# TransIP allows '@' as value to alias the root record.
|
||||
# this provider won't set an '@' value, but can be an existing record
|
||||
if value == self.ROOT_RECORD:
|
||||
value = self._currentZone.name
|
||||
|
||||
if value[-1] != '.':
|
||||
self.log.debug('parseToFQDN: changed %s to %s', value,
|
||||
'{}.{}'.format(value, self._currentZone.name))
|
||||
value = '{}.{}'.format(value, self._currentZone.name)
|
||||
|
||||
return value
|
||||
|
||||
def _get_lowest_ttl(self, records):
|
||||
_ttl = 100000
|
||||
for record in records:
|
||||
_ttl = min(_ttl, record['expire'])
|
||||
return _ttl
|
||||
|
||||
def _data_for_multiple(self, _type, records):
|
||||
|
||||
_values = []
|
||||
for record in records:
|
||||
# Enforce switch from suds.sax.text.Text to string
|
||||
_values.append(str(record['content']))
|
||||
|
||||
return {
|
||||
'ttl': self._get_lowest_ttl(records),
|
||||
'type': _type,
|
||||
'values': _values
|
||||
}
|
||||
|
||||
_data_for_A = _data_for_multiple
|
||||
_data_for_AAAA = _data_for_multiple
|
||||
_data_for_NS = _data_for_multiple
|
||||
_data_for_SPF = _data_for_multiple
|
||||
|
||||
def _data_for_CNAME(self, _type, records):
|
||||
return {
|
||||
'ttl': records[0]['expire'],
|
||||
'type': _type,
|
||||
'value': self._parse_to_fqdn(records[0]['content'])
|
||||
}
|
||||
|
||||
def _data_for_MX(self, _type, records):
|
||||
_values = []
|
||||
for record in records:
|
||||
preference, exchange = record['content'].split(" ", 1)
|
||||
_values.append({
|
||||
'preference': preference,
|
||||
'exchange': self._parse_to_fqdn(exchange)
|
||||
})
|
||||
return {
|
||||
'ttl': self._get_lowest_ttl(records),
|
||||
'type': _type,
|
||||
'values': _values
|
||||
}
|
||||
|
||||
def _data_for_SRV(self, _type, records):
|
||||
_values = []
|
||||
for record in records:
|
||||
priority, weight, port, target = record['content'].split(' ', 3)
|
||||
_values.append({
|
||||
'port': port,
|
||||
'priority': priority,
|
||||
'target': self._parse_to_fqdn(target),
|
||||
'weight': weight
|
||||
})
|
||||
|
||||
return {
|
||||
'type': _type,
|
||||
'ttl': self._get_lowest_ttl(records),
|
||||
'values': _values
|
||||
}
|
||||
|
||||
def _data_for_SSHFP(self, _type, records):
|
||||
_values = []
|
||||
for record in records:
|
||||
algorithm, fp_type, fingerprint = record['content'].split(' ', 2)
|
||||
_values.append({
|
||||
'algorithm': algorithm,
|
||||
'fingerprint': fingerprint.lower(),
|
||||
'fingerprint_type': fp_type
|
||||
})
|
||||
|
||||
return {
|
||||
'type': _type,
|
||||
'ttl': self._get_lowest_ttl(records),
|
||||
'values': _values
|
||||
}
|
||||
|
||||
def _data_for_CAA(self, _type, records):
|
||||
_values = []
|
||||
for record in records:
|
||||
flags, tag, value = record['content'].split(' ', 2)
|
||||
_values.append({
|
||||
'flags': flags,
|
||||
'tag': tag,
|
||||
'value': value
|
||||
})
|
||||
|
||||
return {
|
||||
'type': _type,
|
||||
'ttl': self._get_lowest_ttl(records),
|
||||
'values': _values
|
||||
}
|
||||
|
||||
def _data_for_TXT(self, _type, records):
|
||||
_values = []
|
||||
for record in records:
|
||||
_values.append(record['content'].replace(';', '\\;'))
|
||||
|
||||
return {
|
||||
'type': _type,
|
||||
'ttl': self._get_lowest_ttl(records),
|
||||
'values': _values
|
||||
}
|
||||
@ -23,3 +23,5 @@ python-dateutil==2.8.1
|
||||
requests==2.24.0
|
||||
s3transfer==0.3.3
|
||||
setuptools==44.1.1
|
||||
six==1.15.0
|
||||
transip==2.1.2
|
||||
|
||||
291
tests/test_octodns_provider_transip.py
Normal file
291
tests/test_octodns_provider_transip.py
Normal file
@ -0,0 +1,291 @@
|
||||
#
|
||||
#
|
||||
#
|
||||
|
||||
from __future__ import absolute_import, division, print_function, \
|
||||
unicode_literals
|
||||
|
||||
from os.path import dirname, join
|
||||
from six import text_type
|
||||
|
||||
from suds import WebFault
|
||||
|
||||
from mock import patch
|
||||
from unittest import TestCase
|
||||
|
||||
from octodns.provider.transip import TransipProvider
|
||||
from octodns.provider.yaml import YamlProvider
|
||||
from octodns.zone import Zone
|
||||
from transip.service.objects import DnsEntry
|
||||
|
||||
|
||||
class MockFault(object):
|
||||
faultstring = ""
|
||||
faultcode = ""
|
||||
|
||||
def __init__(self, code, string, *args, **kwargs):
|
||||
self.faultstring = string
|
||||
self.faultcode = code
|
||||
|
||||
|
||||
class MockResponse(object):
|
||||
dnsEntries = []
|
||||
|
||||
|
||||
class MockDomainService(object):
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
self.mockupEntries = []
|
||||
self.throw_auth_fault = False
|
||||
|
||||
def mockup(self, records):
|
||||
|
||||
provider = TransipProvider('', '', '')
|
||||
|
||||
_dns_entries = []
|
||||
for record in records:
|
||||
if record._type in provider.SUPPORTS:
|
||||
entries_for = getattr(provider,
|
||||
'_entries_for_{}'.format(record._type))
|
||||
|
||||
# Root records have '@' as name
|
||||
name = record.name
|
||||
if name == '':
|
||||
name = provider.ROOT_RECORD
|
||||
|
||||
_dns_entries.extend(entries_for(name, record))
|
||||
|
||||
# Add a non-supported type
|
||||
# so it triggers the "is supported" (transip.py:115) check and
|
||||
# give 100% code coverage
|
||||
_dns_entries.append(
|
||||
DnsEntry('@', '3600', 'BOGUS', 'ns01.transip.nl.'))
|
||||
|
||||
self.mockupEntries = _dns_entries
|
||||
|
||||
# Skips authentication layer and returns the entries loaded by "Mockup"
|
||||
def get_info(self, domain_name):
|
||||
|
||||
if self.throw_auth_fault:
|
||||
self.raiseInvalidAuth()
|
||||
|
||||
# Special 'domain' to trigger error
|
||||
if str(domain_name) == str('notfound.unit.tests'):
|
||||
self.raiseZoneNotFound()
|
||||
|
||||
result = MockResponse()
|
||||
result.dnsEntries = self.mockupEntries
|
||||
return result
|
||||
|
||||
def set_dns_entries(self, domain_name, dns_entries):
|
||||
|
||||
# Special 'domain' to trigger error
|
||||
if str(domain_name) == str('failsetdns.unit.tests'):
|
||||
self.raiseSaveError()
|
||||
|
||||
return True
|
||||
|
||||
def raiseZoneNotFound(self):
|
||||
fault = MockFault(str('102'), '102 is zone not found')
|
||||
document = {}
|
||||
raise WebFault(fault, document)
|
||||
|
||||
def raiseInvalidAuth(self):
|
||||
fault = MockFault(str('200'), '200 is invalid auth')
|
||||
document = {}
|
||||
raise WebFault(fault, document)
|
||||
|
||||
def raiseSaveError(self):
|
||||
fault = MockFault(str('200'), '202 random error')
|
||||
document = {}
|
||||
raise WebFault(fault, document)
|
||||
|
||||
|
||||
class TestTransipProvider(TestCase):
|
||||
|
||||
bogus_key = str("""-----BEGIN RSA PRIVATE KEY-----
|
||||
MIIEowIBAAKCAQEA0U5HGCkLrz423IyUf3u4cKN2WrNz1x5KNr6PvH2M/zxas+zB
|
||||
elbxkdT3AQ+wmfcIvOuTmFRTHv35q2um1aBrPxVw+2s+lWo28VwIRttwIB1vIeWu
|
||||
lSBnkEZQRLyPI2tH0i5QoMX4CVPf9rvij3Uslimi84jdzDfPFIh6jZ6C8nLipOTG
|
||||
0IMhge1ofVfB0oSy5H+7PYS2858QLAf5ruYbzbAxZRivS402wGmQ0d0Lc1KxraAj
|
||||
kiMM5yj/CkH/Vm2w9I6+tLFeASE4ub5HCP5G/ig4dbYtqZMQMpqyAbGxd5SOVtyn
|
||||
UHagAJUxf8DT3I8PyjEHjxdOPUsxNyRtepO/7QIDAQABAoIBAQC7fiZ7gxE/ezjD
|
||||
2n6PsHFpHVTBLS2gzzZl0dCKZeFvJk6ODJDImaeuHhrh7X8ifMNsEI9XjnojMhl8
|
||||
MGPzy88mZHugDNK0H8B19x5G8v1/Fz7dG5WHas660/HFkS+b59cfdXOugYiOOn9O
|
||||
08HBBpLZNRUOmVUuQfQTjapSwGLG8PocgpyRD4zx0LnldnJcqYCxwCdev+AAsPnq
|
||||
ibNtOd/MYD37w9MEGcaxLE8wGgkv8yd97aTjkgE+tp4zsM4QE4Rag133tsLLNznT
|
||||
4Qr/of15M3NW/DXq/fgctyRcJjZpU66eCXLCz2iRTnLyyxxDC2nwlxKbubV+lcS0
|
||||
S4hbfd/BAoGBAO8jXxEaiybR0aIhhSR5esEc3ymo8R8vBN3ZMJ+vr5jEPXr/ZuFj
|
||||
/R4cZ2XV3VoQJG0pvIOYVPZ5DpJM7W+zSXtJ/7bLXy4Bnmh/rc+YYgC+AXQoLSil
|
||||
iD2OuB2xAzRAK71DVSO0kv8gEEXCersPT2i6+vC2GIlJvLcYbOdRKWGxAoGBAOAQ
|
||||
aJbRLtKujH+kMdoMI7tRlL8XwI+SZf0FcieEu//nFyerTePUhVgEtcE+7eQ7hyhG
|
||||
fIXUFx/wALySoqFzdJDLc8U8pTLhbUaoLOTjkwnCTKQVprhnISqQqqh/0U5u47IE
|
||||
RWzWKN6OHb0CezNTq80Dr6HoxmPCnJHBHn5LinT9AoGAQSpvZpbIIqz8pmTiBl2A
|
||||
QQ2gFpcuFeRXPClKYcmbXVLkuhbNL1BzEniFCLAt4LQTaRf9ghLJ3FyCxwVlkpHV
|
||||
zV4N6/8hkcTpKOraL38D/dXJSaEFJVVuee/hZl3tVJjEEpA9rDwx7ooLRSdJEJ6M
|
||||
ciq55UyKBSdt4KssSiDI2RECgYBL3mJ7xuLy5bWfNsrGiVvD/rC+L928/5ZXIXPw
|
||||
26oI0Yfun7ulDH4GOroMcDF/GYT/Zzac3h7iapLlR0WYI47xxGI0A//wBZLJ3QIu
|
||||
krxkDo2C9e3Y/NqnHgsbOQR3aWbiDT4wxydZjIeXS3LKA2fl6Hyc90PN3cTEOb8I
|
||||
hq2gRQKBgEt0SxhhtyB93SjgTzmUZZ7PiEf0YJatfM6cevmjWHexrZH+x31PB72s
|
||||
fH2BQyTKKzoCLB1k/6HRaMnZdrWyWSZ7JKz3AHJ8+58d0Hr8LTrzDM1L6BbjeDct
|
||||
N4OiVz1I3rbZGYa396lpxO6ku8yCglisL1yrSP6DdEUp66ntpKVd
|
||||
-----END RSA PRIVATE KEY-----""")
|
||||
|
||||
def make_expected(self):
|
||||
expected = Zone('unit.tests.', [])
|
||||
source = YamlProvider('test', join(dirname(__file__), 'config'))
|
||||
source.populate(expected)
|
||||
return expected
|
||||
|
||||
@patch('octodns.provider.transip.TransipProvider._domain_service',
|
||||
return_value=MockDomainService())
|
||||
def test_init(self, _):
|
||||
|
||||
# No key nor key_file
|
||||
with self.assertRaises(Exception) as ctx:
|
||||
TransipProvider('test', 'unittest')
|
||||
|
||||
self.assertEquals(
|
||||
str('Missing `key` or `key_file` parameter in config'),
|
||||
str(ctx.exception))
|
||||
|
||||
# With key
|
||||
TransipProvider('test', 'unittest', key=self.bogus_key)
|
||||
|
||||
# With key_file
|
||||
TransipProvider('test', 'unittest', key_file='/fake/path')
|
||||
|
||||
@patch('suds.client.Client.__init__', new=lambda *args, **kwargs: None)
|
||||
def test_domain_service(self):
|
||||
# Special case smoke test for DomainService to get coverage
|
||||
TransipProvider('test', 'unittest', key=self.bogus_key)
|
||||
|
||||
@patch('octodns.provider.transip.TransipProvider._domain_service',
|
||||
return_value=MockDomainService())
|
||||
def test_populate(self, _):
|
||||
_expected = self.make_expected()
|
||||
|
||||
# Unhappy Plan - Not authenticated
|
||||
# Live test against API, will fail in an unauthorized error
|
||||
with self.assertRaises(WebFault) as ctx:
|
||||
provider = TransipProvider('test', 'unittest', self.bogus_key)
|
||||
provider._client.throw_auth_fault = True
|
||||
zone = Zone('unit.tests.', [])
|
||||
provider.populate(zone, True)
|
||||
|
||||
self.assertEquals(str('WebFault'),
|
||||
str(ctx.exception.__class__.__name__))
|
||||
|
||||
self.assertEquals(str('200'), ctx.exception.fault.faultcode)
|
||||
|
||||
# No more auth problems
|
||||
provider._client.throw_auth_fault = False
|
||||
|
||||
# Unhappy Plan - Zone does not exists
|
||||
# Will trigger an exception if provider is used as a target for a
|
||||
# non-existing zone
|
||||
with self.assertRaises(Exception) as ctx:
|
||||
provider = TransipProvider('test', 'unittest', self.bogus_key)
|
||||
zone = Zone('notfound.unit.tests.', [])
|
||||
provider.populate(zone, True)
|
||||
|
||||
self.assertEquals(str('TransipNewZoneException'),
|
||||
str(ctx.exception.__class__.__name__))
|
||||
|
||||
self.assertEquals(
|
||||
'populate: (102) Transip used as target' +
|
||||
' for non-existing zone: notfound.unit.tests.',
|
||||
text_type(ctx.exception))
|
||||
|
||||
# Happy Plan - Zone does not exists
|
||||
# Won't trigger an exception if provider is NOT used as a target for a
|
||||
# non-existing zone.
|
||||
provider = TransipProvider('test', 'unittest', self.bogus_key)
|
||||
zone = Zone('notfound.unit.tests.', [])
|
||||
provider.populate(zone, False)
|
||||
|
||||
# Happy Plan - Populate with mockup records
|
||||
provider = TransipProvider('test', 'unittest', self.bogus_key)
|
||||
provider._client.mockup(_expected.records)
|
||||
zone = Zone('unit.tests.', [])
|
||||
provider.populate(zone, False)
|
||||
|
||||
# Transip allows relative values for types like cname, mx.
|
||||
# Test is these are correctly appended with the domain
|
||||
provider._currentZone = zone
|
||||
self.assertEquals("www.unit.tests.", provider._parse_to_fqdn("www"))
|
||||
self.assertEquals("www.unit.tests.",
|
||||
provider._parse_to_fqdn("www.unit.tests."))
|
||||
self.assertEquals("www.sub.sub.sub.unit.tests.",
|
||||
provider._parse_to_fqdn("www.sub.sub.sub"))
|
||||
self.assertEquals("unit.tests.",
|
||||
provider._parse_to_fqdn("@"))
|
||||
|
||||
# Happy Plan - Even if the zone has no records the zone should exist
|
||||
provider = TransipProvider('test', 'unittest', self.bogus_key)
|
||||
zone = Zone('unit.tests.', [])
|
||||
exists = provider.populate(zone, True)
|
||||
self.assertTrue(exists, 'populate should return true')
|
||||
|
||||
return
|
||||
|
||||
@patch('octodns.provider.transip.TransipProvider._domain_service',
|
||||
return_value=MockDomainService())
|
||||
def test_plan(self, _):
|
||||
_expected = self.make_expected()
|
||||
|
||||
# Test Happy plan, only create
|
||||
provider = TransipProvider('test', 'unittest', self.bogus_key)
|
||||
plan = provider.plan(_expected)
|
||||
|
||||
self.assertEqual(15, plan.change_counts['Create'])
|
||||
self.assertEqual(0, plan.change_counts['Update'])
|
||||
self.assertEqual(0, plan.change_counts['Delete'])
|
||||
|
||||
return
|
||||
|
||||
@patch('octodns.provider.transip.TransipProvider._domain_service',
|
||||
return_value=MockDomainService())
|
||||
def test_apply(self, _):
|
||||
_expected = self.make_expected()
|
||||
|
||||
# Test happy flow. Create all supoorted records
|
||||
provider = TransipProvider('test', 'unittest', self.bogus_key)
|
||||
plan = provider.plan(_expected)
|
||||
self.assertEqual(15, len(plan.changes))
|
||||
changes = provider.apply(plan)
|
||||
self.assertEqual(changes, len(plan.changes))
|
||||
|
||||
# Test unhappy flow. Trigger 'not found error' in apply stage
|
||||
# This should normally not happen as populate will capture it first
|
||||
# but just in case.
|
||||
changes = [] # reset changes
|
||||
with self.assertRaises(Exception) as ctx:
|
||||
provider = TransipProvider('test', 'unittest', self.bogus_key)
|
||||
plan = provider.plan(_expected)
|
||||
plan.desired.name = 'notfound.unit.tests.'
|
||||
changes = provider.apply(plan)
|
||||
|
||||
# Changes should not be set due to an Exception
|
||||
self.assertEqual([], changes)
|
||||
|
||||
self.assertEquals(str('WebFault'),
|
||||
str(ctx.exception.__class__.__name__))
|
||||
|
||||
self.assertEquals(str('102'), ctx.exception.fault.faultcode)
|
||||
|
||||
# Test unhappy flow. Trigger a unrecoverable error while saving
|
||||
_expected = self.make_expected() # reset expected
|
||||
changes = [] # reset changes
|
||||
|
||||
with self.assertRaises(Exception) as ctx:
|
||||
provider = TransipProvider('test', 'unittest', self.bogus_key)
|
||||
plan = provider.plan(_expected)
|
||||
plan.desired.name = 'failsetdns.unit.tests.'
|
||||
changes = provider.apply(plan)
|
||||
|
||||
# Changes should not be set due to an Exception
|
||||
self.assertEqual([], changes)
|
||||
|
||||
self.assertEquals(str('TransipException'),
|
||||
str(ctx.exception.__class__.__name__))
|
||||
Reference in New Issue
Block a user