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

"""Configuration for the test hosts requested by the user."""

from __future__ import annotations

import abc
import dataclasses
import enum
import os
import pickle
import sys
import typing as t

from .constants import (
    SUPPORTED_PYTHON_VERSIONS,
    SUPPORTED_POWERSHELL_VERSIONS,
)

from .io import (
    open_binary_file,
)

from .completion import (
    AuditMode,
    CGroupVersion,
    CompletionConfig,
    docker_completion,
    DockerCompletionConfig,
    InventoryCompletionConfig,
    network_completion,
    NetworkRemoteCompletionConfig,
    PosixCompletionConfig,
    PosixRemoteCompletionConfig,
    PosixSshCompletionConfig,
    remote_completion,
    RemoteCompletionConfig,
    windows_completion,
    WindowsRemoteCompletionConfig,
    filter_completion,
)

from .util import (
    find_python,
    get_available_python_versions,
    str_to_version,
    version_to_str,
    Architecture,
    find_executable,
)


@dataclasses.dataclass(frozen=True)
class OriginCompletionConfig(PosixCompletionConfig):
    """Pseudo completion config for the origin."""

    def __init__(self) -> None:
        super().__init__(name='origin')

    @property
    def supported_pythons(self) -> list[str]:
        """Return a list of the supported Python versions."""
        current_version = version_to_str(sys.version_info[:2])
        versions = [version for version in SUPPORTED_PYTHON_VERSIONS if version == current_version] + \
                   [version for version in SUPPORTED_PYTHON_VERSIONS if version != current_version]
        return versions

    def get_python_path(self, version: str) -> str:
        """Return the path of the requested Python version."""
        version = find_python(version)
        return version

    @property
    def supported_powershells(self) -> list[str]:
        """Return a list of the supported PowerShell versions."""
        return SUPPORTED_POWERSHELL_VERSIONS

    def get_powershell_path(self, version: str | None) -> str | None:
        """
        Return the path of the requested PowerShell version, or None if it is not found.
        If no version is specified, look for the default unversioned 'pwsh' interpreter.
        """
        if not version:
            pass  # FIXME: when does the origin have no version, and what to do about it?

        return find_executable(f'pwsh{version or ""}', required=bool(version))  # FIXME: is this the behavior we want?

    @property
    def is_default(self) -> bool:
        """True if the completion entry is only used for defaults, otherwise False."""
        return False


@dataclasses.dataclass(frozen=True)
class HostContext:
    """Context used when getting and applying defaults for host configurations."""

    controller_config: t.Optional['PosixConfig']

    @property
    def controller(self) -> bool:
        """True if the context is for the controller, otherwise False."""
        return not self.controller_config


@dataclasses.dataclass
class HostConfig(metaclass=abc.ABCMeta):
    """Base class for host configuration."""

    @abc.abstractmethod
    def get_defaults(self, context: HostContext) -> CompletionConfig:
        """Return the default settings."""

    @abc.abstractmethod
    def apply_defaults(self, context: HostContext, defaults: CompletionConfig) -> None:
        """Apply default settings."""

    @property
    def is_managed(self) -> bool:
        """
        True if the host is a managed instance, otherwise False.
        Managed instances are used exclusively by ansible-test and can safely have destructive operations performed without explicit permission from the user.
        """
        return False


@dataclasses.dataclass
class PythonConfig(metaclass=abc.ABCMeta):
    """Configuration for Python."""

    version: t.Optional[str] = None
    path: t.Optional[str] = None

    @property
    def tuple(self) -> tuple[int, ...]:
        """Return the Python version as a tuple."""
        return str_to_version(self.version)

    @property
    def major_version(self) -> int:
        """Return the Python major version."""
        return self.tuple[0]

    def apply_defaults(self, context: HostContext, defaults: PosixCompletionConfig) -> None:
        """Apply default settings."""
        if self.version in (None, 'default'):
            self.version = defaults.get_default_python(context.controller)

        if self.path:
            if self.path.endswith('/'):
                self.path = os.path.join(self.path, f'python{self.version}')

            # FUTURE: If the host is origin, the python path could be validated here.
        else:
            self.path = defaults.get_python_path(self.version)

    @property
    @abc.abstractmethod
    def is_managed(self) -> bool:
        """
        True if this Python is a managed instance, otherwise False.
        Managed instances are used exclusively by ansible-test and can safely have requirements installed without explicit permission from the user.
        """


@dataclasses.dataclass
class NativePythonConfig(PythonConfig):
    """Configuration for native Python."""

    @property
    def is_managed(self) -> bool:
        """
        True if this Python is a managed instance, otherwise False.
        Managed instances are used exclusively by ansible-test and can safely have requirements installed without explicit permission from the user.
        """
        return False


@dataclasses.dataclass
class VirtualPythonConfig(PythonConfig):
    """Configuration for Python in a virtual environment."""

    system_site_packages: t.Optional[bool] = None

    def apply_defaults(self, context: HostContext, defaults: PosixCompletionConfig) -> None:
        """Apply default settings."""
        super().apply_defaults(context, defaults)

        if self.system_site_packages is None:
            self.system_site_packages = False

    @property
    def is_managed(self) -> bool:
        """
        True if this Python is a managed instance, otherwise False.
        Managed instances are used exclusively by ansible-test and can safely have requirements installed without explicit permission from the user.
        """
        return True


@dataclasses.dataclass
class PowerShellConfig:
    """Configuration for PowerShell."""

    version: str | None = None
    path: str | None = None

    @property
    def tuple(self) -> tuple[int, ...]:
        """Return the PowerShell version as a tuple."""
        return str_to_version(self.version)

    @property
    def major_version(self) -> int:
        """Return the PowerShell major version."""
        return self.tuple[0]

    def apply_defaults(self, context: HostContext, defaults: PosixCompletionConfig) -> None:
        """Apply default settings."""
        if self.version in (None, 'default'):
            self.version = defaults.get_default_powershell()

        if self.path:
            if self.path.endswith('/'):
                self.path = os.path.join(self.path, f'pwsh{self.version or ""}')  # FIXME: is the version adding and/or fallback behavior what we want?

            # FUTURE: If the host is origin, the pwsh path could be validated here.
        else:
            self.path = defaults.get_powershell_path(self.version)


@dataclasses.dataclass
class PosixConfig(HostConfig, metaclass=abc.ABCMeta):
    """Base class for POSIX host configuration."""

    python: t.Optional[PythonConfig] = None
    powershell: PowerShellConfig | None = None

    @property
    @abc.abstractmethod
    def have_root(self) -> bool:
        """True if root is available, otherwise False."""

    @abc.abstractmethod
    def get_defaults(self, context: HostContext) -> PosixCompletionConfig:
        """Return the default settings."""

    def apply_defaults(self, context: HostContext, defaults: CompletionConfig) -> None:
        """Apply default settings."""
        assert isinstance(defaults, PosixCompletionConfig)

        super().apply_defaults(context, defaults)

        self.python = self.python or NativePythonConfig()
        self.python.apply_defaults(context, defaults)

        self.powershell = self.powershell or PowerShellConfig()
        self.powershell.apply_defaults(context, defaults)


@dataclasses.dataclass
class ControllerHostConfig(PosixConfig, metaclass=abc.ABCMeta):
    """Base class for host configurations which support the controller."""

    @abc.abstractmethod
    def get_default_targets(self, context: HostContext) -> list[ControllerConfig]:
        """Return the default targets for this host config."""


@dataclasses.dataclass
class RemoteConfig(HostConfig, metaclass=abc.ABCMeta):
    """Base class for remote host configuration."""

    name: t.Optional[str] = None
    provider: t.Optional[str] = None
    arch: t.Optional[str] = None

    @property
    def platform(self) -> str:
        """The name of the platform."""
        return self.name.partition('/')[0]

    @property
    def version(self) -> str:
        """The version of the platform."""
        return self.name.partition('/')[2]

    def apply_defaults(self, context: HostContext, defaults: CompletionConfig) -> None:
        """Apply default settings."""
        assert isinstance(defaults, RemoteCompletionConfig)

        super().apply_defaults(context, defaults)

        if self.provider == 'default':
            self.provider = None

        self.provider = self.provider or defaults.provider or 'aws'
        self.arch = self.arch or defaults.arch or Architecture.X86_64

    @property
    def is_managed(self) -> bool:
        """
        True if this host is a managed instance, otherwise False.
        Managed instances are used exclusively by ansible-test and can safely have destructive operations performed without explicit permission from the user.
        """
        return True


@dataclasses.dataclass
class PosixSshConfig(PosixConfig):
    """Configuration for a POSIX SSH host."""

    user: t.Optional[str] = None
    host: t.Optional[str] = None
    port: t.Optional[int] = None

    def get_defaults(self, context: HostContext) -> PosixSshCompletionConfig:
        """Return the default settings."""
        return PosixSshCompletionConfig(
            user=self.user,
            host=self.host,
        )

    @property
    def have_root(self) -> bool:
        """True if root is available, otherwise False."""
        return self.user == 'root'


@dataclasses.dataclass
class InventoryConfig(HostConfig):
    """Configuration using inventory."""

    path: t.Optional[str] = None

    def get_defaults(self, context: HostContext) -> InventoryCompletionConfig:
        """Return the default settings."""
        return InventoryCompletionConfig()

    def apply_defaults(self, context: HostContext, defaults: CompletionConfig) -> None:
        """Apply default settings."""
        assert isinstance(defaults, InventoryCompletionConfig)


@dataclasses.dataclass
class DockerConfig(ControllerHostConfig, PosixConfig):
    """Configuration for a docker host."""

    name: t.Optional[str] = None
    image: t.Optional[str] = None
    memory: t.Optional[int] = None
    privileged: t.Optional[bool] = None
    seccomp: t.Optional[str] = None
    cgroup: t.Optional[CGroupVersion] = None
    audit: t.Optional[AuditMode] = None

    def get_defaults(self, context: HostContext) -> DockerCompletionConfig:
        """Return the default settings."""
        return filter_completion(docker_completion()).get(self.name) or DockerCompletionConfig(
            name=self.name,
            image=self.name,
            placeholder=True,
        )

    def get_default_targets(self, context: HostContext) -> list[ControllerConfig]:
        """Return the default targets for this host config."""
        if self.name in filter_completion(docker_completion()):
            defaults = self.get_defaults(context)
            pythons = {version: defaults.get_python_path(version) for version in defaults.supported_pythons}
        else:
            pythons = {context.controller_config.python.version: context.controller_config.python.path}

        return [ControllerConfig(python=NativePythonConfig(version=version, path=path)) for version, path in pythons.items()]

    def apply_defaults(self, context: HostContext, defaults: CompletionConfig) -> None:
        """Apply default settings."""
        assert isinstance(defaults, DockerCompletionConfig)

        super().apply_defaults(context, defaults)

        self.name = defaults.name
        self.image = defaults.image

        if self.seccomp is None:
            self.seccomp = defaults.seccomp

        if self.cgroup is None:
            self.cgroup = defaults.cgroup_enum

        if self.audit is None:
            self.audit = defaults.audit_enum

        if self.privileged is None:
            self.privileged = False

    @property
    def is_managed(self) -> bool:
        """
        True if this host is a managed instance, otherwise False.
        Managed instances are used exclusively by ansible-test and can safely have destructive operations performed without explicit permission from the user.
        """
        return True

    @property
    def have_root(self) -> bool:
        """True if root is available, otherwise False."""
        return True


@dataclasses.dataclass
class PosixRemoteConfig(RemoteConfig, ControllerHostConfig, PosixConfig):
    """Configuration for a POSIX remote host."""

    become: t.Optional[str] = None

    def get_defaults(self, context: HostContext) -> PosixRemoteCompletionConfig:
        """Return the default settings."""
        # pylint: disable=unexpected-keyword-arg  # see: https://github.com/PyCQA/pylint/issues/7434
        return filter_completion(remote_completion()).get(self.name) or remote_completion().get(self.platform) or PosixRemoteCompletionConfig(
            name=self.name,
            placeholder=True,
        )

    def get_default_targets(self, context: HostContext) -> list[ControllerConfig]:
        """Return the default targets for this host config."""
        if self.name in filter_completion(remote_completion()):
            defaults = self.get_defaults(context)
            pythons = {version: defaults.get_python_path(version) for version in defaults.supported_pythons}
        else:
            pythons = {context.controller_config.python.version: context.controller_config.python.path}

        return [ControllerConfig(python=NativePythonConfig(version=version, path=path)) for version, path in pythons.items()]

    def apply_defaults(self, context: HostContext, defaults: CompletionConfig) -> None:
        """Apply default settings."""
        assert isinstance(defaults, PosixRemoteCompletionConfig)

        super().apply_defaults(context, defaults)

        self.become = self.become or defaults.become

    @property
    def have_root(self) -> bool:
        """True if root is available, otherwise False."""
        return True


@dataclasses.dataclass
class WindowsConfig(HostConfig, metaclass=abc.ABCMeta):
    """Base class for Windows host configuration."""


@dataclasses.dataclass
class WindowsRemoteConfig(RemoteConfig, WindowsConfig):
    """Configuration for a remote Windows host."""

    connection: t.Optional[str] = None

    def get_defaults(self, context: HostContext) -> WindowsRemoteCompletionConfig:
        """Return the default settings."""
        return filter_completion(windows_completion()).get(self.name) or windows_completion().get(self.platform)

    def apply_defaults(self, context: HostContext, defaults: CompletionConfig) -> None:
        """Apply default settings."""
        assert isinstance(defaults, WindowsRemoteCompletionConfig)

        super().apply_defaults(context, defaults)

        self.connection = self.connection or defaults.connection


@dataclasses.dataclass
class WindowsInventoryConfig(InventoryConfig, WindowsConfig):
    """Configuration for Windows hosts using inventory."""


@dataclasses.dataclass
class NetworkConfig(HostConfig, metaclass=abc.ABCMeta):
    """Base class for network host configuration."""


@dataclasses.dataclass
class NetworkRemoteConfig(RemoteConfig, NetworkConfig):
    """Configuration for a remote network host."""

    collection: t.Optional[str] = None
    connection: t.Optional[str] = None

    def get_defaults(self, context: HostContext) -> NetworkRemoteCompletionConfig:
        """Return the default settings."""
        return filter_completion(network_completion()).get(self.name) or NetworkRemoteCompletionConfig(
            name=self.name,
            placeholder=True,
        )

    def apply_defaults(self, context: HostContext, defaults: CompletionConfig) -> None:
        """Apply default settings."""
        assert isinstance(defaults, NetworkRemoteCompletionConfig)

        super().apply_defaults(context, defaults)

        self.collection = self.collection or defaults.collection
        self.connection = self.connection or defaults.connection


@dataclasses.dataclass
class NetworkInventoryConfig(InventoryConfig, NetworkConfig):
    """Configuration for network hosts using inventory."""


@dataclasses.dataclass
class OriginConfig(ControllerHostConfig, PosixConfig):
    """Configuration for the origin host."""

    def get_defaults(self, context: HostContext) -> OriginCompletionConfig:
        """Return the default settings."""
        return OriginCompletionConfig()

    def get_default_targets(self, context: HostContext) -> list[ControllerConfig]:
        """Return the default targets for this host config."""
        return [ControllerConfig(python=NativePythonConfig(version=version, path=path)) for version, path in get_available_python_versions().items()]

    @property
    def have_root(self) -> bool:
        """True if root is available, otherwise False."""
        return os.getuid() == 0


@dataclasses.dataclass
class ControllerConfig(PosixConfig):
    """Configuration for the controller host."""

    controller: t.Optional[PosixConfig] = None

    def get_defaults(self, context: HostContext) -> PosixCompletionConfig:
        """Return the default settings."""
        return context.controller_config.get_defaults(context)

    def apply_defaults(self, context: HostContext, defaults: CompletionConfig) -> None:
        """Apply default settings."""
        assert isinstance(defaults, PosixCompletionConfig)

        self.controller = context.controller_config

        if not self.python and not defaults.supported_pythons:
            # The user did not specify a target Python and supported Pythons are unknown, so use the controller Python specified by the user instead.
            self.python = context.controller_config.python

        if not self.powershell and not defaults.supported_powershells:
            # FIXME: is this logic correct?
            # The user did not specify a target PowerShell and supported versions are unknown, so use the controller version specified by the user instead.
            self.powershell = context.controller_config.powershell

        super().apply_defaults(context, defaults)

    @property
    def is_managed(self) -> bool:
        """
        True if the host is a managed instance, otherwise False.
        Managed instances are used exclusively by ansible-test and can safely have destructive operations performed without explicit permission from the user.
        """
        return self.controller.is_managed

    @property
    def have_root(self) -> bool:
        """True if root is available, otherwise False."""
        return self.controller.have_root


class FallbackReason(enum.Enum):
    """Reason fallback was performed."""

    ENVIRONMENT = enum.auto()
    PYTHON = enum.auto()


@dataclasses.dataclass(frozen=True)
class FallbackDetail:
    """Details about controller fallback behavior."""

    reason: FallbackReason
    message: str


@dataclasses.dataclass(frozen=True)
class HostSettings:
    """Host settings for the controller and targets."""

    controller: ControllerHostConfig
    targets: list[HostConfig]
    skipped_python_versions: list[str]
    filtered_args: list[str]
    controller_fallback: t.Optional[FallbackDetail]

    def serialize(self, path: str) -> None:
        """Serialize the host settings to the given path."""
        with open_binary_file(path, 'wb') as settings_file:
            pickle.dump(self, settings_file)

    @staticmethod
    def deserialize(path: str) -> HostSettings:
        """Deserialize host settings from the path."""
        with open_binary_file(path) as settings_file:
            return pickle.load(settings_file)

    def apply_defaults(self) -> None:
        """Apply defaults to the host settings."""
        context = HostContext(controller_config=None)
        self.controller.apply_defaults(context, self.controller.get_defaults(context))

        for target in self.targets:
            context = HostContext(controller_config=self.controller)
            target.apply_defaults(context, target.get_defaults(context))