diff options
Diffstat (limited to 'test/lib/ansible_test/_internal/provider/layout')
4 files changed, 446 insertions, 0 deletions
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, + ) |