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

Add idna tests of eligible_zones and fix some messages/bugs

Browse Source
This commit is contained in:
Ross McFarland
2022-08-18 20:10:11 -07:00
parent 33284381f8
commit 2d0f2ccc5c
3 changed files with 66 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
octodns/manager.py
View File

@@ -471,33 +471,22 @@ class Manager(object):
aliased_zones = {}
futures = []
for zone_name, config in zones.items():
self.log.info('sync: zone=%s', idna_decode(zone_name))
decoded_zone_name = idna_decode(zone_name)
self.log.info('sync: zone=%s', decoded_zone_name)
if 'alias' in config:
source_zone = config['alias']
# Check that the source zone is defined.
if source_zone not in self.config['zones']:
self.log.error(
f'Invalid alias zone {zone_name}, '
f'target {source_zone} does not exist'
)
raise ManagerException(
f'Invalid alias zone {zone_name}: '
f'source zone {source_zone} does '
'not exist'
)
msg = f'Invalid alias zone {decoded_zone_name}: source zone {idna_decode(source_zone)} does not exist'
self.log.error(msg)
raise ManagerException(msg)
# Check that the source zone is not an alias zone itself.
if 'alias' in self.config['zones'][source_zone]:
self.log.error(
f'Invalid alias zone {zone_name}, '
f'target {source_zone} is an alias zone'
)
raise ManagerException(
f'Invalid alias zone {zone_name}: '
f'source zone {source_zone} is an '
'alias zone'
)
msg = f'Invalid alias zone {decoded_zone_name}: source zone {idna_decode(source_zone)} is an alias zone'
self.log.error(msg)
raise ManagerException(msg)
aliased_zones[zone_name] = source_zone
continue
@@ -506,12 +495,16 @@ class Manager(object):
try:
sources = config['sources']
except KeyError:
raise ManagerException(f'Zone {zone_name} is missing sources')
raise ManagerException(
f'Zone {decoded_zone_name} is missing sources'
)
try:
targets = config['targets']
except KeyError:
raise ManagerException(f'Zone {zone_name} is missing targets')
raise ManagerException(
f'Zone {decoded_zone_name} is missing targets'
)
processors = config.get('processors', [])
@@ -540,7 +533,8 @@ class Manager(object):
processors = collected
except KeyError:
raise ManagerException(
f'Zone {zone_name}, unknown ' f'processor: {processor}'
f'Zone {decoded_zone_name}, unknown '
f'processor: {processor}'
)
try:
@@ -553,7 +547,7 @@ class Manager(object):
sources = collected
except KeyError:
raise ManagerException(
f'Zone {zone_name}, unknown ' f'source: {source}'
f'Zone {decoded_zone_name}, unknown ' f'source: {source}'
)
try:
@@ -568,7 +562,7 @@ class Manager(object):
targets = trgs
except KeyError:
raise ManagerException(
f'Zone {zone_name}, unknown ' f'target: {target}'
f'Zone {decoded_zone_name}, unknown ' f'target: {target}'
)
futures.append(
@@ -600,7 +594,7 @@ class Manager(object):
desired_config = desired[zone_source]
except KeyError:
raise ManagerException(
f'Zone {zone_name} cannot be sync '
f'Zone {idna_decode(zone_name)} cannot be synced '
f'without zone {zone_source} sinced '
'it is aliased'
)

3
octodns/provider/yaml.py
View File

@@ -148,7 +148,7 @@ class YamlProvider(BaseProvider):
self.log = logging.getLogger(f'{klass}[{id}]')
self.log.debug(
'__init__: id=%s, directory=%s, default_ttl=%d, '
'nforce_order=%d, populate_should_replace=%d',
'enforce_order=%d, populate_should_replace=%d',
id,
directory,
default_ttl,
@@ -254,6 +254,7 @@ class YamlProvider(BaseProvider):
del d['ttl']
if record._octodns:
d['octodns'] = record._octodns
# we want to output the utf-8 version of the name
data[record.decoded_name].append(d)
# Flatten single element lists

46
tests/test_octodns_manager.py
View File

@@ -183,6 +183,50 @@ class TestManager(TestCase):
).sync(dry_run=False, force=True)
self.assertEqual(33, tc)
def test_idna_eligible_zones(self):
# loading w/simple, but we'll be blowing it away and doing some manual
# stuff
manager = Manager(get_config_filename('simple.yaml'))
# these configs won't be valid, but that's fine we can test what we're
# after based on exceptions raised
manager.config['zones'] = manager._config_zones(
{'déjà.vu.': {}, 'deja.vu.': {}, idna_encode('こんにちは.jp.'): {}}
)
from pprint import pprint
pprint(manager.config['zones'])
# refer to them with utf-8
with self.assertRaises(ManagerException) as ctx:
manager.sync(eligible_zones=('déjà.vu.',))
self.assertEqual('Zone déjà.vu. is missing sources', str(ctx.exception))
with self.assertRaises(ManagerException) as ctx:
manager.sync(eligible_zones=('deja.vu.',))
self.assertEqual('Zone deja.vu. is missing sources', str(ctx.exception))
with self.assertRaises(ManagerException) as ctx:
manager.sync(eligible_zones=('こんにちは.jp.',))
self.assertEqual(
'Zone こんにちは.jp. is missing sources', str(ctx.exception)
)
# refer to them with idna (exceptions are still utf-8
with self.assertRaises(ManagerException) as ctx:
manager.sync(eligible_zones=(idna_encode('déjà.vu.'),))
self.assertEqual('Zone déjà.vu. is missing sources', str(ctx.exception))
with self.assertRaises(ManagerException) as ctx:
manager.sync(eligible_zones=(idna_encode('deja.vu.'),))
self.assertEqual('Zone deja.vu. is missing sources', str(ctx.exception))
with self.assertRaises(ManagerException) as ctx:
manager.sync(eligible_zones=(idna_encode('こんにちは.jp.'),))
self.assertEqual(
'Zone こんにちは.jp. is missing sources', str(ctx.exception)
)
def test_eligible_sources(self):
with TemporaryDirectory() as tmpdir:
environ['YAML_TMP_DIR'] = tmpdir.dirname
@@ -237,7 +281,7 @@ class TestManager(TestCase):
get_config_filename('simple-alias-zone.yaml')
).sync(eligible_zones=["alias.tests."])
self.assertEqual(
'Zone alias.tests. cannot be sync without zone '
'Zone alias.tests. cannot be synced without zone '
'unit.tests. sinced it is aliased',
str(ctx.exception),
)