mirror of
https://github.com/freeipa/ansible-freeipa.git
synced 2026-03-26 21:33:05 +00:00
24efad73fa
Recent pylint versions warn against the use of an 'else' in a 'try-except' block if using a 'return' on the 'except' part is is the idom used by ansible-freeipa when retrieving IPA data objects. This change removes the usage of the 'else:' in such cases, and modify the templates so that new modules do not have the same issue in the future.
209 lines
7.6 KiB
Python
209 lines
7.6 KiB
Python
"""Filter tests based on plugin modifications."""
|
|
|
|
import sys
|
|
import os
|
|
from importlib.machinery import SourceFileLoader
|
|
import types
|
|
from unittest import mock
|
|
import yaml
|
|
|
|
|
|
PYTHON_IMPORT = __import__
|
|
|
|
|
|
def get_plugins_from_playbook(playbook):
|
|
"""Get all plugins called in the given playbook."""
|
|
def get_tasks(task_block):
|
|
"""
|
|
Get all plugins used on tasks.
|
|
|
|
Recursively process "block", "include_tasks" and "import_tasks".
|
|
"""
|
|
_result = set()
|
|
for tasks in task_block:
|
|
for task in tasks:
|
|
original_task = task
|
|
if "." in task:
|
|
task = task.split(".")[-1]
|
|
if task == "block":
|
|
_result.update(get_tasks(tasks["block"]))
|
|
elif task in ["include_tasks", "import_tasks"
|
|
"ansible.builtin.include_tasks",
|
|
"ansible.builtin.import_tasks"]:
|
|
parent = os.path.dirname(playbook)
|
|
include_task = tasks[task]
|
|
if isinstance(include_task, dict):
|
|
include_file = os.path.join(
|
|
parent, include_task["file"]
|
|
)
|
|
else:
|
|
include_file = os.path.join(parent, include_task)
|
|
_result.update(get_plugins_from_playbook(include_file))
|
|
elif task in ["include_role",
|
|
"ansible.builtin.include_role"]:
|
|
_result.add(f"_{tasks[original_task]['name']}")
|
|
elif task.startswith("ipa"):
|
|
# assume we are only interested in 'ipa*' modules/roles
|
|
_result.add(task)
|
|
elif task == "role":
|
|
# not really a "task", but we'll handle the same way.
|
|
_result.add(f"_{tasks[task]}")
|
|
return _result
|
|
|
|
def load_playbook(filename):
|
|
"""Load playbook file using Python's YAML parser."""
|
|
if not (filename.endswith("yml") or filename.endswith("yaml")):
|
|
return []
|
|
# print("Processing:", playbook)
|
|
try:
|
|
with open(filename, "rt") as playbook_file:
|
|
data = yaml.safe_load(playbook_file)
|
|
except yaml.scanner.ScannerError: # If not a YAML/JSON file.
|
|
return []
|
|
except yaml.parser.ParserError: # If not a YAML/JSON file.
|
|
return []
|
|
return data if data else []
|
|
|
|
data = load_playbook(playbook)
|
|
task_blocks = [t.get("tasks", []) if "tasks" in t else [] for t in data]
|
|
role_blocks = [t.get("roles", []) if "roles" in t else [] for t in data]
|
|
# assume file is a list of tasks if no "tasks" entry found.
|
|
if not task_blocks:
|
|
task_blocks = [data]
|
|
_result = set()
|
|
for task_block in task_blocks:
|
|
_result.update(get_tasks(task_block))
|
|
# roles
|
|
for role_block in role_blocks:
|
|
_result.update(get_tasks(role_block))
|
|
|
|
return _result
|
|
|
|
|
|
def import_mock(name, *args):
|
|
"""Intercept 'import' calls and store module name."""
|
|
if not hasattr(import_mock, "call_list"):
|
|
setattr(import_mock, "call_list", set()) # noqa: B010
|
|
import_mock.call_list.add(name) # pylint: disable=no-member
|
|
try:
|
|
# print("NAME:", name)
|
|
return PYTHON_IMPORT(name, *args)
|
|
except ModuleNotFoundError:
|
|
# We're not really interested in loading the module
|
|
# if it can't be imported, it is not something we really care.
|
|
return mock.Mock()
|
|
except Exception: # pylint: disable=broad-except
|
|
print(
|
|
"An unexpected error occured. Do you have all requirements set?",
|
|
file=sys.stderr
|
|
)
|
|
sys.exit(1)
|
|
|
|
|
|
def parse_playbooks(test_module):
|
|
"""Load all playbooks for 'test_module' directory."""
|
|
if test_module.name[0] in [".", "_"] or test_module.name == "pytests":
|
|
return []
|
|
_files = set()
|
|
for arg in os.scandir(test_module):
|
|
if arg.is_dir():
|
|
_files.update(parse_playbooks(arg))
|
|
else:
|
|
for playbook in get_plugins_from_playbook(arg.path):
|
|
if playbook.startswith("_"):
|
|
source = f"roles/{playbook[1:]}"
|
|
if os.path.isdir(source):
|
|
_files.add(source)
|
|
else:
|
|
source = f"plugins/modules/{playbook}.py"
|
|
if os.path.isfile(source):
|
|
_files.add(source)
|
|
# If a plugin imports a module from the repository,
|
|
# we'l find it by patching the builtin __import__
|
|
# function and importing the module from the source
|
|
# file. The modules imported as a result of the import
|
|
# will be added to the import_mock.call_list list.
|
|
with mock.patch(
|
|
"builtins.__import__", side_effect=import_mock
|
|
):
|
|
# pylint: disable=no-value-for-parameter
|
|
loader = SourceFileLoader(playbook, source)
|
|
loader.exec_module(types.ModuleType(loader.name))
|
|
# pylint: disable=no-member
|
|
candidates = [
|
|
f.split(".")[1:]
|
|
for f in import_mock.call_list
|
|
if f.startswith("ansible.")
|
|
]
|
|
# pylint: enable=no-member
|
|
files = [
|
|
"plugins/" + "/".join(f) + ".py"
|
|
for f in candidates
|
|
]
|
|
_files.update([f for f in files if os.path.isfile(f)])
|
|
else:
|
|
source = f"roles/{playbook}"
|
|
if os.path.isdir(source):
|
|
_files.add(source)
|
|
|
|
return _files
|
|
|
|
|
|
def map_test_module_sources(base):
|
|
"""Create a map of 'test-modules' to 'plugin-sources', from 'base'."""
|
|
# Find root directory of playbook tests.
|
|
script_dir = os.path.dirname(__file__)
|
|
test_root = os.path.realpath(os.path.join(script_dir, f"../{base}"))
|
|
# create modules:source_files map
|
|
_result = {}
|
|
for test_module in [d for d in os.scandir(test_root) if d.is_dir()]:
|
|
_depends_on = parse_playbooks(test_module)
|
|
if _depends_on:
|
|
_result[test_module.name] = _depends_on
|
|
return _result
|
|
|
|
|
|
def usage(err=0):
|
|
print("filter_plugins.py [-h|--help] [-p|--pytest] PY_SRC...")
|
|
print(
|
|
"""
|
|
Print a comma-separated list of modules that should be tested if
|
|
PY_SRC is modified.
|
|
|
|
Options:
|
|
|
|
-h, --help Print this message and exit.
|
|
-p, --pytest Evaluate pytest tests (playbooks only).
|
|
"""
|
|
)
|
|
sys.exit(err)
|
|
|
|
|
|
def main():
|
|
"""Program entry point."""
|
|
if "-h" in sys.argv or "--help" in sys.argv:
|
|
usage()
|
|
_base = "tests"
|
|
if "-p" in sys.argv or "--pytest" in sys.argv:
|
|
_base = "tests/pytests"
|
|
call_args = [x for x in sys.argv[1:] if x not in ["-p", "--pytest"]]
|
|
_mapping = map_test_module_sources(_base)
|
|
_test_suits = (
|
|
[
|
|
_module for _module, _files in _mapping.items()
|
|
for _arg in call_args
|
|
for _file in _files
|
|
if _file.startswith(_arg)
|
|
] + [
|
|
_role for _role in [x for x in _mapping if x.endswith("_role")]
|
|
for _arg in call_args
|
|
if _arg.startswith("roles/ipa" + _role[:-5])
|
|
]
|
|
)
|
|
if _test_suits:
|
|
print(",".join(sorted(_test_suits)))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|