Tool for removing flatmapping from community.general and community.network

unflatmap.py

#!/usr/bin/python3
# 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

import os
import subprocess
import sys

import yaml


def main(argv):
    if len(argv) < 3:
        print(f'Syntax: {argv[0]} <collection_name> <removal_version>')
        sys.exit(1)

    collection = argv[1]
    removal_version = argv[2]

    with open('meta/runtime.yml', 'rb') as f:
        routing = yaml.safe_load(f)

    to_move = []
    rewrite_imports = {}
    plugin_test_directories = {}
    plugin_test_files = {}
    for plugin_type in ('modules', 'action'):
        if plugin_type not in routing['plugin_routing']:
            continue
        result = {}
        for module, data in routing['plugin_routing'][plugin_type].items():
            if 'redirect' not in data:
                result[module] = data
                continue
            if not data['redirect'].startswith(f'{collection}.'):
                result[module] = data
                continue
            real_name = f'{collection}.{module}'
            long_name = data['redirect'][len(f'{collection}.'):]
            long_name_head = '.'.join(long_name.split('.')[:-1])
            last_long_name_part = long_name.split('.')[-1]
            if last_long_name_part != module:
                result[module] = data.copy()
                result[module]['redirect'] = f'{collection}.{last_long_name_part}'
                continue
            to_move.append((f'plugins/{plugin_type}/{"/".join(long_name.split("."))}.py', f'plugins/{plugin_type}/{module}.py'))
            result[long_name] = data.copy()
            result[long_name]['redirect'] = real_name
            if 'deprecation' not in result[long_name]:
                result[long_name]['deprecation'] = {
                    'removal_version': removal_version,
                    'warning_text':
                        f'You are using an internal name to access the {real_name} {plugin_type}.'
                        ' This has never been supported or documented, and will stop working in'
                        f' {collection} {removal_version}.'
                }
            rewrite_imports[f'ansible_collections.{collection}.plugins.{plugin_type}.{long_name_head}'] = f'ansible_collections.{collection}.plugins.{plugin_type}'
            rewrite_imports[f'ansible_collections.{collection}.plugins.{plugin_type}.{long_name}'] = f'ansible_collections.{collection}.plugins.{plugin_type}.{last_long_name_part}'
            rewrite_imports[f'ansible_collections.{collection}.tests.unit.plugins.{plugin_type}.{long_name_head}'] = f'ansible_collections.{collection}.tests.unit.plugins.{plugin_type}'
            plugin_test_directories[os.path.join('tests', 'unit', 'plugins', plugin_type, *long_name_head.split('.'))] = os.path.join('tests', 'unit', 'plugins', plugin_type)
            plugin_test_directories[os.path.join('tests', 'unit', 'plugins', plugin_type, *long_name.split('.'))] = os.path.join('tests', 'unit', 'plugins', plugin_type, module)
            plugin_test_files[os.path.join('tests', 'unit', 'plugins', plugin_type, *long_name_head.split('.'), f'test_{module}.py')] = os.path.join('tests', 'unit', 'plugins', plugin_type, f'test_{module}.py')
        routing['plugin_routing'][plugin_type] = result

    for src, dst in to_move:
        subprocess.run(['git', 'mv', src, dst])

    with open('meta/runtime.yml', 'wb') as f:
        f.write('''---
# 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

'''.encode('utf-8'))
        f.write(yaml.dump(routing, default_flow_style=False, encoding='utf-8', sort_keys=False))

    unit_test_files = []
    for root, dirs, files in os.walk('tests/unit'):
        for file in files:
            path = os.path.join(root, file)
            if path in plugin_test_files:
                subprocess.run(['git', 'mv', path, plugin_test_files[path]])
                unit_test_files.append(plugin_test_files[path])
                continue
            longest_src = None
            longest_dst = None
            for src, dst in plugin_test_directories.items():
                if path.startswith(src + '/') and (longest_src is None or len(longest_src) < len(src)):
                    longest_src = src
                    longest_dst = dst
            if longest_src is not None and longest_dst is not None:
                if file == '__init__.py':
                    subprocess.run(['git', 'rm', path])
                    continue
                new_path = longest_dst + path[len(longest_src):]
                os.makedirs(os.path.dirname(new_path), exist_ok=True)
                subprocess.run(['git', 'mv', path, new_path])
                unit_test_files.append(new_path)
                continue
            unit_test_files.append(path)

    for path in unit_test_files:
        if path.endswith('.py'):
            with open(path, 'rt') as f:
                lines = f.readlines()
            for index, line in enumerate(lines):
                longest_src = None
                longest_dst = None
                for src, dst in rewrite_imports.items():
                    if src in line and (longest_src is None or len(longest_src) < len(src)):
                        longest_src = src
                        longest_dst = dst
                if longest_src is not None:
                    line = line.replace(longest_src, longest_dst)
                    lines[index] = line
            with open(path, 'wt') as f:
                for line in lines:
                    f.write(line)


main(sys.argv)