Checks Ansible's ansible_builtin_runtime.yml for any missing or extra plugin entries

runtime_checker.py

#!/usr/bin/env python3

import asyncio
import json
import os
import subprocess
import sys
import tempfile
import yaml

PLUGIN_TYPES = [
    'become',
    'cache',
    'callback',
    'cliconf',
    'connection',
    'httpapi',
    'inventory',
    'lookup',
    'netconf',
    'shell',
    'vars',
    'module',
    'strategy',
]

IGNORE_EXTRAS = {
    'callback': {'formerly_core_callback', 'formerly_core_missing_callback', 'formerly_core_removed_callback'},
    'connection': {'redirected_local'},
    'inventory': {'formerly_core_inventory'},
    'lookup': {'formerly_core_lookup'},
    'shell': {'formerly_core_powershell'},
    'module': {'async_status.ps1', 'formerly_core_ping', 'setup.ps1', 'slurp.ps1', 'uses_redirected_action'},
}

def ansible_checkout(ansible_dir, branch):
    subprocess.run(['git', 'checkout', branch], cwd=ansible_dir, stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL)


async def ansible_doc(ansible_dir, *args):
    command_args = list(args)
    if '--json' not in command_args:
        command_args.append('--json')

    #command_args.insert(0, sys.executable)
    command_args.insert(0, os.path.join(ansible_dir, 'bin', 'ansible-doc'))

    doc_process = await asyncio.create_subprocess_exec(sys.executable, *command_args,
        stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, env={
            'PYTHONPATH': os.path.join(ansible_dir, 'lib'),
            'ANSIBLE_COLLECTIONS_PATHS': '',  # Don't pick up any 3rd party collections
        })
    stdout, _ = await doc_process.communicate()

    return json.loads(stdout.decode().strip())


def get_ansible_plugins(ansible_dir):
    if asyncio.get_event_loop().is_closed():
        asyncio.set_event_loop(asyncio.new_event_loop())

    loop = asyncio.get_event_loop()
    commands = asyncio.gather(*[ansible_doc(ansible_dir, '--list', '-t', t) for t in PLUGIN_TYPES])
    results = loop.run_until_complete(commands)
    loop.close()

    return {t: set(results[i].keys()) for i, t in enumerate(PLUGIN_TYPES)}


def filter_aliases(ansible_dir, modules):
    cmd = [
        sys.executable,
        os.path.join(ansible_dir, 'bin', 'ansible-doc'),
        '-t',
        'module',
        '--json',
    ]
    cmd.extend(modules)
    process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env={
        'PYTHONPATH': os.path.join(ansible_dir, 'lib'),
        'ANSIBLE_COLLECTIONS_PATHS': '',  # Don't pick up any 3rd party collections
    })
    stdout, _ = process.communicate()

    doc_output = json.loads(stdout)

    aliased_modules = set(m for m, d in doc_output.items() if m != d['doc']['module'])

    return modules.difference(aliased_modules)


def main():
    with tempfile.TemporaryDirectory() as tempdir:
        ansible_dir = os.path.join(tempdir, 'ansible')
        subprocess.call(['git', 'clone', 'https://github.com/ansible/ansible.git', ansible_dir])

        ansible_checkout(ansible_dir, 'stable-2.10')
        ansible_210 = get_ansible_plugins(ansible_dir)

        runtime_path = os.path.join(ansible_dir, 'lib', 'ansible', 'config', 'ansible_builtin_runtime.yml')
        with open(runtime_path, mode='r') as runtime_fd:
            builtin_runtime = yaml.safe_load(runtime_fd)

        ansible_checkout(ansible_dir, 'stable-2.9')
        ansible_29 = get_ansible_plugins(ansible_dir)

        missing_plugins = {}
        extra_plugins = {}
        for plugin_type, plugins in ansible_29.items():
            ansible_entries = ansible_210.get(plugin_type, {})

            routing_type = 'modules' if plugin_type == 'module' else plugin_type
            routed_entries = set(builtin_runtime['plugin_routing'].get(routing_type, {}).keys())
            total_210_entries = ansible_entries.union(routed_entries)

            missing_entries = plugins.difference(total_210_entries)
            extra_entries = total_210_entries.difference(plugins.union(IGNORE_EXTRAS.get(plugin_type, {})))

            if missing_entries:
                missing_plugins[plugin_type] = sorted(list(missing_entries))

            # ansible-doc does not return aliased modules with --list. We need to filter out the extra entries in
            # redirection that are just an alias for another module
            if extra_entries and plugin_type == 'module':
                extra_entries = filter_aliases(ansible_dir, extra_entries)

            if extra_entries:
                extra_plugins[plugin_type] = sorted(list(extra_entries))

        if missing_plugins:
            print("ansible_builtin_runtime.yml is missing the following routing entries")
            print(yaml.safe_dump(missing_plugins))

        if extra_plugins:
            print("ansible_builtin_runtime.yml contains extra routing entries")
            print(yaml.safe_dump(extra_plugins))


if __name__ == '__main__':
    main()

runtime_result.yml

cliconf:
- apconos
connection:
- aws_ssm
httpapi:
- vmware
lookup:
- lmdb_kv
- openshift
- unvault
module:
- apconos_command
- aws_acm
- aws_step_functions_state_machine
- aws_step_functions_state_machine_execution
- ce_is_is_instance
- ce_is_is_interface
- ce_is_is_view
- ce_lacp
- ce_lldp
- ce_lldp_interface
- ce_mdn_interface
- ce_multicast_global
- ce_multicast_igmp_enable
- ce_static_route_bfd
- cloudformation_exports_info
- cloudwatchlogs_log_group_metric_filter
- ec2_tag_info
- ecs_domain
- ecs_tag
- eos_acl_interfaces
- eos_acls
- eos_ospfv2
- eos_static_routes
- exos_l2_interfaces
- exos_lldp_interfaces
- exos_vlans
- gcp_compute_network_endpoint_group
- gcp_compute_network_endpoint_group_info
- gcp_compute_node_group
- gcp_compute_node_group_info
- gcp_compute_node_template
- gcp_compute_node_template_info
- gcp_compute_region_backend_service
- gcp_compute_region_backend_service_info
- gcp_compute_reservation
- gcp_compute_reservation_info
- gcp_compute_target_instance
- gcp_compute_target_instance_info
- gcp_logging_metric
- gcp_logging_metric_info
- gcp_runtimeconfig_config
- gcp_runtimeconfig_config_info
- gcp_runtimeconfig_variable
- gcp_runtimeconfig_variable_info
- gcp_serviceusage_service
- gcp_serviceusage_service_info
- hcloud_floating_ip
- hetzner_firewall
- hetzner_firewall_info
- hwc_ecs_instance
- hwc_evs_disk
- hwc_vpc_eip
- hwc_vpc_peering_connect
- hwc_vpc_port
- hwc_vpc_private_ip
- hwc_vpc_route
- hwc_vpc_security_group
- hwc_vpc_security_group_rule
- hwc_vpc_subnet
- iam_policy_info
- iam_saml_federation
- iam_user_info
- ios_acl_interfaces
- ios_acls
- ios_ospfv2
- ios_static_routes
- iosxr_acl_interfaces
- iosxr_acls
- iosxr_ospfv2
- iosxr_static_routes
- ipwcli_dns
- junos_acl_interfaces
- junos_acls
- junos_ospfv2
- junos_static_routes
- lbu
- ldap_attrs
- mas
- mongodb_info
- mso_schema_template_external_epg_contract
- mso_schema_template_external_epg_subnet
- mysql_query
- nxos_acl_interfaces
- nxos_acls
- nxos_devicealias
- nxos_hsrp_interfaces
- nxos_lldp_interfaces
- nxos_ospfv2
- nxos_static_routes
- nxos_vsan
- nxos_zone_zoneset
- onyx_aaa
- onyx_bfd
- onyx_ntp
- onyx_ntp_servers_peers
- onyx_snmp
- onyx_snmp_hosts
- onyx_snmp_users
- onyx_syslog_files
- onyx_syslog_remote
- onyx_username
- ovh_monthly_billing
- ovirt_vnic_profile_info
- packet_ip_subnet
- packet_project
- packet_volume
- packet_volume_attachment
- podman_container_info
- podman_volume_info
- postgresql_subscription
- postgresql_user_obj_stat_info
- ucs_query
- vmware_guest_controller
- vmware_guest_cross_vc_clone
- vmware_guest_register_operation
- vmware_guest_serial_port
- vmware_guest_tools_info
- vmware_host_auto_start
- vmware_host_dns
- vmware_vsan_health_info
- vyos_firewall_global
- vyos_firewall_interfaces
- vyos_firewall_rules
- vyos_ospfv2
- vyos_static_routes
- win_initialize_disk
- x509_crl
- x509_crl_info
- zabbix_host_events_info
- zabbix_service
- zabbix_template_info
- zabbix_user
- zabbix_user_info
- zabbix_valuemap