ansible/test/lib/ansible_test/_internal/data.py

"""Context information for the current invocation of ansible-test."""
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type

import os

from . import types as t

from .util import (
    ApplicationError,
    import_plugins,
    ANSIBLE_ROOT,
    is_subdir,
    ANSIBLE_IS_INSTALLED,
)

from .provider import (
    find_path_provider,
    get_path_provider_classes,
    ProviderNotFoundForPath,
)

from .provider.source import (
    SourceProvider,
)

from .provider.source.unversioned import (
    UnversionedSource,
)

from .provider.layout import (
    ContentLayout,
    InstallLayout,
    LayoutProvider,
)


class UnexpectedSourceRoot(ApplicationError):
    """Exception generated when a source root is found below a layout root."""
    def __init__(self, source_root, layout_root):  # type: (str, str) -> None
        super(UnexpectedSourceRoot, self).__init__('Source root "%s" cannot be below layout root "%s".' % (source_root, layout_root))

        self.source_root = source_root
        self.layout_root = layout_root


class DataContext:
    """Data context providing details about the current execution environment for ansible-test."""
    def __init__(self):
        content_path = os.environ.get('ANSIBLE_TEST_CONTENT_ROOT')
        current_path = os.getcwd()

        self.__layout_providers = get_path_provider_classes(LayoutProvider)
        self.__source_providers = get_path_provider_classes(SourceProvider)
        self.payload_callbacks = []  # type: t.List[t.Callable[t.List[t.Tuple[str, str]], None]]

        if content_path:
            content = self.create_content_layout(self.__layout_providers, self.__source_providers, content_path, False)

            if content.is_ansible:
                install = InstallLayout(ANSIBLE_ROOT, content.all_files())
            else:
                install = None
        elif is_subdir(current_path, ANSIBLE_ROOT):
            content = self.create_content_layout(self.__layout_providers, self.__source_providers, ANSIBLE_ROOT, False)
            install = InstallLayout(ANSIBLE_ROOT, content.all_files())
        else:
            content = self.create_content_layout(self.__layout_providers, self.__source_providers, current_path, True)
            install = None

        self.__install = install  # type: t.Optional[InstallLayout]
        self.content = content  # type: ContentLayout

    @staticmethod
    def create_content_layout(layout_providers,  # type: t.List[t.Type[LayoutProvider]]
                              source_providers,  # type: t.List[t.Type[SourceProvider]]
                              root,  # type: str
                              walk,  # type: bool
                              ):  # type: (...) -> ContentLayout
        """Create a content layout using the given providers and root path."""
        layout_provider = find_path_provider(LayoutProvider, layout_providers, root, walk)

        try:
            source_provider = find_path_provider(SourceProvider, source_providers, root, walk)
        except ProviderNotFoundForPath:
            source_provider = UnversionedSource(layout_provider.root)

        if source_provider.root != layout_provider.root and is_subdir(source_provider.root, layout_provider.root):
            raise UnexpectedSourceRoot(source_provider.root, layout_provider.root)

        layout = layout_provider.create(layout_provider.root, source_provider.get_paths(layout_provider.root))

        return layout

    @staticmethod
    def create_install_layout(source_providers):  # type: (t.List[t.Type[SourceProvider]]) -> InstallLayout
        """Create an install layout using the given source provider."""
        try:
            source_provider = find_path_provider(SourceProvider, source_providers, ANSIBLE_ROOT, False)
        except ProviderNotFoundForPath:
            source_provider = UnversionedSource(ANSIBLE_ROOT)

        paths = source_provider.get_paths(ANSIBLE_ROOT)

        return InstallLayout(ANSIBLE_ROOT, paths)

    @property
    def install(self):  # type: () -> InstallLayout
        """Return the install context, loaded on demand."""
        if not self.__install:
            self.__install = self.create_install_layout(self.__source_providers)

        return self.__install

    def register_payload_callback(self, callback):  # type: (t.Callable[t.List[t.Tuple[str, str]], None]) -> None
        """Register the given payload callback."""
        self.payload_callbacks.append(callback)


def data_init():  # type: () -> DataContext
    """Initialize provider plugins."""
    provider_types = (
        'layout',
        'source',
    )

    for provider_type in provider_types:
        import_plugins('provider/%s' % provider_type)

    try:
        context = DataContext()
    except ProviderNotFoundForPath:
        options = [
            ' - an Ansible collection: {...}/ansible_collections/{namespace}/{collection}/',
        ]

        if not ANSIBLE_IS_INSTALLED:
            options.insert(0, ' - the Ansible source: %s/' % ANSIBLE_ROOT)

        raise ApplicationError('''The current working directory must be at or below:

%s

Current working directory: %s''' % ('\n'.join(options), os.getcwd()))

    return context


def data_context():  # type: () -> DataContext
    """Return the current data context."""
    try:
        return data_context.instance
    except AttributeError:
        data_context.instance = data_init()
        return data_context.instance