diff options
author | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-04-28 16:04:21 +0000 |
---|---|---|
committer | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-04-28 16:04:21 +0000 |
commit | 8a754e0858d922e955e71b253c139e071ecec432 (patch) | |
tree | 527d16e74bfd1840c85efd675fdecad056c54107 /test/lib/ansible_test/_internal/host_configs.py | |
parent | Initial commit. (diff) | |
download | ansible-core-upstream.tar.xz ansible-core-upstream.zip |
Adding upstream version 2.14.3.upstream/2.14.3upstream
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to 'test/lib/ansible_test/_internal/host_configs.py')
-rw-r--r-- | test/lib/ansible_test/_internal/host_configs.py | 523 |
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)) |