c22dde61 |
#!/usr/bin/env python |
eb1214ba |
# (c) 2016-2017, Toshio Kuratomi <tkuratomi@ansible.com>
#
# This file is part of Ansible
#
# Ansible is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Ansible is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Ansible. If not, see <http://www.gnu.org/licenses/>.
# Make coding more python3-ish
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type |
c22dde61 |
import ast
import csv
import os
import sys |
ab43a269 |
from collections import defaultdict |
c22dde61 |
from distutils.version import StrictVersion |
c53b7322 |
from pprint import pformat, pprint |
c22dde61 |
|
af2073d0 |
from ansible.parsing.metadata import DEFAULT_METADATA, ParseError, extract_metadata
from ansible.plugins.loader import module_loader |
c22dde61 |
# There's a few files that are not new-style modules. Have to blacklist them
NONMODULE_PY_FILES = frozenset(('async_wrapper.py',))
NONMODULE_MODULE_NAMES = frozenset(os.path.splitext(p)[0] for p in NONMODULE_PY_FILES)
|
ab43a269 |
|
c22dde61 |
class MissingModuleError(Exception):
"""Thrown when unable to find a plugin"""
pass
def usage():
print("""Usage: |
eb1214ba |
metadata-tool.py report [--version X]
metadata-tool.py add [--version X] [--overwrite] CSVFILE
metadata-tool.py add-default [--version X] [--overwrite]
medatada-tool.py upgrade [--version X]""") |
c22dde61 |
sys.exit(1)
def parse_args(arg_string):
if len(arg_string) < 1:
usage()
action = arg_string[0]
version = None
if '--version' in arg_string:
version_location = arg_string.index('--version')
arg_string.pop(version_location)
version = arg_string.pop(version_location)
overwrite = False
if '--overwrite' in arg_string:
overwrite = True
arg_string.remove('--overwrite')
csvfile = None
if len(arg_string) == 2:
csvfile = arg_string[1]
elif len(arg_string) > 2:
usage()
return action, {'version': version, 'overwrite': overwrite, 'csvfile': csvfile}
|
9a5a61bd |
|
c22dde61 |
def find_documentation(module_data):
"""Find the DOCUMENTATION metadata for a module file"""
start_line = -1
mod_ast_tree = ast.parse(module_data)
for child in mod_ast_tree.body:
if isinstance(child, ast.Assign):
for target in child.targets:
if target.id == 'DOCUMENTATION':
start_line = child.lineno - 1
break
return start_line
def remove_metadata(module_data, start_line, start_col, end_line, end_col):
"""Remove a section of a module file"""
lines = module_data.split('\n')
new_lines = lines[:start_line]
if start_col != 0:
new_lines.append(lines[start_line][:start_col])
next_line = lines[end_line]
if len(next_line) - 1 != end_col:
new_lines.append(next_line[end_col:])
if len(lines) > end_line:
new_lines.extend(lines[end_line + 1:])
return '\n'.join(new_lines)
def insert_metadata(module_data, new_metadata, insertion_line, targets=('ANSIBLE_METADATA',)):
"""Insert a new set of metadata at a specified line""" |
c53b7322 |
assignments = ' = '.join(targets)
pretty_metadata = pformat(new_metadata, width=1).split('\n')
new_lines = [] |
797664d9 |
new_lines.append('{0} = {1}'.format(assignments, pretty_metadata[0])) |
c53b7322 |
if len(pretty_metadata) > 1:
for line in pretty_metadata[1:]: |
797664d9 |
new_lines.append('{0}{1}'.format(' ' * (len(assignments) - 1 + len(' = {')), line)) |
c53b7322 |
old_lines = module_data.split('\n') |
af2073d0 |
lines = old_lines[:insertion_line] + new_lines + old_lines[insertion_line:] |
c22dde61 |
return '\n'.join(lines)
def parse_assigned_metadata_initial(csvfile):
"""
Fields:
:0: Module name
:1: Core (x if so)
:2: Extras (x if so)
:3: Category
:4: Supported/SLA |
eb1214ba |
:5: Curated |
c22dde61 |
:6: Stable
:7: Deprecated
:8: Notes
:9: Team Notes |
d3dd11b2 |
:10: Notes 2
:11: final supported_by field |
c22dde61 |
"""
with open(csvfile, 'rb') as f:
for record in csv.reader(f):
module = record[0]
|
c53b7322 |
if record[12] == 'core': |
c22dde61 |
supported_by = 'core' |
c53b7322 |
elif record[12] == 'curated': |
eb1214ba |
supported_by = 'curated' |
c53b7322 |
elif record[12] == 'community': |
d3dd11b2 |
supported_by = 'community'
else: |
c53b7322 |
print('Module %s has no supported_by field. Using community' % record[0])
supported_by = 'community' |
ab43a269 |
supported_by = DEFAULT_METADATA['supported_by'] |
c22dde61 |
status = []
if record[6]:
status.append('stableinterface')
if record[7]:
status.append('deprecated')
if not status: |
ab43a269 |
status.extend(DEFAULT_METADATA['status']) |
c22dde61 |
|
eb1214ba |
yield (module, {'version': DEFAULT_METADATA['metadata_version'], 'supported_by': supported_by, 'status': status}) |
c22dde61 |
def parse_assigned_metadata(csvfile):
"""
Fields:
:0: Module name
:1: supported_by string. One of the valid support fields |
af2073d0 |
core, community, certified, network |
c22dde61 |
:2: stableinterface
:3: preview
:4: deprecated
:5: removed
|
c262dbfd |
https://docs.ansible.com/ansible/latest/dev_guide/developing_modules_documenting.html#ansible-metadata-block |
c22dde61 |
"""
with open(csvfile, 'rb') as f:
for record in csv.reader(f):
module = record[0]
supported_by = record[1]
status = []
if record[2]:
status.append('stableinterface')
if record[4]:
status.append('deprecated')
if record[5]:
status.append('removed')
if not status or record[3]:
status.append('preview')
|
af2073d0 |
yield (module, {'metadata_version': '1.1', 'supported_by': supported_by, 'status': status}) |
c22dde61 |
def write_metadata(filename, new_metadata, version=None, overwrite=False):
with open(filename, 'rb') as f:
module_data = f.read()
try: |
3ee997b7 |
current_metadata, start_line, start_col, end_line, end_col, targets = \
extract_metadata(module_data=module_data, offsets=True) |
c22dde61 |
except SyntaxError:
if filename.endswith('.py'):
raise
# Probably non-python modules. These should all have python
# documentation files where we can place the data |
797664d9 |
raise ParseError('Could not add metadata to {0}'.format(filename)) |
c22dde61 |
if current_metadata is None: |
b89cb956 |
# No current metadata so we can just add it |
c22dde61 |
start_line = find_documentation(module_data)
if start_line < 0:
if os.path.basename(filename) in NONMODULE_PY_FILES:
# These aren't new-style modules
return
|
797664d9 |
raise Exception('Module file {0} had no ANSIBLE_METADATA or DOCUMENTATION'.format(filename)) |
c22dde61 |
module_data = insert_metadata(module_data, new_metadata, start_line, targets=('ANSIBLE_METADATA',))
|
eb1214ba |
elif overwrite or (version is not None and ('metadata_version' not in current_metadata or
StrictVersion(current_metadata['metadata_version']) < StrictVersion(version))): |
c22dde61 |
# Current metadata that we do not want. Remove the current
# metadata and put the new version in its place
module_data = remove_metadata(module_data, start_line, start_col, end_line, end_col)
module_data = insert_metadata(module_data, new_metadata, start_line, targets=targets)
else:
# Current metadata and we don't want to overwrite it
return
# Save the new version of the module
with open(filename, 'wb') as f:
f.write(module_data)
|
ab43a269 |
def return_metadata(plugins): |
eb1214ba |
"""Get the metadata for all modules |
ab43a269 |
|
eb1214ba |
Handle duplicate module names
:arg plugins: List of plugins to look for
:returns: Mapping of plugin name to metadata dictionary
""" |
ab43a269 |
metadata = {}
for name, filename in plugins:
# There may be several files for a module (if it is written in another
# language, for instance) but only one of them (the .py file) should
# contain the metadata.
if name not in metadata or metadata[name] is not None:
with open(filename, 'rb') as f:
module_data = f.read() |
3ee997b7 |
metadata[name] = extract_metadata(module_data=module_data, offsets=True)[0] |
ab43a269 |
return metadata
|
eb1214ba |
|
c22dde61 |
def metadata_summary(plugins, version=None):
"""Compile information about the metadata status for a list of modules
:arg plugins: List of plugins to look for. Each entry in the list is
a tuple of (module name, full path to module)
:kwarg version: If given, make sure the modules have this version of |
ab43a269 |
metadata or higher. |
c22dde61 |
:returns: A tuple consisting of a list of modules with no metadata at the
required version and a list of files that have metadata at the
required version.
"""
no_metadata = {}
has_metadata = {} |
ab43a269 |
supported_by = defaultdict(set)
status = defaultdict(set) |
eb1214ba |
requested_version = StrictVersion(version) |
ab43a269 |
all_mods_metadata = return_metadata(plugins) |
c22dde61 |
for name, filename in plugins: |
ab43a269 |
# Does the module have metadata? |
c22dde61 |
if name not in no_metadata and name not in has_metadata: |
ab43a269 |
metadata = all_mods_metadata[name] |
c22dde61 |
if metadata is None:
no_metadata[name] = filename |
eb1214ba |
elif version is not None and ('metadata_version' not in metadata or StrictVersion(metadata['metadata_version']) < requested_version): |
c22dde61 |
no_metadata[name] = filename
else:
has_metadata[name] = filename
|
ab43a269 |
# What categories does the plugin belong in?
if all_mods_metadata[name] is None:
# No metadata for this module. Use the default metadata
supported_by[DEFAULT_METADATA['supported_by']].add(filename)
status[DEFAULT_METADATA['status'][0]].add(filename)
else:
supported_by[all_mods_metadata[name]['supported_by']].add(filename)
for one_status in all_mods_metadata[name]['status']:
status[one_status].add(filename)
return list(no_metadata.values()), list(has_metadata.values()), supported_by, status |
c22dde61 |
|
eb1214ba |
# Filters to convert between metadata versions |
9a5a61bd |
|
eb1214ba |
def convert_metadata_pre_1_0_to_1_0(metadata):
"""
Convert pre-1.0 to 1.0 metadata format
:arg metadata: The old metadata
:returns: The new metadata
Changes from pre-1.0 to 1.0:
* ``version`` field renamed to ``metadata_version``
* ``supported_by`` field value ``unmaintained`` has been removed (change to
``community`` and let an external list track whether a module is unmaintained)
* ``supported_by`` field value ``committer`` has been renamed to ``curated``
"""
new_metadata = {'metadata_version': '1.0',
'supported_by': metadata['supported_by'],
'status': metadata['status']
}
if new_metadata['supported_by'] == 'unmaintained':
new_metadata['supported_by'] = 'community'
elif new_metadata['supported_by'] == 'committer':
new_metadata['supported_by'] = 'curated'
return new_metadata
|
af2073d0 |
def convert_metadata_1_0_to_1_1(metadata):
"""
Convert 1.0 to 1.1 metadata format
:arg metadata: The old metadata
:returns: The new metadata
Changes from 1.0 to 1.1:
* ``supported_by`` field value ``curated`` has been removed
* ``supported_by`` field value ``certified`` has been added
* ``supported_by`` field value ``network`` has been added
"""
new_metadata = {'metadata_version': '1.1',
'supported_by': metadata['supported_by'],
'status': metadata['status']
}
if new_metadata['supported_by'] == 'unmaintained':
new_metadata['supported_by'] = 'community'
elif new_metadata['supported_by'] == 'curated':
new_metadata['supported_by'] = 'certified'
return new_metadata
|
c22dde61 |
# Subcommands |
9a5a61bd |
|
c22dde61 |
def add_from_csv(csv_file, version=None, overwrite=False):
"""Implement the subcommand to add metadata from a csv file
"""
# Add metadata for everything from the CSV file
diagnostic_messages = [] |
af2073d0 |
for module_name, new_metadata in parse_assigned_metadata(csv_file): |
c22dde61 |
filename = module_loader.find_plugin(module_name, mod_type='.py')
if filename is None: |
797664d9 |
diagnostic_messages.append('Unable to find the module file for {0}'.format(module_name)) |
c22dde61 |
continue
try:
write_metadata(filename, new_metadata, version, overwrite)
except ParseError as e:
diagnostic_messages.append(e.args[0])
continue
if diagnostic_messages:
pprint(diagnostic_messages)
return 0
def add_default(version=None, overwrite=False):
"""Implement the subcommand to add default metadata to modules
Add the default metadata to any plugin which lacks it.
:kwarg version: If given, the metadata must be at least this version.
Otherwise, treat the module as not having existing metadata.
:kwarg overwrite: If True, overwrite any existing metadata. Otherwise,
do not modify files which have metadata at an appropriate version
"""
# List of all plugins
plugins = module_loader.all(path_only=True)
plugins = ((os.path.splitext((os.path.basename(p)))[0], p) for p in plugins)
plugins = (p for p in plugins if p[0] not in NONMODULE_MODULE_NAMES)
# Iterate through each plugin
processed = set()
diagnostic_messages = []
for name, filename in (info for info in plugins if info[0] not in processed):
try: |
ab43a269 |
write_metadata(filename, DEFAULT_METADATA, version, overwrite) |
c22dde61 |
except ParseError as e:
diagnostic_messages.append(e.args[0])
continue
processed.add(name)
if diagnostic_messages:
pprint(diagnostic_messages)
return 0
|
eb1214ba |
def upgrade_metadata(version=None):
"""Implement the subcommand to upgrade the default metadata in modules.
:kwarg version: If given, the version of the metadata to upgrade to. If
not given, upgrade to the latest format version.
"""
if version is None:
# Number larger than any of the defined metadata formats.
version = 9999999
requested_version = StrictVersion(version)
# List all plugins
plugins = module_loader.all(path_only=True)
plugins = ((os.path.splitext((os.path.basename(p)))[0], p) for p in plugins)
plugins = (p for p in plugins if p[0] not in NONMODULE_MODULE_NAMES)
processed = set()
diagnostic_messages = []
for name, filename in (info for info in plugins if info[0] not in processed):
# For each plugin, read the existing metadata
with open(filename, 'rb') as f:
module_data = f.read() |
3ee997b7 |
metadata = extract_metadata(module_data=module_data, offsets=True)[0] |
eb1214ba |
# If the metadata isn't the requested version, convert it to the new
# version
if 'metadata_version' not in metadata or metadata['metadata_version'] != version:
#
# With each iteration of metadata, add a new conditional to
# upgrade from the previous version
#
if 'metadata_version' not in metadata:
# First version, pre-1.0 final metadata
metadata = convert_metadata_pre_1_0_to_1_0(metadata)
if metadata['metadata_version'] == '1.0' and StrictVersion('1.0') < requested_version: |
af2073d0 |
metadata = convert_metadata_1_0_to_1_1(metadata)
if metadata['metadata_version'] == '1.1' and StrictVersion('1.1') < requested_version:
# 1.1 version => XXX. We don't yet have anything beyond 1.1 |
eb1214ba |
# so there's nothing here
pass
# Replace the existing metadata with the new format
try:
write_metadata(filename, metadata, version, overwrite=True)
except ParseError as e:
diagnostic_messages.append(e.args[0])
continue
processed.add(name)
if diagnostic_messages:
pprint(diagnostic_messages)
return 0
|
c22dde61 |
def report(version=None):
"""Implement the report subcommand
Print out all the modules that have metadata and all the ones that do not.
:kwarg version: If given, the metadata must be at least this version.
Otherwise return it as not having metadata
"""
# List of all plugins
plugins = module_loader.all(path_only=True)
plugins = ((os.path.splitext((os.path.basename(p)))[0], p) for p in plugins) |
eb1214ba |
plugins = (p for p in plugins if p[0] not in NONMODULE_MODULE_NAMES) |
ab43a269 |
plugins = list(plugins) |
c22dde61 |
|
ab43a269 |
no_metadata, has_metadata, support, status = metadata_summary(plugins, version=version) |
c22dde61 |
print('== Has metadata ==')
pprint(sorted(has_metadata)) |
ab43a269 |
print('')
|
c22dde61 |
print('== Has no metadata ==')
pprint(sorted(no_metadata))
print('') |
ab43a269 |
print('== Supported by core ==')
pprint(sorted(support['core'])) |
af2073d0 |
print('== Supported by value certified ==')
pprint(sorted(support['certified']))
print('== Supported by value network ==')
pprint(sorted(support['network'])) |
ab43a269 |
print('== Supported by community ==')
pprint(sorted(support['community']))
print('')
print('== Status: stableinterface ==')
pprint(sorted(status['stableinterface']))
print('== Status: preview ==')
pprint(sorted(status['preview']))
print('== Status: deprecated ==')
pprint(sorted(status['deprecated']))
print('== Status: removed ==')
pprint(sorted(status['removed']))
print('')
print('== Summary ==')
print('No Metadata: {0} Has Metadata: {1}'.format(len(no_metadata), len(has_metadata))) |
af2073d0 |
print('Support level: core: {0} community: {1} certified: {2} network: {3}'.format(len(support['core']),
len(support['community']), len(support['certified']), len(support['network']))) |
5942de60 |
print('Status StableInterface: {0} Status Preview: {1} Status Deprecated: {2} Status Removed: {3}'.format(len(status['stableinterface']), |
9a5a61bd |
len(status['preview']), len(status['deprecated']), len(status['removed']))) |
c22dde61 |
return 0
if __name__ == '__main__':
action, args = parse_args(sys.argv[1:])
if action == 'report':
rc = report(version=args['version'])
elif action == 'add':
rc = add_from_csv(args['csvfile'], version=args['version'], overwrite=args['overwrite'])
elif action == 'add-default':
rc = add_default(version=args['version'], overwrite=args['overwrite']) |
eb1214ba |
elif action == 'upgrade':
rc = upgrade_metadata(version=args['version']) |
c22dde61 |
sys.exit(rc) |