1 files changed, 523 insertions, 0 deletions
diff --git a/test/lib/ansible_test/_internal/host_configs.py b/test/lib/ansible_test/_internal/host_configs.py
new file mode 100644
index 0000000..48d5fd3
--- /dev/null
+++ b/test/lib/ansible_test/_internal/host_configs.py
@@ -0,0 +1,523 @@
+"""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,
+)
+
+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,
+)
+
+
+@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 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 PosixConfig(HostConfig, metaclass=abc.ABCMeta):
+    """Base class for POSIX host configuration."""
+    python: t.Optional[PythonConfig] = 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)
+
+
+@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."""
+    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)
+
+
+@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
+
+        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))