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