#!/usr/bin/env python # Copyright (c) Ansible Project # GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt) # SPDX-License-Identifier: GPL-3.0-or-later from __future__ import (absolute_import, division, print_function) __metaclass__ = type import importlib import os import re import sys import textwrap import yaml PROVIDERS = ['hetzner', 'hosttech'] DEPENDENT_FRAGMENTS = [ ('RECORD_TYPE_CHOICES', [ ('record_type', 'options.type', [ 'module_record', 'module_record_info', 'module_record_set', 'module_record_set_info', ]), ]), ('RECORD_DEFAULT_TTL', [ ('record_default_ttl', 'options.ttl', [ 'module_record', 'module_record_set', ]), ]), ('RECORD_TYPE_CHOICES_RECORD_SETS_MODULE', [ ('record_type', 'options.record_sets.suboptions.type', [ 'module_record_sets', ]), ('record_default_ttl', 'options.record_sets.suboptions.ttl', [ 'module_record_sets', ]), ]), ('RECORD_TYPE_CHOICES_RECORDS_INVENTORY', [ ('record_type', 'options.simple_filters.suboptions.type', [ 'inventory_records', ]), ]), ('ZONE_ID_TYPE', [ ('zone_id_type', 'options.zone_id', [ 'module_record', 'module_record_info', 'module_record_set', 'module_record_set_info', 'module_record_sets', 'module_zone_info', 'inventory_records', ]), ]), ] def get_provider_informations(providers): files_to_remove = [] def add_init_py(path): path = os.path.join(path, '__init__.py') if os.path.exists(path): return with open(path, 'wb') as f: f.write(b'') files_to_remove.append(path) try: sys.path.append(os.path.join('..', '..', '..')) add_init_py(os.path.join('..', '..')) add_init_py(os.path.join('..')) add_init_py(os.path.join('.')) add_init_py(os.path.join('plugins')) add_init_py(os.path.join('plugins', 'module_utils')) provider_infos = {} errors = [] for provider in providers: add_init_py(os.path.join('plugins', 'module_utils', provider)) full_py_path = 'ansible_collections.community.dns.plugins.module_utils.{0}.api'.format(provider) full_pathname = os.path.join('plugins', 'module_utils', provider, 'api.py') try: loader = importlib.machinery.SourceFileLoader(full_py_path, full_pathname) spec = importlib.util.spec_from_loader(full_py_path, loader) the_module = importlib.util.module_from_spec(spec) loader.exec_module(the_module) except Exception as e: errors.append('{0}: Error while importing module {1}: {2}'.format(full_pathname, full_py_path, e)) continue create_provider_info_fn_name = 'create_{0}_provider_information'.format(provider) try: create_provider_info_fn = provider_information = the_module.__dict__[create_provider_info_fn_name] provider_infos[provider] = create_provider_info_fn() except KeyError as e: errors.append('{0}: Cannot find function {1}'.format(full_pathname, create_provider_info_fn)) except Exception as e: errors.append('{0}: Error while invoking function {1}: {2}'.format(full_pathname, create_provider_info_fn, e)) return provider_infos, errors finally: for path in files_to_remove: os.remove(path) class DocFragmentParseError(Exception): def __init__(self, path, error_message): self.path = path self.error_message = error_message super(DocFragmentParseError, self).__init__('Error while parsing {0}: {1}'.format(path, error_message)) DOC_FRAGMENT_START_MATCHER = re.compile(r"^ ([A-Z_]+) = r'''$") class Dumper(yaml.SafeDumper): def ignore_aliases(self, *args): return True def increase_indent(self, flow=False, *args, **kwargs): self.best_indent = kwargs.pop('ident_override', 4) return super().increase_indent(*args, **kwargs) def expect_block_sequence(self): self.increase_indent(flow=False, indentless=False, ident_override=2) self.state = self.expect_first_block_sequence_item class DocFragment: def __init__(self, path, prefix_lines, name, lines): self.prefix_lines = prefix_lines self.name = name self.lines = lines try: self.data = yaml.safe_load('\n'.join(self.lines)) except Exception as e: raise DocFragmentParseError(path, 'Error while parsing part {0}: {1}'.format(name, e)) def recreate_lines(self): data = yaml.dump(self.data, default_flow_style=False, indent=4, Dumper=Dumper, sort_keys=False) self.lines = data.splitlines() def serialize_lines(self): return self.prefix_lines + [" {0} = r'''".format(self.name)] + self.lines + ["'''"] class DocFragmentFile: def __init__(self, path): with open(path, 'r', encoding='utf-8') as f: lines = f.read().splitlines() self.prefix = [] self.fragments = [] self.fragments_by_name = {} where = 'prefix' for line in lines: if where == 'prefix': self.prefix.append(line) if line == 'class ModuleDocFragment(object):': where = 'body' body_prefix = [] body_name = None body_lines = [] elif where == 'body': if body_name is None: m = DOC_FRAGMENT_START_MATCHER.match(line) if m: body_name = m.group(1) else: body_prefix.append(line) elif line == "'''": fragment = DocFragment(path, body_prefix, body_name, body_lines) self.fragments.append(fragment) self.fragments_by_name[body_name] = fragment body_prefix = [] body_name = None body_lines = [] else: body_lines.append(line) if where == 'prefix': raise DocFragmentParseError(path, 'Cannot find body') def serialize_to_string(self): lines = [] lines.extend(self.prefix) for fragment in self.fragments: lines.extend(fragment.serialize_lines()) lines.append('') return '\n'.join(lines) def doc_fragment_fn(name): return os.path.join('plugins', 'doc_fragments', '{0}.py'.format(name)) def load_doc_fragment(name): fn = doc_fragment_fn(name) return DocFragmentFile(fn) def load_single_doc_fragment(name): fragment = 'DOCUMENTATION' if '.' in name: name, fragment = name.split('.', 1) fragment = fragment.upper() doc_fragment = load_doc_fragment(name) return doc_fragment.fragments_by_name[fragment] def write_doc_fragment(name, doc_fragment): fn = doc_fragment_fn(name) data = doc_fragment.serialize_to_string() with open(fn, 'w', encoding='utf-8') as f: f.write(data) def compare_doc_fragment(name, doc_fragment): fn = doc_fragment_fn(name) data = doc_fragment.serialize_to_string() with open(fn, 'r', encoding='utf-8') as f: compare_data = f.read() return data == compare_data def augment_fragment(provider_fragment, provider_info): data = { 'record_type': {'choices': sorted(provider_info.get_supported_record_types())}, 'zone_id_type': {'type': provider_info.get_zone_id_type()}, 'record_id_type': {'type': provider_info.get_record_id_type()}, 'record_default_ttl': {'default': provider_info.get_record_default_ttl()} } for fragment_name, fragment_insertion_data in DEPENDENT_FRAGMENTS: insertion_fragment = provider_fragment.fragments_by_name.get(fragment_name) if insertion_fragment is None: insertion_fragment = DocFragment('', [], fragment_name, []) provider_fragment.fragments.append(insertion_fragment) provider_fragment.fragments_by_name[fragment_name] = insertion_fragment insertion_fragment.data = {} all_doc_fragment_names = set() for what, insertion_point, doc_fragment_names in fragment_insertion_data: all_doc_fragment_names.update(doc_fragment_names) doc_fragments = [load_single_doc_fragment(doc_fragment) for doc_fragment in doc_fragment_names] insertion_point = insertion_point.split('.') insertion_pos = insertion_fragment.data original_pos = doc_fragments[0].data # FIXME for depth, part in enumerate(insertion_point): if part not in insertion_pos: insertion_pos[part] = {} insertion_pos = insertion_pos[part] original_pos = original_pos[part] if depth >= 2: for x in original_pos: if x not in insertion_pos: insertion_pos[x] = original_pos[x] insertion_pos.update(data[what]) insertion_fragment.prefix_lines = [ '', ' # WARNING: This section is automatically generated by update-docs-fragments.py.', ] insertion_fragment.prefix_lines.extend([ ' # {0}'.format(line) for line in textwrap.wrap( 'It is used to augment the docs fragment{0} {1}.'.format( 's' if len(all_doc_fragment_names) != 1 else '', ', '.join(sorted(all_doc_fragment_names))), width=80) ]) insertion_fragment.prefix_lines.append(' # DO NOT EDIT MANUALLY!') insertion_fragment.recreate_lines() def main(program, arguments): lint = '--lint' in arguments provider_infos, errors = get_provider_informations(PROVIDERS) try: for provider, provider_info in sorted(provider_infos.items()): try: doc_fragment = load_doc_fragment(provider) augment_fragment(doc_fragment, provider_info) if not compare_doc_fragment(provider, doc_fragment): path = doc_fragment_fn(provider) if lint: errors.append('{0}: Needs to be updated by update-docs-fragments.py'.format(path)) else: print('Writing {0}...'.format(path)) write_doc_fragment(provider, doc_fragment) except DocFragmentParseError as e: errors.append('{0}: Error while parsing docs fragment: {1}'.format(e.path, e.error_message)) except Exception as e: errors.append('{0}: Unexpected error: {1}'.format(program, e)) for error in errors: print(error) return 5 if errors else 0 if __name__ == '__main__': sys.exit(main(sys.argv[0], sys.argv[1:]))