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

Implement a sketch of --output-provider support for dump

Browse Source
This commit is contained in:
Ross McFarland
2022-04-14 10:19:33 -07:00
parent 67b7fe333d
commit 17c294042f
4 changed files with 56 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
octodns/cmds/dump.py
View File

@@ -18,6 +18,9 @@ def main():
parser.add_argument('--output-dir', required=True,
help='The directory into which the results will be '
'written (Note: will overwrite existing files)')
parser.add_argument('--output-provider', required=False,
help='The configured provider to use when dumping '
'records. Must support copy() and directory')
parser.add_argument('--lenient', action='store_true', default=False,
help='Ignore record validations and do a best effort '
'dump')
@@ -31,8 +34,9 @@ def main():
args = parser.parse_args()
manager = Manager(args.config_file)
manager.dump(args.zone, args.output_dir, args.lenient, args.split,
*args.source)
manager.dump(zone=args.zone, output_dir=args.output_dir,
output_provider=args.output_provider, lenient=args.lenient,
split=args.split, sources=args.source)
if __name__ == '__main__':

40
octodns/manager.py
View File

@@ -549,25 +549,45 @@ class Manager(object):
return zb.changes(za, _AggregateTarget(a + b))
def dump(self, zone, output_dir, lenient, split, source, *sources):
def dump(self, zone, output_dir, sources, lenient=False, split=False,
output_provider=None):
'''
Dump zone data from the specified source
'''
self.log.info('dump: zone=%s, sources=%s', zone, sources)
# We broke out source to force at least one to be passed, add it to any
# others we got.
sources = [source] + list(sources)
self.log.info('dump: zone=%s, output_dir=%s, output_provider=%s, '
'lenient=%s, split=%s, sources=%s', zone, output_dir,
output_provider, lenient, split, sources)
try:
sources = [self.providers[s] for s in sources]
except KeyError as e:
raise ManagerException(f'Unknown source: {e.args[0]}')
clz = YamlProvider
if split:
clz = SplitYamlProvider
target = clz('dump', output_dir)
if output_provider:
self.log.info('dump: using specified output_provider=%s',
output_provider)
try:
target = self.providers[output_provider]
except KeyError as e:
raise ManagerException(f'Unknown output_provider: {e.args[0]}')
if not hasattr(target, 'directory'):
msg = f'output_provider={output_provider}, does not support ' \
'directory'
raise ManagerException(msg)
if not hasattr(target, 'copy'):
msg = f'output_provider={output_provider}, does not support ' \
'copy'
raise ManagerException(msg)
target = target.copy()
self.log.info('dump: setting directory of output_provider copy to '
'%s', output_dir)
target.directory = output_dir
else:
self.log.info('dump: using custom YamlProvider')
clz = YamlProvider
if split:
clz = SplitYamlProvider
target = clz('dump', output_dir)
zone = Zone(zone, self.configured_sub_zones(zone))
for source in sources:

6
octodns/provider/yaml.py
View File

@@ -126,6 +126,12 @@ class YamlProvider(BaseProvider):
self.populate_should_replace = populate_should_replace
self.supports_root_ns = supports_root_ns
def copy(self):
args = dict(self.__dict__)
args['id'] = f'{args["id"]}-copy'
del args['log']
return self.__class__(**args)
@property
def SUPPORTS_ROOT_NS(self):
return self.supports_root_ns

25
tests/test_octodns_manager.py
View File

@@ -283,24 +283,26 @@ class TestManager(TestCase):
manager = Manager(get_config_filename('simple.yaml'))
with self.assertRaises(ManagerException) as ctx:
manager.dump('unit.tests.', tmpdir.dirname, False, False,
'nope')
manager.dump(zone='unit.tests.', output_dir=tmpdir.dirname,
sources=['nope'])
self.assertEqual('Unknown source: nope', str(ctx.exception))
manager.dump('unit.tests.', tmpdir.dirname, False, False, 'in')
manager.dump(zone='unit.tests.', output_dir=tmpdir.dirname,
sources=['in'])
# make sure this fails with an IOError and not a KeyError when
# tyring to find sub zones
with self.assertRaises(IOError):
manager.dump('unknown.zone.', tmpdir.dirname, False, False,
'in')
manager.dump(zone='unknown.zone.', output_dir=tmpdir.dirname,
sources=['in'])
def test_dump_empty(self):
with TemporaryDirectory() as tmpdir:
environ['YAML_TMP_DIR'] = tmpdir.dirname
manager = Manager(get_config_filename('simple.yaml'))
manager.dump('empty.', tmpdir.dirname, False, False, 'in')
manager.dump(zone='empty.', output_dir=tmpdir.dirname,
sources=['in'])
with open(join(tmpdir.dirname, 'empty.yaml')) as fh:
data = safe_load(fh, False)
@@ -312,17 +314,18 @@ class TestManager(TestCase):
manager = Manager(get_config_filename('simple-split.yaml'))
with self.assertRaises(ManagerException) as ctx:
manager.dump('unit.tests.', tmpdir.dirname, False, True,
'nope')
manager.dump(zone='unit.tests.', output_dir=tmpdir.dirname,
split=True, sources=['nope'])
self.assertEqual('Unknown source: nope', str(ctx.exception))
manager.dump('unit.tests.', tmpdir.dirname, False, True, 'in')
manager.dump(zone='unit.tests.', output_dir=tmpdir.dirname,
split=True, sources=['in'])
# make sure this fails with an OSError and not a KeyError when
# tyring to find sub zones
with self.assertRaises(OSError):
manager.dump('unknown.zone.', tmpdir.dirname, False, True,
'in')
manager.dump(zone='unknown.zone.', output_dir=tmpdir.dirname,
split=True, sources=['in'])
def test_validate_configs(self):
Manager(get_config_filename('simple-validate.yaml')).validate_configs()