1
0
Fork 0
mirror of https://github.com/ansible-collections/community.general.git synced 2024-09-14 20:13:21 +02:00
community.general/ansible_testing/modules.py

396 lines
13 KiB
Python
Raw Normal View History

2015-05-18 18:00:34 +02:00
#!/usr/bin/env python
from __future__ import print_function
import os
2015-06-30 20:51:01 +02:00
import re
import abc
2015-05-18 18:00:34 +02:00
import ast
import sys
2015-05-18 18:00:34 +02:00
import argparse
from fnmatch import fnmatch
2015-05-18 18:00:34 +02:00
2015-06-30 20:51:18 +02:00
from ansible.executor.module_common import REPLACER_WINDOWS
2015-05-18 18:00:34 +02:00
from ansible.utils.module_docs import get_docstring, BLACKLIST_MODULES
BLACKLIST_DIRS = frozenset(('.git',))
2015-06-30 20:51:01 +02:00
INDENT_REGEX = re.compile(r'(^[ \t]*)', flags=re.M)
2015-05-18 21:11:15 +02:00
class Validator(object):
"""Validator instances are intended to be run on a single object. if you
are scanning multiple objects for problems, you'll want to have a separate
Validator for each one."""
__metaclass__ = abc.ABCMeta
def __init__(self):
self.reset()
def reset(self):
"""Reset the test results"""
self.errors = []
self.warnings = []
@abc.abstractproperty
def object_name(self):
"""Name of the object we validated"""
pass
@abc.abstractmethod
def validate(self, reset=True):
"""Run this method to generate the test results"""
if reset:
self.reset()
def report(self, warnings=False):
"""Print out the test results"""
if self.errors or (warnings and self.warnings):
print('=' * 76)
print(self.object_path)
print('=' * 76)
ret = []
for error in self.errors:
print('ERROR: %s' % error)
ret.append(1)
if warnings:
for warning in self.warnings:
print('WARNING: %s' % warning)
ret.append(1)
if self.errors or (warnings and self.warnings):
print()
return len(ret)
class ModuleValidator(Validator):
2015-06-30 20:50:01 +02:00
BLACKLIST_PATTERNS = ('.git*', '*.pyc', '*.pyo', '.*', '*.md')
2015-05-18 21:11:15 +02:00
BLACKLIST_FILES = frozenset(('.git', '.gitignore', '.travis.yml',
'.gitattributes', '.gitmodules', 'COPYING',
2015-06-30 20:50:01 +02:00
'__init__.py', 'VERSION', 'test-docs.sh'))
BLACKLIST = BLACKLIST_FILES.union(BLACKLIST_MODULES)
BOTTOM_IMPORTS = frozenset((
'ansible.module_utils.basic',
'ansible.module_utils.urls',
'ansible.module_utils.facts',
'ansible.module_utils.splitter',
'ansible.module_utils.known_hosts',
))
BOTTOM_IMPORTS_BLACKLIST = frozenset((
'command.py',
))
PS_DOC_BLACKLIST = frozenset((
'slurp.ps1',
'setup.ps1'
))
def __init__(self, path, root=None):
super(ModuleValidator, self).__init__()
self._root = root
2015-05-18 18:00:34 +02:00
self.path = path
self.basename = os.path.basename(self.path)
self.name, _ = os.path.splitext(self.basename)
with open(path) as f:
2015-05-18 22:30:58 +02:00
self.text = f.read()
self.length = len(self.text.splitlines())
try:
2015-05-18 22:30:58 +02:00
self.ast = ast.parse(self.text)
except:
self.ast = None
@property
def object_name(self):
return self.basename
@property
def object_path(self):
if self._root:
return self.path.replace(self._root, '').lstrip('/')
return self.object_name
def _python_module(self):
if self.path.endswith('.py'):
return True
return False
def _powershell_module(self):
if self.path.endswith('.ps1'):
return True
return False
2015-05-18 18:00:34 +02:00
def _just_docs(self):
try:
for child in self.ast.body:
if not isinstance(child, ast.Assign):
return False
return True
except AttributeError:
return False
2015-05-18 18:00:34 +02:00
def _is_bottom_import_blacklisted(self):
return self.object_name in self.BOTTOM_IMPORTS_BLACKLIST
def _check_interpreter(self, powershell=False):
if powershell:
if not self.text.startswith('#!powershell\n'):
self.errors.append('Interpreter line is not "#!powershell"')
return
2015-05-18 22:30:58 +02:00
if not self.text.startswith('#!/usr/bin/python'):
self.errors.append('Interpreter line is not "#!/usr/bin/python"')
2015-05-19 18:25:11 +02:00
def _check_for_sys_exit(self):
if 'sys.exit(' in self.text:
self.errors.append('sys.exit() call found')
def _check_for_gpl3_header(self):
if ('GNU General Public License' not in self.text and
'version 3' not in self.text):
self.errors.append('GPLv3 license header not found')
2015-06-30 20:51:01 +02:00
def _check_for_tabs(self):
indent = INDENT_REGEX.findall(self.text)
for i in indent:
if '\t' in i:
self.errors.append('indentation contains tabs')
break
2015-05-18 22:41:16 +02:00
def _find_json_import(self):
for child in self.ast.body:
if isinstance(child, ast.Import):
for name in child.names:
if name.name == 'json':
self.warnings.append('JSON import found, '
'already provided by '
'ansible.module_utils.basic')
def _find_module_utils(self, main):
2015-05-18 18:00:34 +02:00
linenos = []
for child in self.ast.body:
found_module_utils_import = False
if isinstance(child, ast.ImportFrom):
if child.module.startswith('ansible.module_utils.'):
found_module_utils_import = True
if child.module in self.BOTTOM_IMPORTS:
if (child.lineno < main - 10 and
not self._is_bottom_import_blacklisted()):
self.errors.append('%s import not near main()' %
child.module)
else:
self.warnings.append('%s import not near main()' %
child.module)
2015-05-18 18:00:34 +02:00
linenos.append(child.lineno)
if not child.names:
self.errors.append('%s: not a "from" import"' %
child.module)
found_alias = False
for name in child.names:
if isinstance(name, ast.alias):
found_alias = True
if name.asname or name.name != '*':
self.errors.append('%s: did not import "*"' %
child.module)
if found_module_utils_import and not found_alias:
self.errors.append('%s: did not import "*"' % child.module)
if not linenos:
self.errors.append('Did not find a module_utils import')
return linenos
def _find_main_call(self):
lineno = False
if_bodies = []
for child in self.ast.body:
if isinstance(child, ast.If):
try:
if child.test.left.id == '__name__':
if_bodies.extend(child.body)
except AttributeError:
pass
bodies = self.ast.body
bodies.extend(if_bodies)
for child in bodies:
if isinstance(child, ast.Expr):
if isinstance(child.value, ast.Call):
if (isinstance(child.value.func, ast.Name) and
child.value.func.id == 'main'):
lineno = child.lineno
if lineno < self.length - 1:
self.errors.append('Call to main() not the last '
2015-05-18 18:00:34 +02:00
'line')
if not lineno:
self.errors.append('Did not find a call to main')
return lineno or 0
def _find_has_import(self):
for child in self.ast.body:
found_try_except_import = False
found_has = False
if isinstance(child, ast.TryExcept):
bodies = child.body
2015-05-18 18:59:53 +02:00
for handler in child.handlers:
bodies.extend(handler.body)
2015-05-18 18:00:34 +02:00
for grandchild in bodies:
if isinstance(grandchild, ast.Import):
found_try_except_import = True
2015-05-18 18:59:53 +02:00
if isinstance(grandchild, ast.Assign):
2015-05-18 18:00:34 +02:00
for target in grandchild.targets:
if target.id.lower().startswith('has_'):
found_has = True
if found_try_except_import and not found_has:
self.warnings.append('Found Try/Except block without HAS_ '
'assginment')
def _find_ps_replacers(self):
if 'WANT_JSON' not in self.text:
self.errors.append('WANT_JSON not found in module')
if REPLACER_WINDOWS not in self.text:
self.errors.append('"%s" not found in module' % REPLACER_WINDOWS)
def _find_ps_docs_py_file(self):
if self.object_name in self.PS_DOC_BLACKLIST:
return
py_path = self.path.replace('.ps1', '.py')
if not os.path.isfile(py_path):
self.errors.append('Missing python documentation file')
2015-05-18 18:00:34 +02:00
def validate(self):
super(ModuleValidator, self).validate()
# Blacklists -- these files are not checked
2015-05-18 21:11:15 +02:00
if not frozenset((self.basename,
self.name)).isdisjoint(self.BLACKLIST):
return
for pat in self.BLACKLIST_PATTERNS:
if fnmatch(self.basename, pat):
return
# if self._powershell_module():
# self.warnings.append('Cannot check powershell modules at this '
# 'time. Skipping')
# return
if not self._python_module() and not self._powershell_module():
2015-05-18 21:11:15 +02:00
self.errors.append('Official Ansible modules must have a .py '
'extension for python modules or a .ps1 '
'for powershell modules')
return
if self._python_module() and self.ast is None:
self.errors.append('Python SyntaxError while parsing module')
2015-05-18 18:00:34 +02:00
return
if self._python_module():
doc, examples, ret = get_docstring(self.path)
if not bool(doc):
self.errors.append('Invalid or no DOCUMENTATION provided')
if not bool(examples):
self.errors.append('No EXAMPLES provided')
if not bool(ret):
self.warnings.append('No RETURN provided')
2015-05-18 18:00:34 +02:00
if self._python_module() and not self._just_docs():
2015-05-19 18:25:11 +02:00
self._check_for_sys_exit()
2015-05-18 22:41:16 +02:00
self._find_json_import()
2015-05-18 18:00:34 +02:00
main = self._find_main_call()
2015-05-19 20:35:50 +02:00
self._find_module_utils(main)
2015-05-18 18:00:34 +02:00
self._find_has_import()
2015-06-30 20:51:01 +02:00
self._check_for_tabs()
2015-05-18 18:00:34 +02:00
if self._powershell_module():
self._find_ps_replacers()
self._find_ps_docs_py_file()
self._check_for_gpl3_header()
if not self._just_docs():
self._check_interpreter(powershell=self._powershell_module())
2015-05-18 18:00:34 +02:00
class PythonPackageValidator(Validator):
def __init__(self, path, root=None):
super(PythonPackageValidator, self).__init__()
2015-05-18 18:00:34 +02:00
self._root = root
self.path = path
self.basename = os.path.basename(path)
@property
def object_name(self):
return self.basename
@property
def object_path(self):
if self._root:
return self.path.replace(self._root, '').lstrip('/')
return self.object_name
def validate(self):
super(PythonPackageValidator, self).validate()
init_file = os.path.join(self.path, '__init__.py')
if not os.path.exists(init_file):
2015-05-18 21:11:15 +02:00
self.errors.append('Ansible module subdirectories must contain an '
'__init__.py')
2015-05-18 18:00:34 +02:00
def main():
parser = argparse.ArgumentParser()
2015-05-19 20:36:03 +02:00
parser.add_argument('modules', help='Path to module or module directory')
2015-05-18 18:00:34 +02:00
parser.add_argument('-w', '--warnings', help='Show warnings',
action='store_true')
args = parser.parse_args()
args.modules = os.path.abspath(args.modules.rstrip('/'))
exit = []
2015-05-19 20:36:03 +02:00
# Allow testing against a single file
if os.path.isfile(args.modules):
mv = ModuleValidator(os.path.abspath(args.modules), root=args.modules)
2015-05-19 20:36:03 +02:00
mv.validate()
exit.append(mv.report(args.warnings))
sys.exit(sum(exit))
2015-05-18 18:00:34 +02:00
for root, dirs, files in os.walk(args.modules):
basedir = root[len(args.modules)+1:].split('/', 1)[0]
if basedir in BLACKLIST_DIRS:
continue
for dirname in dirs:
if root == args.modules and dirname in BLACKLIST_DIRS:
continue
path = os.path.join(root, dirname)
pv = PythonPackageValidator(os.path.abspath(path),
root=args.modules)
pv.validate()
exit.append(pv.report(args.warnings))
2015-05-18 18:00:34 +02:00
for filename in files:
path = os.path.join(root, filename)
mv = ModuleValidator(os.path.abspath(path), root=args.modules)
2015-05-18 18:00:34 +02:00
mv.validate()
exit.append(mv.report(args.warnings))
sys.exit(sum(exit))
2015-05-18 18:00:34 +02:00
if __name__ == '__main__':
main()