mirror of
https://github.com/github/octodns.git
synced 2024-05-11 05:55:00 +00:00
257 lines
7.9 KiB
Python
257 lines
7.9 KiB
Python
#
|
|
#
|
|
#
|
|
|
|
from __future__ import absolute_import, division, print_function, \
|
|
unicode_literals
|
|
|
|
from mock import Mock, call, patch
|
|
from nsone.rest.errors import AuthException, ResourceException
|
|
from unittest import TestCase
|
|
|
|
from octodns.record import Delete, Record, Update
|
|
from octodns.provider.ns1 import Ns1Provider
|
|
from octodns.zone import Zone
|
|
|
|
|
|
class DummyZone(object):
|
|
|
|
def __init__(self, records):
|
|
self.data = {
|
|
'records': records
|
|
}
|
|
|
|
|
|
class TestNs1Provider(TestCase):
|
|
zone = Zone('unit.tests.', [])
|
|
expected = set()
|
|
expected.add(Record.new(zone, '', {
|
|
'ttl': 32,
|
|
'type': 'A',
|
|
'value': '1.2.3.4',
|
|
}))
|
|
expected.add(Record.new(zone, 'foo', {
|
|
'ttl': 33,
|
|
'type': 'A',
|
|
'values': ['1.2.3.4', '1.2.3.5'],
|
|
}))
|
|
expected.add(Record.new(zone, 'cname', {
|
|
'ttl': 34,
|
|
'type': 'CNAME',
|
|
'value': 'foo.unit.tests.',
|
|
}))
|
|
expected.add(Record.new(zone, '', {
|
|
'ttl': 35,
|
|
'type': 'MX',
|
|
'values': [{
|
|
'priority': 10,
|
|
'value': 'mx1.unit.tests.',
|
|
}, {
|
|
'priority': 20,
|
|
'value': 'mx2.unit.tests.',
|
|
}]
|
|
}))
|
|
expected.add(Record.new(zone, 'naptr', {
|
|
'ttl': 36,
|
|
'type': 'NAPTR',
|
|
'values': [{
|
|
'flags': 'U',
|
|
'order': 100,
|
|
'preference': 100,
|
|
'regexp': '!^.*$!sip:info@bar.example.com!',
|
|
'replacement': '.',
|
|
'service': 'SIP+D2U',
|
|
}, {
|
|
'flags': 'S',
|
|
'order': 10,
|
|
'preference': 100,
|
|
'regexp': '!^.*$!sip:info@bar.example.com!',
|
|
'replacement': '.',
|
|
'service': 'SIP+D2U',
|
|
}]
|
|
}))
|
|
expected.add(Record.new(zone, '', {
|
|
'ttl': 37,
|
|
'type': 'NS',
|
|
'values': ['ns1.unit.tests.', 'ns2.unit.tests.'],
|
|
}))
|
|
expected.add(Record.new(zone, '_srv._tcp', {
|
|
'ttl': 38,
|
|
'type': 'SRV',
|
|
'values': [{
|
|
'priority': 10,
|
|
'weight': 20,
|
|
'port': 30,
|
|
'target': 'foo-1.unit.tests.',
|
|
}, {
|
|
'priority': 12,
|
|
'weight': 30,
|
|
'port': 30,
|
|
'target': 'foo-2.unit.tests.',
|
|
}]
|
|
}))
|
|
expected.add(Record.new(zone, 'sub', {
|
|
'ttl': 39,
|
|
'type': 'NS',
|
|
'values': ['ns3.unit.tests.', 'ns4.unit.tests.'],
|
|
}))
|
|
|
|
nsone_records = [{
|
|
'type': 'A',
|
|
'ttl': 32,
|
|
'short_answers': ['1.2.3.4'],
|
|
'domain': 'unit.tests.',
|
|
}, {
|
|
'type': 'A',
|
|
'ttl': 33,
|
|
'short_answers': ['1.2.3.4', '1.2.3.5'],
|
|
'domain': 'foo.unit.tests.',
|
|
}, {
|
|
'type': 'CNAME',
|
|
'ttl': 34,
|
|
'short_answers': ['foo.unit.tests.'],
|
|
'domain': 'cname.unit.tests.',
|
|
}, {
|
|
'type': 'MX',
|
|
'ttl': 35,
|
|
'short_answers': ['10 mx1.unit.tests.', '20 mx2.unit.tests.'],
|
|
'domain': 'unit.tests.',
|
|
}, {
|
|
'type': 'NAPTR',
|
|
'ttl': 36,
|
|
'short_answers': [
|
|
'10 100 S SIP+D2U !^.*$!sip:info@bar.example.com! .',
|
|
'100 100 U SIP+D2U !^.*$!sip:info@bar.example.com! .'
|
|
],
|
|
'domain': 'naptr.unit.tests.',
|
|
}, {
|
|
'type': 'NS',
|
|
'ttl': 37,
|
|
'short_answers': ['ns1.unit.tests.', 'ns2.unit.tests.'],
|
|
'domain': 'unit.tests.',
|
|
}, {
|
|
'type': 'SRV',
|
|
'ttl': 38,
|
|
'short_answers': ['12 30 30 foo-2.unit.tests.',
|
|
'10 20 30 foo-1.unit.tests.'],
|
|
'domain': '_srv._tcp.unit.tests.',
|
|
}, {
|
|
'type': 'NS',
|
|
'ttl': 39,
|
|
'short_answers': ['ns3.unit.tests.', 'ns4.unit.tests.'],
|
|
'domain': 'sub.unit.tests.',
|
|
}]
|
|
|
|
@patch('nsone.NSONE.loadZone')
|
|
def test_populate(self, load_mock):
|
|
provider = Ns1Provider('test', 'api-key')
|
|
|
|
# Bad auth
|
|
load_mock.side_effect = AuthException('unauthorized')
|
|
zone = Zone('unit.tests.', [])
|
|
with self.assertRaises(AuthException) as ctx:
|
|
provider.populate(zone)
|
|
self.assertEquals(load_mock.side_effect, ctx.exception)
|
|
|
|
# General error
|
|
load_mock.reset_mock()
|
|
load_mock.side_effect = ResourceException('boom')
|
|
zone = Zone('unit.tests.', [])
|
|
with self.assertRaises(ResourceException) as ctx:
|
|
provider.populate(zone)
|
|
self.assertEquals(load_mock.side_effect, ctx.exception)
|
|
self.assertEquals(('unit.tests',), load_mock.call_args[0])
|
|
|
|
# Non-existant zone doesn't populate anything
|
|
load_mock.reset_mock()
|
|
load_mock.side_effect = \
|
|
ResourceException('server error: zone not found')
|
|
zone = Zone('unit.tests.', [])
|
|
provider.populate(zone)
|
|
self.assertEquals(set(), zone.records)
|
|
self.assertEquals(('unit.tests',), load_mock.call_args[0])
|
|
|
|
# Existing zone w/o records
|
|
load_mock.reset_mock()
|
|
nsone_zone = DummyZone([])
|
|
load_mock.side_effect = [nsone_zone]
|
|
zone = Zone('unit.tests.', [])
|
|
provider.populate(zone)
|
|
self.assertEquals(set(), zone.records)
|
|
self.assertEquals(('unit.tests',), load_mock.call_args[0])
|
|
|
|
# Existing zone w/records
|
|
load_mock.reset_mock()
|
|
nsone_zone = DummyZone(self.nsone_records)
|
|
load_mock.side_effect = [nsone_zone]
|
|
zone = Zone('unit.tests.', [])
|
|
provider.populate(zone)
|
|
self.assertEquals(self.expected, zone.records)
|
|
self.assertEquals(('unit.tests',), load_mock.call_args[0])
|
|
|
|
@patch('nsone.NSONE.createZone')
|
|
@patch('nsone.NSONE.loadZone')
|
|
def test_sync(self, load_mock, create_mock):
|
|
provider = Ns1Provider('test', 'api-key')
|
|
|
|
desired = Zone('unit.tests.', [])
|
|
desired.records.update(self.expected)
|
|
|
|
plan = provider.plan(desired)
|
|
# everything except the root NS
|
|
expected_n = len(self.expected) - 1
|
|
self.assertEquals(expected_n, len(plan.changes))
|
|
|
|
# Fails, general error
|
|
load_mock.reset_mock()
|
|
create_mock.reset_mock()
|
|
load_mock.side_effect = ResourceException('boom')
|
|
with self.assertRaises(ResourceException) as ctx:
|
|
provider.apply(plan)
|
|
self.assertEquals(load_mock.side_effect, ctx.exception)
|
|
|
|
# Fails, bad auth
|
|
load_mock.reset_mock()
|
|
create_mock.reset_mock()
|
|
load_mock.side_effect = \
|
|
ResourceException('server error: zone not found')
|
|
create_mock.side_effect = AuthException('unauthorized')
|
|
with self.assertRaises(AuthException) as ctx:
|
|
provider.apply(plan)
|
|
self.assertEquals(create_mock.side_effect, ctx.exception)
|
|
|
|
# non-existant zone, create
|
|
load_mock.reset_mock()
|
|
create_mock.reset_mock()
|
|
load_mock.side_effect = \
|
|
ResourceException('server error: zone not found')
|
|
create_mock.side_effect = None
|
|
got_n = provider.apply(plan)
|
|
self.assertEquals(expected_n, got_n)
|
|
|
|
# Update & delete
|
|
load_mock.reset_mock()
|
|
create_mock.reset_mock()
|
|
nsone_zone = DummyZone(self.nsone_records + [{
|
|
'type': 'A',
|
|
'ttl': 42,
|
|
'short_answers': ['9.9.9.9'],
|
|
'domain': 'delete-me.unit.tests.',
|
|
}])
|
|
nsone_zone.data['records'][0]['short_answers'][0] = '2.2.2.2'
|
|
nsone_zone.loadRecord = Mock()
|
|
load_mock.side_effect = [nsone_zone, nsone_zone]
|
|
plan = provider.plan(desired)
|
|
self.assertEquals(2, len(plan.changes))
|
|
self.assertIsInstance(plan.changes[0], Update)
|
|
self.assertIsInstance(plan.changes[1], Delete)
|
|
|
|
got_n = provider.apply(plan)
|
|
self.assertEquals(2, got_n)
|
|
nsone_zone.loadRecord.assert_has_calls([
|
|
call('unit.tests', u'A'),
|
|
call().update(answers=[u'1.2.3.4'], ttl=32),
|
|
call('delete-me', u'A'),
|
|
call().delete()
|
|
])
|