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

Always require a root NS in desired, sketch out SUPPORTS_ROOT_NS tests

Browse Source
This commit is contained in:
Ross McFarland
2022-02-10 14:51:01 -08:00
parent 08f98a5e65
commit 5215930109
9 changed files with 208 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
octodns/manager.py
View File

@@ -12,7 +12,6 @@ from sys import stdout
import logging
from .provider.base import BaseProvider
from .provider.plan import Plan
from .provider.yaml import SplitYamlProvider, YamlProvider
from .record import Record
from .yaml import safe_load
@@ -520,8 +519,7 @@ class Manager(object):
source.populate(zone, lenient=lenient)
plan = target.plan(zone)
if plan is None:
plan = Plan(zone, zone, [], False)
# We require at least root NS so there'll always be a plan
target.apply(plan)
def validate_configs(self):

25
octodns/provider/base.py
View File

@@ -49,6 +49,7 @@ class BaseProvider(BaseSource):
provider configuration.
'''
have_root_ns = False
for record in desired.records:
if record._type not in self.SUPPORTS:
msg = f'{record._type} records not supported for {record.fqdn}'
@@ -94,14 +95,22 @@ class BaseProvider(BaseSource):
record = record.copy()
record.values = [record.value]
desired.add_record(record, replace=True)
elif record._type == 'NS' and record.name == '' and \
not self.SUPPORTS_ROOT_NS:
# ignore, we can't manage root NS records
msg = \
f'root NS record not supported for {record.fqdn}'
fallback = 'ignoring it'
self.supports_warn_or_except(msg, fallback)
desired.remove_record(record)
elif record._type == 'NS' and record.name == '':
if self.SUPPORTS_ROOT_NS:
# record that we saw a root NS record
have_root_ns = True
else:
# ignore, we can't manage root NS records
msg = \
f'root NS record not supported for {record.fqdn}'
fallback = 'ignoring it'
self.supports_warn_or_except(msg, fallback)
desired.remove_record(record)
if self.SUPPORTS_ROOT_NS and not have_root_ns:
raise SupportsException(f'{self.id}: provider supports root NS '
'record management, but no record '
f'configured in {desired.name}; aborting')
return desired

5
tests/config/dynamic.tests.yaml
View File

@@ -1,4 +1,9 @@
---
? ''
: - type: NS
values:
- 8.8.8.8.
- 9.9.9.9.
a:
dynamic:
pools:

5
tests/config/empty.yaml
View File

@@ -1 +1,6 @@
---
? ''
: - type: NS
values:
- 4.4.4.4.
- 5.5.5.5.

6
tests/config/split/dynamic.tests.tst/ns.yaml Normal file
View File

@@ -0,0 +1,6 @@
---
? ''
: - type: NS
values:
- 8.8.8.8.
- 9.9.9.9.

5
tests/config/subzone.unit.tests.yaml
View File

@@ -1,4 +1,9 @@
---
? ''
: - type: NS
values:
- 6.6.6.6.
- 7.7.7.7.
2:
type: A
value: 2.4.4.4

19
tests/test_octodns_manager.py
View File

@@ -113,14 +113,14 @@ class TestManager(TestCase):
tc = Manager(get_config_filename('always-dry-run.yaml')) \
.sync(dry_run=False)
# only the stuff from subzone, unit.tests. is always-dry-run
self.assertEqual(3, tc)
self.assertEqual(4, tc)
def test_simple(self):
with TemporaryDirectory() as tmpdir:
environ['YAML_TMP_DIR'] = tmpdir.dirname
tc = Manager(get_config_filename('simple.yaml')) \
.sync(dry_run=False)
self.assertEqual(27, tc)
self.assertEqual(30, tc)
# try with just one of the zones
tc = Manager(get_config_filename('simple.yaml')) \
@@ -130,28 +130,28 @@ class TestManager(TestCase):
# the subzone, with 2 targets
tc = Manager(get_config_filename('simple.yaml')) \
.sync(dry_run=False, eligible_zones=['subzone.unit.tests.'])
self.assertEqual(6, tc)
self.assertEqual(8, tc)
# and finally the empty zone
# and finally the empty zone (only root NS)
tc = Manager(get_config_filename('simple.yaml')) \
.sync(dry_run=False, eligible_zones=['empty.'])
self.assertEqual(0, tc)
self.assertEqual(1, tc)
# Again with force
tc = Manager(get_config_filename('simple.yaml')) \
.sync(dry_run=False, force=True)
self.assertEqual(27, tc)
self.assertEqual(30, tc)
# Again with max_workers = 1
tc = Manager(get_config_filename('simple.yaml'), max_workers=1) \
.sync(dry_run=False, force=True)
self.assertEqual(27, tc)
self.assertEqual(30, tc)
# Include meta
tc = Manager(get_config_filename('simple.yaml'), max_workers=1,
include_meta=True) \
.sync(dry_run=False, force=True)
self.assertEqual(31, tc)
self.assertEqual(34, tc)
def test_eligible_sources(self):
with TemporaryDirectory() as tmpdir:
@@ -301,7 +301,8 @@ class TestManager(TestCase):
with open(join(tmpdir.dirname, 'empty.yaml')) as fh:
data = safe_load(fh, False)
self.assertFalse(data)
# just to root NS
self.assertEqual(1, len(data))
def test_dump_split(self):
with TemporaryDirectory() as tmpdir:

145
tests/test_octodns_provider_base.py
View File

@@ -606,3 +606,148 @@ class TestBaseProvider(TestCase):
strict.supports_warn_or_except('Hello World!', 'Will not see')
self.assertEqual('minimal: Hello World!', str(ctx.exception))
strict.log.warning.assert_not_called()
class TestBaseProviderSupportsRootNs(TestCase):
class Provider(BaseProvider):
log = getLogger('Provider')
SUPPORTS = set(('A', 'NS'))
SUPPORTS_GEO = False
SUPPORTS_ROOT_NS = False
strict_supports = False
def __init__(self, existing=None):
super().__init__('test')
self.existing = existing
def populate(self, zone, target=False, lenient=False):
if self.existing:
for record in self.existing.records:
zone.add_record(record)
return True
return False
zone = Zone('unit.tests.', [])
a_record = Record.new(zone, 'ptr', {
'type': 'A',
'ttl': 3600,
'values': ['1.2.3.4', '2.3.4.5'],
})
ns_record = Record.new(zone, 'sub', {
'type': 'NS',
'ttl': 3600,
'values': ['ns2.foo.com.', 'ns2.bar.com.'],
})
no_root = zone.copy()
no_root.add_record(a_record)
no_root.add_record(ns_record)
root_ns_record = Record.new(zone, '', {
'type': 'NS',
'ttl': 3600,
'values': ['ns1.foo.com.', 'ns1.bar.com.'],
})
has_root = no_root.copy()
has_root.add_record(root_ns_record)
other_root_ns_record = Record.new(zone, '', {
'type': 'NS',
'ttl': 3600,
'values': ['ns4.foo.com.', 'ns4.bar.com.'],
})
different_root = no_root.copy()
different_root.add_record(other_root_ns_record)
# False
def test_supports_root_ns_false_matches(self):
# provider has a matching existing root record
provider = self.Provider(self.has_root)
provider.SUPPORTS_ROOT_NS = False
# matching root NS in the desired
plan = provider.plan(self.has_root)
# there's an existing root record that matches the desired. we don't
# support them, so it doesn't matter whether it matches or not.
# _process_desired_zone will have removed the desired one since it
# can't be managed. this will not result in a plan as
# _process_existing_zone will have done so as well.
self.assertFalse(plan)
def test_supports_root_ns_false_different(self):
# provider has a non-matching existing record
provider = self.Provider(self.different_root)
provider.SUPPORTS_ROOT_NS = False
# different root is in the desired
plan = provider.plan(self.has_root)
# the mis-match doesn't matter since we can't manage the records
# anyway, they will have been removed from the desired and existing.
self.assertFalse(plan)
def test_supports_root_ns_false_missing(self):
# provider has an existing record
provider = self.Provider(self.has_root)
provider.SUPPORTS_ROOT_NS = False
# desired doesn't have a root
plan = provider.plan(self.no_root)
# the mis-match doesn't matter since we can't manage the records
# anyway, they will have been removed from the desired and existing.
self.assertFalse(plan)
# True
def test_supports_root_ns_true_matches(self):
# provider has a matching existing root record
provider = self.Provider(self.has_root)
provider.SUPPORTS_ROOT_NS = True
# case where we have a root NS in the desired
plan = provider.plan(self.has_root)
# there's an existing root record that matches the desired state so no
# changes are needed
self.assertFalse(plan)
def test_supports_root_ns_true_different(self):
# provider has a non-matching existing record
provider = self.Provider(self.different_root)
provider.SUPPORTS_ROOT_NS = True
# case where we have a root NS in the desired
plan = provider.plan(self.has_root)
# there's an existing root record that doesn't match the desired state.
# we'll get a plan that modifies it.
self.assertTrue(plan)
change = plan.changes[0]
self.assertEqual(self.other_root_ns_record, change.existing)
self.assertEqual(self.root_ns_record, change.new)
def test_supports_root_ns_true_create_zone(self):
# provider has no existing records (create)
provider = self.Provider()
provider.SUPPORTS_ROOT_NS = True
# case where we have a root NS in the desired
plan = provider.plan(self.has_root)
# there's no existing root record since we're creating the zone so
# we'll get a plan that creates everything, including it
self.assertTrue(plan)
self.assertEqual(3, len(plan.changes))
change = [c for c in plan.changes
if c.new.name == '' and c.new._type == 'NS'][0]
self.assertFalse(change.existing)
self.assertEqual(self.root_ns_record, change.new)
def test_supports_root_ns_true_missing(self):
# provider has a matching existing root record
provider = self.Provider(self.has_root)
provider.SUPPORTS_ROOT_NS = True
# case where we don't have a configured/desired root NS record. this is
# dangerous so we expect an exception to be thrown to prevent trying to
# delete the critical record
with self.assertRaises(SupportsException) as ctx:
provider.plan(self.no_root)
self.assertEqual('test: provider supports root NS record management, '
'but no record configured in unit.tests.; aborting',
str(ctx.exception))

24
tests/test_octodns_provider_yaml.py
View File

@@ -37,7 +37,7 @@ class TestYamlProvider(TestCase):
self.assertEqual(23, len(zone.records))
source.populate(dynamic_zone)
self.assertEqual(6, len(dynamic_zone.records))
self.assertEqual(7, len(dynamic_zone.records))
# Assumption here is that a clean round-trip means that everything
# worked as expected, data that went in came back out and could be
@@ -67,11 +67,11 @@ class TestYamlProvider(TestCase):
# Dynamic plan
plan = target.plan(dynamic_zone)
self.assertEqual(6, len([c for c in plan.changes
self.assertEqual(7, len([c for c in plan.changes
if isinstance(c, Create)]))
self.assertFalse(isfile(dynamic_yaml_file))
# Apply it
self.assertEqual(6, target.apply(plan))
self.assertEqual(7, target.apply(plan))
self.assertTrue(isfile(dynamic_yaml_file))
# There should be no changes after the round trip
@@ -152,6 +152,10 @@ class TestYamlProvider(TestCase):
self.assertTrue('value' in dyna)
# self.assertTrue('dynamic' in dyna)
dyna = data.pop('')
self.assertTrue('values' in dyna)
# self.assertTrue('dynamic' in dyna)
# make sure nothing is left
self.assertEqual([], list(data.keys()))
@@ -160,9 +164,9 @@ class TestYamlProvider(TestCase):
zone = Zone('empty.', [])
# without it we see everything
# without it we see everything (root NS record)
source.populate(zone)
self.assertEqual(0, len(zone.records))
self.assertEqual(1, len(zone.records))
def test_unsorted(self):
source = YamlProvider('test', join(dirname(__file__), 'config'))
@@ -251,7 +255,7 @@ class TestSplitYamlProvider(TestCase):
self.assertEqual(20, len(zone.records))
source.populate(dynamic_zone)
self.assertEqual(5, len(dynamic_zone.records))
self.assertEqual(6, len(dynamic_zone.records))
with TemporaryDirectory() as td:
# Add some subdirs to make sure that it can create them
@@ -272,11 +276,11 @@ class TestSplitYamlProvider(TestCase):
# Dynamic plan
plan = target.plan(dynamic_zone)
self.assertEqual(5, len([c for c in plan.changes
self.assertEqual(6, len([c for c in plan.changes
if isinstance(c, Create)]))
self.assertFalse(isdir(dynamic_zone_dir))
# Apply it
self.assertEqual(5, target.apply(plan))
self.assertEqual(6, target.apply(plan))
self.assertTrue(isdir(dynamic_zone_dir))
# There should be no changes after the round trip
@@ -401,7 +405,7 @@ class TestOverridingYamlProvider(TestCase):
# Load the base, should see the 5 records
base.populate(zone)
got = {r.name: r for r in zone.records}
self.assertEqual(6, len(got))
self.assertEqual(7, len(got))
# We get the "dynamic" A from the base config
self.assertTrue('dynamic' in got['a'].data)
# No added
@@ -410,7 +414,7 @@ class TestOverridingYamlProvider(TestCase):
# Load the overrides, should replace one and add 1
override.populate(zone)
got = {r.name: r for r in zone.records}
self.assertEqual(7, len(got))
self.assertEqual(8, len(got))
# 'a' was replaced with a generic record
self.assertEqual({
'ttl': 3600,