mirror/github-octodns
1
0
Fork 0
mirror of https://github.com/github/octodns.git synced 2024-05-11 05:55:00 +00:00
Code Issues Releases Activity

Merge branch 'master' into noramlize-ip-addrs

Browse Source
This commit is contained in:
Ross McFarland
2021-08-23 12:23:52 -07:00
committed by GitHub
parent 074de66988 cf60476183
commit 7538b91cdf
5 changed files with 246 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_octodns_provider_base.py
+27 -1
View File
@@ -6,11 +6,12 @@ from __future__ import absolute_import, division, print_function, \
unicode_literals
from logging import getLogger
from mock import MagicMock, call
from six import text_type
from unittest import TestCase
from octodns.processor.base import BaseProcessor
from octodns.provider.base import BaseProvider
from octodns.provider.base import BaseProvider, ProviderException
from octodns.provider.plan import Plan, UnsafePlan
from octodns.record import Create, Delete, Record, Update
from octodns.zone import Zone
@@ -21,6 +22,7 @@ class HelperProvider(BaseProvider):
SUPPORTS = set(('A',))
id = 'test'
strict_supports = False
def __init__(self, extra_changes=[], apply_disabled=False,
include_change_callback=None):
@@ -443,3 +445,27 @@ class TestBaseProvider(TestCase):
delete_pcent_threshold=safe_pcent).raise_if_unsafe()
self.assertTrue('Too many deletes' in text_type(ctx.exception))
def test_supports_warn_or_except(self):
class MinimalProvider(BaseProvider):
SUPPORTS = set()
SUPPORTS_GEO = False
def __init__(self, **kwargs):
self.log = MagicMock()
super(MinimalProvider, self).__init__('minimal', **kwargs)
normal = MinimalProvider(strict_supports=False)
# Should log and not expect
normal.supports_warn_or_except('Hello World!', 'Goodbye')
normal.log.warning.assert_called_once()
normal.log.warning.assert_has_calls([
call('Hello World!; Goodbye')
])
strict = MinimalProvider(strict_supports=True)
# Should log and not expect
with self.assertRaises(ProviderException) as ctx:
strict.supports_warn_or_except('Hello World!', 'Will not see')
self.assertEquals('minimal: Hello World!', text_type(ctx.exception))
strict.log.warning.assert_not_called()
tests/test_octodns_provider_route53.py
+133
View File
@@ -394,6 +394,139 @@ class TestRoute53Provider(TestCase):
return (provider, stubber)
def test_process_desired_zone(self):
provider, stubber = self._get_stubbed_fallback_auth_provider()
# No records, essentially a no-op
desired = Zone('unit.tests.', [])
got = provider._process_desired_zone(desired)
self.assertEquals(desired.records, got.records)
# Record without any geos
desired = Zone('unit.tests.', [])
record = Record.new(desired, 'a', {
'ttl': 30,
'type': 'A',
'value': '1.2.3.4',
'dynamic': {
'pools': {
'one': {
'values': [{
'value': '2.2.3.4',
}],
},
},
'rules': [{
'pool': 'one',
}],
},
})
desired.add_record(record)
got = provider._process_desired_zone(desired)
self.assertEquals(desired.records, got.records)
self.assertEquals(1, len(list(got.records)[0].dynamic.rules))
self.assertFalse('geos' in list(got.records)[0].dynamic.rules[0].data)
# Record where all geos are supported
desired = Zone('unit.tests.', [])
record = Record.new(desired, 'a', {
'ttl': 30,
'type': 'A',
'value': '1.2.3.4',
'dynamic': {
'pools': {
'one': {
'values': [{
'value': '1.2.3.4',
}],
},
'two': {
'values': [{
'value': '2.2.3.4',
}],
},
},
'rules': [{
'geos': ['EU', 'NA-US-OR'],
'pool': 'two',
}, {
'pool': 'one',
}],
},
})
desired.add_record(record)
got = provider._process_desired_zone(desired)
self.assertEquals(2, len(list(got.records)[0].dynamic.rules))
self.assertEquals(['EU', 'NA-US-OR'],
list(got.records)[0].dynamic.rules[0].data['geos'])
self.assertFalse('geos' in list(got.records)[0].dynamic.rules[1].data)
# Record with NA-CA-* only rule which is removed
desired = Zone('unit.tests.', [])
record = Record.new(desired, 'a', {
'ttl': 30,
'type': 'A',
'value': '1.2.3.4',
'dynamic': {
'pools': {
'one': {
'values': [{
'value': '1.2.3.4',
}],
},
'two': {
'values': [{
'value': '2.2.3.4',
}],
},
},
'rules': [{
'geos': ['NA-CA-BC'],
'pool': 'two',
}, {
'pool': 'one',
}],
},
})
desired.add_record(record)
got = provider._process_desired_zone(desired)
self.assertEquals(1, len(list(got.records)[0].dynamic.rules))
self.assertFalse('geos' in list(got.records)[0].dynamic.rules[0].data)
# Record with NA-CA-* rule combined with other geos, filtered
desired = Zone('unit.tests.', [])
record = Record.new(desired, 'a', {
'ttl': 30,
'type': 'A',
'value': '1.2.3.4',
'dynamic': {
'pools': {
'one': {
'values': [{
'value': '1.2.3.4',
}],
},
'two': {
'values': [{
'value': '2.2.3.4',
}],
},
},
'rules': [{
'geos': ['EU', 'NA-CA-NB', 'NA-US-OR'],
'pool': 'two',
}, {
'pool': 'one',
}],
},
})
desired.add_record(record)
got = provider._process_desired_zone(desired)
self.assertEquals(2, len(list(got.records)[0].dynamic.rules))
self.assertEquals(['EU', 'NA-US-OR'],
list(got.records)[0].dynamic.rules[0].data['geos'])
self.assertFalse('geos' in list(got.records)[0].dynamic.rules[1].data)
def test_populate_with_fallback(self):
provider, stubber = self._get_stubbed_fallback_auth_provider()