diff options
Diffstat (limited to 'test/lib/ansible_test/_internal/provider')
10 files changed, 747 insertions, 0 deletions
diff --git a/test/lib/ansible_test/_internal/provider/__init__.py b/test/lib/ansible_test/_internal/provider/__init__.py new file mode 100644 index 0000000..61d7baf --- /dev/null +++ b/test/lib/ansible_test/_internal/provider/__init__.py @@ -0,0 +1,72 @@ +"""Provider (plugin) infrastructure for ansible-test.""" +from __future__ import annotations + +import abc +import os +import typing as t + +from ..util import ( + ApplicationError, + get_subclasses, +) + + +def get_path_provider_classes(provider_type: t.Type[TPathProvider]) -> list[t.Type[TPathProvider]]: + """Return a list of path provider classes of the given type.""" + return sorted(get_subclasses(provider_type), key=lambda subclass: (subclass.priority, subclass.__name__)) + + +def find_path_provider(provider_type: t.Type[TPathProvider], + provider_classes: list[t.Type[TPathProvider]], + path: str, + walk: bool, + ) -> TPathProvider: + """Return the first found path provider of the given type for the given path.""" + sequences = sorted(set(pc.sequence for pc in provider_classes if pc.sequence > 0)) + + for sequence in sequences: + candidate_path = path + tier_classes = [pc for pc in provider_classes if pc.sequence == sequence] + + while True: + for provider_class in tier_classes: + if provider_class.is_content_root(candidate_path): + return provider_class(candidate_path) + + if not walk: + break + + parent_path = os.path.dirname(candidate_path) + + if parent_path == candidate_path: + break + + candidate_path = parent_path + + raise ProviderNotFoundForPath(provider_type, path) + + +class ProviderNotFoundForPath(ApplicationError): + """Exception generated when a path based provider cannot be found for a given path.""" + def __init__(self, provider_type: t.Type, path: str) -> None: + super().__init__('No %s found for path: %s' % (provider_type.__name__, path)) + + self.provider_type = provider_type + self.path = path + + +class PathProvider(metaclass=abc.ABCMeta): + """Base class for provider plugins that are path based.""" + sequence = 500 + priority = 500 + + def __init__(self, root: str) -> None: + self.root = root + + @staticmethod + @abc.abstractmethod + def is_content_root(path: str) -> bool: + """Return True if the given path is a content root for this provider.""" + + +TPathProvider = t.TypeVar('TPathProvider', bound=PathProvider) diff --git a/test/lib/ansible_test/_internal/provider/layout/__init__.py b/test/lib/ansible_test/_internal/provider/layout/__init__.py new file mode 100644 index 0000000..aa6693f --- /dev/null +++ b/test/lib/ansible_test/_internal/provider/layout/__init__.py @@ -0,0 +1,236 @@ +"""Code for finding content.""" +from __future__ import annotations + +import abc +import collections +import os +import typing as t + +from ...util import ( + ANSIBLE_SOURCE_ROOT, +) + +from .. import ( + PathProvider, +) + + +class Layout: + """Description of content locations and helper methods to access content.""" + def __init__(self, + root: str, + paths: list[str], + ) -> None: + self.root = root + + self.__paths = paths # contains both file paths and symlinked directory paths (ending with os.path.sep) + self.__files = [path for path in paths if not path.endswith(os.path.sep)] # contains only file paths + self.__paths_tree = paths_to_tree(self.__paths) + self.__files_tree = paths_to_tree(self.__files) + + def all_files(self, include_symlinked_directories: bool = False) -> list[str]: + """Return a list of all file paths.""" + if include_symlinked_directories: + return self.__paths + + return self.__files + + def walk_files(self, directory: str, include_symlinked_directories: bool = False) -> list[str]: + """Return a list of file paths found recursively under the given directory.""" + if include_symlinked_directories: + tree = self.__paths_tree + else: + tree = self.__files_tree + + parts = directory.rstrip(os.path.sep).split(os.path.sep) + item = get_tree_item(tree, parts) + + if not item: + return [] + + directories = collections.deque(item[0].values()) + + files = list(item[1]) + + while directories: + item = directories.pop() + directories.extend(item[0].values()) + files.extend(item[1]) + + return files + + def get_dirs(self, directory: str) -> list[str]: + """Return a list directory paths found directly under the given directory.""" + parts = directory.rstrip(os.path.sep).split(os.path.sep) + item = get_tree_item(self.__files_tree, parts) + return [os.path.join(directory, key) for key in item[0].keys()] if item else [] + + def get_files(self, directory: str) -> list[str]: + """Return a list of file paths found directly under the given directory.""" + parts = directory.rstrip(os.path.sep).split(os.path.sep) + item = get_tree_item(self.__files_tree, parts) + return item[1] if item else [] + + +class ContentLayout(Layout): + """Information about the current Ansible content being tested.""" + def __init__(self, + root: str, + paths: list[str], + plugin_paths: dict[str, str], + collection: t.Optional[CollectionDetail], + test_path: str, + results_path: str, + sanity_path: str, + sanity_messages: t.Optional[LayoutMessages], + integration_path: str, + integration_targets_path: str, + integration_vars_path: str, + integration_messages: t.Optional[LayoutMessages], + unit_path: str, + unit_module_path: str, + unit_module_utils_path: str, + unit_messages: t.Optional[LayoutMessages], + unsupported: bool = False, + ) -> None: + super().__init__(root, paths) + + self.plugin_paths = plugin_paths + self.collection = collection + self.test_path = test_path + self.results_path = results_path + self.sanity_path = sanity_path + self.sanity_messages = sanity_messages + self.integration_path = integration_path + self.integration_targets_path = integration_targets_path + self.integration_vars_path = integration_vars_path + self.integration_messages = integration_messages + self.unit_path = unit_path + self.unit_module_path = unit_module_path + self.unit_module_utils_path = unit_module_utils_path + self.unit_messages = unit_messages + self.unsupported = unsupported + + self.is_ansible = root == ANSIBLE_SOURCE_ROOT + + @property + def prefix(self) -> str: + """Return the collection prefix or an empty string if not a collection.""" + if self.collection: + return self.collection.prefix + + return '' + + @property + def module_path(self) -> t.Optional[str]: + """Return the path where modules are found, if any.""" + return self.plugin_paths.get('modules') + + @property + def module_utils_path(self) -> t.Optional[str]: + """Return the path where module_utils are found, if any.""" + return self.plugin_paths.get('module_utils') + + @property + def module_utils_powershell_path(self) -> t.Optional[str]: + """Return the path where powershell module_utils are found, if any.""" + if self.is_ansible: + return os.path.join(self.plugin_paths['module_utils'], 'powershell') + + return self.plugin_paths.get('module_utils') + + @property + def module_utils_csharp_path(self) -> t.Optional[str]: + """Return the path where csharp module_utils are found, if any.""" + if self.is_ansible: + return os.path.join(self.plugin_paths['module_utils'], 'csharp') + + return self.plugin_paths.get('module_utils') + + +class LayoutMessages: + """Messages generated during layout creation that should be deferred for later display.""" + def __init__(self) -> None: + self.info: list[str] = [] + self.warning: list[str] = [] + self.error: list[str] = [] + + +class CollectionDetail: + """Details about the layout of the current collection.""" + def __init__(self, + name: str, + namespace: str, + root: str, + ) -> None: + self.name = name + self.namespace = namespace + self.root = root + self.full_name = '%s.%s' % (namespace, name) + self.prefix = '%s.' % self.full_name + self.directory = os.path.join('ansible_collections', namespace, name) + + +class LayoutProvider(PathProvider): + """Base class for layout providers.""" + PLUGIN_TYPES = ( + 'action', + 'become', + 'cache', + 'callback', + 'cliconf', + 'connection', + 'doc_fragments', + 'filter', + 'httpapi', + 'inventory', + 'lookup', + 'module_utils', + 'modules', + 'netconf', + 'shell', + 'strategy', + 'terminal', + 'test', + 'vars', + # The following are plugin directories not directly supported by ansible-core, but used in collections + # (https://github.com/ansible-collections/overview/blob/main/collection_requirements.rst#modules--plugins) + 'plugin_utils', + 'sub_plugins', + ) + + @abc.abstractmethod + def create(self, root: str, paths: list[str]) -> ContentLayout: + """Create a layout using the given root and paths.""" + + +def paths_to_tree(paths: list[str]) -> tuple[dict[str, t.Any], list[str]]: + """Return a filesystem tree from the given list of paths.""" + tree: tuple[dict[str, t.Any], list[str]] = {}, [] + + for path in paths: + parts = path.split(os.path.sep) + root = tree + + for part in parts[:-1]: + if part not in root[0]: + root[0][part] = {}, [] + + root = root[0][part] + + root[1].append(path) + + return tree + + +def get_tree_item(tree: tuple[dict[str, t.Any], list[str]], parts: list[str]) -> t.Optional[tuple[dict[str, t.Any], list[str]]]: + """Return the portion of the tree found under the path given by parts, or None if it does not exist.""" + root = tree + + for part in parts: + root = root[0].get(part) + + if not root: + return None + + return root diff --git a/test/lib/ansible_test/_internal/provider/layout/ansible.py b/test/lib/ansible_test/_internal/provider/layout/ansible.py new file mode 100644 index 0000000..e8d0191 --- /dev/null +++ b/test/lib/ansible_test/_internal/provider/layout/ansible.py @@ -0,0 +1,44 @@ +"""Layout provider for Ansible source.""" +from __future__ import annotations + +import os + +from . import ( + ContentLayout, + LayoutProvider, +) + + +class AnsibleLayout(LayoutProvider): + """Layout provider for Ansible source.""" + @staticmethod + def is_content_root(path: str) -> bool: + """Return True if the given path is a content root for this provider.""" + return os.path.exists(os.path.join(path, 'setup.py')) and os.path.exists(os.path.join(path, 'bin/ansible-test')) + + def create(self, root: str, paths: list[str]) -> ContentLayout: + """Create a Layout using the given root and paths.""" + plugin_paths = dict((p, os.path.join('lib/ansible/plugins', p)) for p in self.PLUGIN_TYPES) + + plugin_paths.update(dict( + modules='lib/ansible/modules', + module_utils='lib/ansible/module_utils', + )) + + return ContentLayout(root, + paths, + plugin_paths=plugin_paths, + collection=None, + test_path='test', + results_path='test/results', + sanity_path='test/sanity', + sanity_messages=None, + integration_path='test/integration', + integration_targets_path='test/integration/targets', + integration_vars_path='test/integration/integration_config.yml', + integration_messages=None, + unit_path='test/units', + unit_module_path='test/units/modules', + unit_module_utils_path='test/units/module_utils', + unit_messages=None, + ) diff --git a/test/lib/ansible_test/_internal/provider/layout/collection.py b/test/lib/ansible_test/_internal/provider/layout/collection.py new file mode 100644 index 0000000..299d0bc --- /dev/null +++ b/test/lib/ansible_test/_internal/provider/layout/collection.py @@ -0,0 +1,126 @@ +"""Layout provider for Ansible collections.""" +from __future__ import annotations + +import os + +from . import ( + ContentLayout, + LayoutProvider, + CollectionDetail, + LayoutMessages, +) + +from ...util import ( + is_valid_identifier, +) + + +class CollectionLayout(LayoutProvider): + """Layout provider for Ansible collections.""" + @staticmethod + def is_content_root(path: str) -> bool: + """Return True if the given path is a content root for this provider.""" + if os.path.basename(os.path.dirname(os.path.dirname(path))) == 'ansible_collections': + return True + + return False + + def create(self, root: str, paths: list[str]) -> ContentLayout: + """Create a Layout using the given root and paths.""" + plugin_paths = dict((p, os.path.join('plugins', p)) for p in self.PLUGIN_TYPES) + + collection_root = os.path.dirname(os.path.dirname(root)) + collection_dir = os.path.relpath(root, collection_root) + + collection_namespace: str + collection_name: str + + collection_namespace, collection_name = collection_dir.split(os.path.sep) + + collection_root = os.path.dirname(collection_root) + + sanity_messages = LayoutMessages() + integration_messages = LayoutMessages() + unit_messages = LayoutMessages() + + # these apply to all test commands + self.__check_test_path(paths, sanity_messages) + self.__check_test_path(paths, integration_messages) + self.__check_test_path(paths, unit_messages) + + # these apply to specific test commands + integration_targets_path = self.__check_integration_path(paths, integration_messages) + self.__check_unit_path(paths, unit_messages) + + return ContentLayout(root, + paths, + plugin_paths=plugin_paths, + collection=CollectionDetail( + name=collection_name, + namespace=collection_namespace, + root=collection_root, + ), + test_path='tests', + results_path='tests/output', + sanity_path='tests/sanity', + sanity_messages=sanity_messages, + integration_path='tests/integration', + integration_targets_path=integration_targets_path.rstrip(os.path.sep), + integration_vars_path='tests/integration/integration_config.yml', + integration_messages=integration_messages, + unit_path='tests/unit', + unit_module_path='tests/unit/plugins/modules', + unit_module_utils_path='tests/unit/plugins/module_utils', + unit_messages=unit_messages, + unsupported=not (is_valid_identifier(collection_namespace) and is_valid_identifier(collection_name)), + ) + + @staticmethod + def __check_test_path(paths: list[str], messages: LayoutMessages) -> None: + modern_test_path = 'tests/' + modern_test_path_found = any(path.startswith(modern_test_path) for path in paths) + legacy_test_path = 'test/' + legacy_test_path_found = any(path.startswith(legacy_test_path) for path in paths) + + if modern_test_path_found and legacy_test_path_found: + messages.warning.append('Ignoring tests in "%s" in favor of "%s".' % (legacy_test_path, modern_test_path)) + elif legacy_test_path_found: + messages.warning.append('Ignoring tests in "%s" that should be in "%s".' % (legacy_test_path, modern_test_path)) + + @staticmethod + def __check_integration_path(paths: list[str], messages: LayoutMessages) -> str: + modern_integration_path = 'roles/test/' + modern_integration_path_found = any(path.startswith(modern_integration_path) for path in paths) + legacy_integration_path = 'tests/integration/targets/' + legacy_integration_path_found = any(path.startswith(legacy_integration_path) for path in paths) + + if modern_integration_path_found and legacy_integration_path_found: + messages.warning.append('Ignoring tests in "%s" in favor of "%s".' % (legacy_integration_path, modern_integration_path)) + integration_targets_path = modern_integration_path + elif legacy_integration_path_found: + messages.info.append('Falling back to tests in "%s" because "%s" was not found.' % (legacy_integration_path, modern_integration_path)) + integration_targets_path = legacy_integration_path + elif modern_integration_path_found: + messages.info.append('Loading tests from "%s".' % modern_integration_path) + integration_targets_path = modern_integration_path + else: + messages.error.append('Cannot run integration tests without "%s" or "%s".' % (modern_integration_path, legacy_integration_path)) + integration_targets_path = modern_integration_path + + return integration_targets_path + + @staticmethod + def __check_unit_path(paths: list[str], messages: LayoutMessages) -> None: + modern_unit_path = 'tests/unit/' + modern_unit_path_found = any(path.startswith(modern_unit_path) for path in paths) + legacy_unit_path = 'tests/units/' # test/units/ will be covered by the warnings for test/ vs tests/ + legacy_unit_path_found = any(path.startswith(legacy_unit_path) for path in paths) + + if modern_unit_path_found and legacy_unit_path_found: + messages.warning.append('Ignoring tests in "%s" in favor of "%s".' % (legacy_unit_path, modern_unit_path)) + elif legacy_unit_path_found: + messages.warning.append('Rename "%s" to "%s" to run unit tests.' % (legacy_unit_path, modern_unit_path)) + elif modern_unit_path_found: + pass # unit tests only run from one directory so no message is needed + else: + messages.error.append('Cannot run unit tests without "%s".' % modern_unit_path) diff --git a/test/lib/ansible_test/_internal/provider/layout/unsupported.py b/test/lib/ansible_test/_internal/provider/layout/unsupported.py new file mode 100644 index 0000000..16aa254 --- /dev/null +++ b/test/lib/ansible_test/_internal/provider/layout/unsupported.py @@ -0,0 +1,40 @@ +"""Layout provider for an unsupported directory layout.""" +from __future__ import annotations + +from . import ( + ContentLayout, + LayoutProvider, +) + + +class UnsupportedLayout(LayoutProvider): + """Layout provider for an unsupported directory layout.""" + sequence = 0 # disable automatic detection + + @staticmethod + def is_content_root(path: str) -> bool: + """Return True if the given path is a content root for this provider.""" + return False + + def create(self, root: str, paths: list[str]) -> ContentLayout: + """Create a Layout using the given root and paths.""" + plugin_paths = dict((p, p) for p in self.PLUGIN_TYPES) + + return ContentLayout(root, + paths, + plugin_paths=plugin_paths, + collection=None, + test_path='', + results_path='', + sanity_path='', + sanity_messages=None, + integration_path='', + integration_targets_path='', + integration_vars_path='', + integration_messages=None, + unit_path='', + unit_module_path='', + unit_module_utils_path='', + unit_messages=None, + unsupported=True, + ) diff --git a/test/lib/ansible_test/_internal/provider/source/__init__.py b/test/lib/ansible_test/_internal/provider/source/__init__.py new file mode 100644 index 0000000..aa8ca47 --- /dev/null +++ b/test/lib/ansible_test/_internal/provider/source/__init__.py @@ -0,0 +1,15 @@ +"""Common code for source providers.""" +from __future__ import annotations + +import abc + +from .. import ( + PathProvider, +) + + +class SourceProvider(PathProvider): + """Base class for source providers.""" + @abc.abstractmethod + def get_paths(self, path: str) -> list[str]: + """Return the list of available content paths under the given path.""" diff --git a/test/lib/ansible_test/_internal/provider/source/git.py b/test/lib/ansible_test/_internal/provider/source/git.py new file mode 100644 index 0000000..37f16bf --- /dev/null +++ b/test/lib/ansible_test/_internal/provider/source/git.py @@ -0,0 +1,69 @@ +"""Source provider for a content root managed by git version control.""" +from __future__ import annotations + +import os + +from ...git import ( + Git, +) + +from ...encoding import ( + to_bytes, +) + +from ...util import ( + SubprocessError, +) + +from . import ( + SourceProvider, +) + + +class GitSource(SourceProvider): + """Source provider for a content root managed by git version control.""" + @staticmethod + def is_content_root(path: str) -> bool: + """Return True if the given path is a content root for this provider.""" + return os.path.exists(os.path.join(path, '.git')) + + def get_paths(self, path: str) -> list[str]: + """Return the list of available content paths under the given path.""" + paths = self.__get_paths(path) + + try: + submodule_paths = Git(path).get_submodule_paths() + except SubprocessError: + if path == self.root: + raise + + # older versions of git require submodule commands to be executed from the top level of the working tree + # git version 2.18.1 (centos8) does not have this restriction + # git version 1.8.3.1 (centos7) does + # fall back to using the top level directory of the working tree only when needed + # this avoids penalizing newer git versions with a potentially slower analysis due to additional submodules + rel_path = os.path.relpath(path, self.root) + os.path.sep + + submodule_paths = Git(self.root).get_submodule_paths() + submodule_paths = [os.path.relpath(p, rel_path) for p in submodule_paths if p.startswith(rel_path)] + + for submodule_path in submodule_paths: + paths.extend(os.path.join(submodule_path, p) for p in self.__get_paths(os.path.join(path, submodule_path))) + + # git reports submodule directories as regular files + paths = [p for p in paths if p not in submodule_paths] + + return paths + + @staticmethod + def __get_paths(path: str) -> list[str]: + """Return the list of available content paths under the given path.""" + git = Git(path) + paths = git.get_file_names(['--cached', '--others', '--exclude-standard']) + deleted_paths = git.get_file_names(['--deleted']) + paths = sorted(set(paths) - set(deleted_paths)) + + # directory symlinks are reported by git as regular files but they need to be treated as directories + paths = [path + os.path.sep if os.path.isdir(to_bytes(path)) else path for path in paths] + + return paths diff --git a/test/lib/ansible_test/_internal/provider/source/installed.py b/test/lib/ansible_test/_internal/provider/source/installed.py new file mode 100644 index 0000000..6b82188 --- /dev/null +++ b/test/lib/ansible_test/_internal/provider/source/installed.py @@ -0,0 +1,40 @@ +"""Source provider for content which has been installed.""" +from __future__ import annotations + +import os + +from . import ( + SourceProvider, +) + + +class InstalledSource(SourceProvider): + """Source provider for content which has been installed.""" + sequence = 0 # disable automatic detection + + @staticmethod + def is_content_root(path: str) -> bool: + """Return True if the given path is a content root for this provider.""" + return False + + def get_paths(self, path: str) -> list[str]: + """Return the list of available content paths under the given path.""" + paths = [] + + kill_extensions = ( + '.pyc', + '.pyo', + ) + + for root, _dummy, file_names in os.walk(path): + rel_root = os.path.relpath(root, path) + + if rel_root == '.': + rel_root = '' + + paths.extend([os.path.join(rel_root, file_name) for file_name in file_names + if not os.path.splitext(file_name)[1] in kill_extensions]) + + # NOTE: directory symlinks are ignored as there should be no directory symlinks for an install + + return paths diff --git a/test/lib/ansible_test/_internal/provider/source/unsupported.py b/test/lib/ansible_test/_internal/provider/source/unsupported.py new file mode 100644 index 0000000..e2f8953 --- /dev/null +++ b/test/lib/ansible_test/_internal/provider/source/unsupported.py @@ -0,0 +1,20 @@ +"""Source provider to use when the layout is unsupported.""" +from __future__ import annotations + +from . import ( + SourceProvider, +) + + +class UnsupportedSource(SourceProvider): + """Source provider to use when the layout is unsupported.""" + sequence = 0 # disable automatic detection + + @staticmethod + def is_content_root(path: str) -> bool: + """Return True if the given path is a content root for this provider.""" + return False + + def get_paths(self, path: str) -> list[str]: + """Return the list of available content paths under the given path.""" + return [] diff --git a/test/lib/ansible_test/_internal/provider/source/unversioned.py b/test/lib/ansible_test/_internal/provider/source/unversioned.py new file mode 100644 index 0000000..d8eff5d --- /dev/null +++ b/test/lib/ansible_test/_internal/provider/source/unversioned.py @@ -0,0 +1,85 @@ +"""Fallback source provider when no other provider matches the content root.""" +from __future__ import annotations + +import os + +from ...constants import ( + TIMEOUT_PATH, +) + +from ...encoding import ( + to_bytes, +) + +from . import ( + SourceProvider, +) + + +class UnversionedSource(SourceProvider): + """Fallback source provider when no other provider matches the content root.""" + sequence = 0 # disable automatic detection + + @staticmethod + def is_content_root(path: str) -> bool: + """Return True if the given path is a content root for this provider.""" + return False + + def get_paths(self, path: str) -> list[str]: + """Return the list of available content paths under the given path.""" + paths = [] + + kill_any_dir = ( + '.idea', + '.pytest_cache', + '__pycache__', + 'ansible.egg-info', + 'ansible_base.egg-info', + 'ansible_core.egg-info', + ) + + kill_sub_dir = { + 'test': ( + 'results', + 'cache', + 'output', + ), + 'tests': ( + 'output', + ), + 'docs/docsite': ( + '_build', + ), + } + + kill_sub_file = { + '': ( + TIMEOUT_PATH, + ), + } + + kill_extensions = ( + '.pyc', + '.pyo', + '.retry', + ) + + for root, dir_names, file_names in os.walk(path): + rel_root = os.path.relpath(root, path) + + if rel_root == '.': + rel_root = '' + + for kill in kill_any_dir + kill_sub_dir.get(rel_root, ()): + if kill in dir_names: + dir_names.remove(kill) + + kill_files = kill_sub_file.get(rel_root, ()) + + paths.extend([os.path.join(rel_root, file_name) for file_name in file_names + if not os.path.splitext(file_name)[1] in kill_extensions and file_name not in kill_files]) + + # include directory symlinks since they will not be traversed and would otherwise go undetected + paths.extend([os.path.join(rel_root, dir_name) + os.path.sep for dir_name in dir_names if os.path.islink(to_bytes(dir_name))]) + + return paths |