You cannot select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
ansible/test/lib/ansible_test/_internal/host_configs.py

524 lines
18 KiB
Python

"""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))