diff options
Diffstat (limited to 'test/lib/ansible_test/_internal/cli/environments.py')
-rw-r--r-- | test/lib/ansible_test/_internal/cli/environments.py | 612 |
1 files changed, 612 insertions, 0 deletions
diff --git a/test/lib/ansible_test/_internal/cli/environments.py b/test/lib/ansible_test/_internal/cli/environments.py new file mode 100644 index 0000000..5063715 --- /dev/null +++ b/test/lib/ansible_test/_internal/cli/environments.py @@ -0,0 +1,612 @@ +"""Command line parsing for test environments.""" +from __future__ import annotations + +import argparse +import enum +import functools +import typing as t + +from ..constants import ( + CONTROLLER_PYTHON_VERSIONS, + REMOTE_PROVIDERS, + SECCOMP_CHOICES, + SUPPORTED_PYTHON_VERSIONS, +) + +from ..util import ( + REMOTE_ARCHITECTURES, +) + +from ..completion import ( + docker_completion, + network_completion, + remote_completion, + windows_completion, + filter_completion, +) + +from ..cli.argparsing import ( + CompositeAction, + CompositeActionCompletionFinder, +) + +from ..cli.argparsing.actions import ( + EnumAction, +) + +from ..cli.actions import ( + DelegatedControllerAction, + NetworkSshTargetAction, + NetworkTargetAction, + OriginControllerAction, + PosixSshTargetAction, + PosixTargetAction, + SanityPythonTargetAction, + UnitsPythonTargetAction, + WindowsSshTargetAction, + WindowsTargetAction, +) + +from ..cli.compat import ( + TargetMode, +) + +from ..config import ( + TerminateMode, +) + +from .completers import ( + complete_choices, + register_completer, +) + +from .converters import ( + key_value_type, +) + +from .epilog import ( + get_epilog, +) + +from ..ci import ( + get_ci_provider, +) + + +class ControllerMode(enum.Enum): + """Type of provisioning to use for the controller.""" + NO_DELEGATION = enum.auto() + ORIGIN = enum.auto() + DELEGATED = enum.auto() + + +def add_environments( + parser: argparse.ArgumentParser, + completer: CompositeActionCompletionFinder, + controller_mode: ControllerMode, + target_mode: TargetMode, +) -> None: + """Add arguments for the environments used to run ansible-test and commands it invokes.""" + no_environment = controller_mode == ControllerMode.NO_DELEGATION and target_mode == TargetMode.NO_TARGETS + + parser.set_defaults(no_environment=no_environment) + + if no_environment: + return + + parser.set_defaults(target_mode=target_mode) + + add_global_options(parser, controller_mode) + add_legacy_environment_options(parser, controller_mode, target_mode) + action_types = add_composite_environment_options(parser, completer, controller_mode, target_mode) + + sections = [f'{heading}\n{content}' + for action_type, documentation_state in CompositeAction.documentation_state.items() if action_type in action_types + for heading, content in documentation_state.sections.items()] + + if not get_ci_provider().supports_core_ci_auth(): + sections.append('Remote provisioning options have been hidden since no Ansible Core CI API key was found.') + + sections.append(get_epilog(completer)) + + parser.formatter_class = argparse.RawDescriptionHelpFormatter + parser.epilog = '\n\n'.join(sections) + + +def add_global_options( + parser: argparse.ArgumentParser, + controller_mode: ControllerMode, +): + """Add global options for controlling the test environment that work with both the legacy and composite options.""" + global_parser = t.cast(argparse.ArgumentParser, parser.add_argument_group(title='global environment arguments')) + + global_parser.add_argument( + '--containers', + metavar='JSON', + help=argparse.SUPPRESS, + ) + + global_parser.add_argument( + '--pypi-proxy', + action='store_true', + help=argparse.SUPPRESS, + ) + + global_parser.add_argument( + '--pypi-endpoint', + metavar='URI', + help=argparse.SUPPRESS, + ) + + global_parser.add_argument( + '--requirements', + action='store_true', + default=False, + help='install command requirements', + ) + + global_parser.add_argument( + '--no-pip-check', + action='store_true', + help=argparse.SUPPRESS, # deprecated, kept for now (with a warning) for backwards compatibility + ) + + add_global_remote(global_parser, controller_mode) + add_global_docker(global_parser, controller_mode) + + +def add_composite_environment_options( + parser: argparse.ArgumentParser, + completer: CompositeActionCompletionFinder, + controller_mode: ControllerMode, + target_mode: TargetMode, +) -> list[t.Type[CompositeAction]]: + """Add composite options for controlling the test environment.""" + composite_parser = t.cast(argparse.ArgumentParser, parser.add_argument_group( + title='composite environment arguments (mutually exclusive with "environment arguments" above)')) + + composite_parser.add_argument( + '--host-path', + help=argparse.SUPPRESS, + ) + + action_types: list[t.Type[CompositeAction]] = [] + + def register_action_type(action_type: t.Type[CompositeAction]) -> t.Type[CompositeAction]: + """Register the provided composite action type and return it.""" + action_types.append(action_type) + return action_type + + if controller_mode == ControllerMode.NO_DELEGATION: + composite_parser.set_defaults(controller=None) + else: + register_completer(composite_parser.add_argument( + '--controller', + metavar='OPT', + action=register_action_type(DelegatedControllerAction if controller_mode == ControllerMode.DELEGATED else OriginControllerAction), + help='configuration for the controller', + ), completer.completer) + + if target_mode == TargetMode.NO_TARGETS: + composite_parser.set_defaults(targets=[]) + elif target_mode == TargetMode.SHELL: + group = composite_parser.add_mutually_exclusive_group() + + register_completer(group.add_argument( + '--target-posix', + metavar='OPT', + action=register_action_type(PosixSshTargetAction), + help='configuration for the target', + ), completer.completer) + + suppress = None if get_ci_provider().supports_core_ci_auth() else argparse.SUPPRESS + + register_completer(group.add_argument( + '--target-windows', + metavar='OPT', + action=WindowsSshTargetAction if suppress else register_action_type(WindowsSshTargetAction), + help=suppress or 'configuration for the target', + ), completer.completer) + + register_completer(group.add_argument( + '--target-network', + metavar='OPT', + action=NetworkSshTargetAction if suppress else register_action_type(NetworkSshTargetAction), + help=suppress or 'configuration for the target', + ), completer.completer) + else: + if target_mode.multiple_pythons: + target_option = '--target-python' + target_help = 'configuration for the target python interpreter(s)' + elif target_mode == TargetMode.POSIX_INTEGRATION: + target_option = '--target' + target_help = 'configuration for the target' + else: + target_option = '--target' + target_help = 'configuration for the target(s)' + + target_actions = { + TargetMode.POSIX_INTEGRATION: PosixTargetAction, + TargetMode.WINDOWS_INTEGRATION: WindowsTargetAction, + TargetMode.NETWORK_INTEGRATION: NetworkTargetAction, + TargetMode.SANITY: SanityPythonTargetAction, + TargetMode.UNITS: UnitsPythonTargetAction, + } + + target_action = target_actions[target_mode] + + register_completer(composite_parser.add_argument( + target_option, + metavar='OPT', + action=register_action_type(target_action), + help=target_help, + ), completer.completer) + + return action_types + + +def add_legacy_environment_options( + parser: argparse.ArgumentParser, + controller_mode: ControllerMode, + target_mode: TargetMode, +): + """Add legacy options for controlling the test environment.""" + environment: argparse.ArgumentParser = parser.add_argument_group( # type: ignore[assignment] # real type private + title='environment arguments (mutually exclusive with "composite environment arguments" below)') + + add_environments_python(environment, target_mode) + add_environments_host(environment, controller_mode, target_mode) + + +def add_environments_python( + environments_parser: argparse.ArgumentParser, + target_mode: TargetMode, +) -> None: + """Add environment arguments to control the Python version(s) used.""" + python_versions: tuple[str, ...] + + if target_mode.has_python: + python_versions = SUPPORTED_PYTHON_VERSIONS + else: + python_versions = CONTROLLER_PYTHON_VERSIONS + + environments_parser.add_argument( + '--python', + metavar='X.Y', + choices=python_versions + ('default',), + help='python version: %s' % ', '.join(python_versions), + ) + + environments_parser.add_argument( + '--python-interpreter', + metavar='PATH', + help='path to the python interpreter', + ) + + +def add_environments_host( + environments_parser: argparse.ArgumentParser, + controller_mode: ControllerMode, + target_mode: TargetMode, +) -> None: + """Add environment arguments for the given host and argument modes.""" + environments_exclusive_group: argparse.ArgumentParser = environments_parser.add_mutually_exclusive_group() # type: ignore[assignment] # real type private + + add_environment_local(environments_exclusive_group) + add_environment_venv(environments_exclusive_group, environments_parser) + + if controller_mode == ControllerMode.DELEGATED: + add_environment_remote(environments_exclusive_group, environments_parser, target_mode) + add_environment_docker(environments_exclusive_group, environments_parser, target_mode) + + if target_mode == TargetMode.WINDOWS_INTEGRATION: + add_environment_windows(environments_parser) + + if target_mode == TargetMode.NETWORK_INTEGRATION: + add_environment_network(environments_parser) + + +def add_environment_network( + environments_parser: argparse.ArgumentParser, +) -> None: + """Add environment arguments for running on a windows host.""" + register_completer(environments_parser.add_argument( + '--platform', + metavar='PLATFORM', + action='append', + help='network platform/version', + ), complete_network_platform) + + register_completer(environments_parser.add_argument( + '--platform-collection', + type=key_value_type, + metavar='PLATFORM=COLLECTION', + action='append', + help='collection used to test platform', + ), complete_network_platform_collection) + + register_completer(environments_parser.add_argument( + '--platform-connection', + type=key_value_type, + metavar='PLATFORM=CONNECTION', + action='append', + help='connection used to test platform', + ), complete_network_platform_connection) + + environments_parser.add_argument( + '--inventory', + metavar='PATH', + help='path to inventory used for tests', + ) + + +def add_environment_windows( + environments_parser: argparse.ArgumentParser, +) -> None: + """Add environment arguments for running on a windows host.""" + register_completer(environments_parser.add_argument( + '--windows', + metavar='VERSION', + action='append', + help='windows version', + ), complete_windows) + + environments_parser.add_argument( + '--inventory', + metavar='PATH', + help='path to inventory used for tests', + ) + + +def add_environment_local( + exclusive_parser: argparse.ArgumentParser, +) -> None: + """Add environment arguments for running on the local (origin) host.""" + exclusive_parser.add_argument( + '--local', + action='store_true', + help='run from the local environment', + ) + + +def add_environment_venv( + exclusive_parser: argparse.ArgumentParser, + environments_parser: argparse.ArgumentParser, +) -> None: + """Add environment arguments for running in ansible-test managed virtual environments.""" + exclusive_parser.add_argument( + '--venv', + action='store_true', + help='run from a virtual environment', + ) + + environments_parser.add_argument( + '--venv-system-site-packages', + action='store_true', + help='enable system site packages') + + +def add_global_docker( + parser: argparse.ArgumentParser, + controller_mode: ControllerMode, +) -> None: + """Add global options for Docker.""" + if controller_mode != ControllerMode.DELEGATED: + parser.set_defaults( + docker_no_pull=False, + docker_network=None, + docker_terminate=None, + prime_containers=False, + dev_systemd_debug=False, + dev_probe_cgroups=None, + ) + + return + + parser.add_argument( + '--docker-no-pull', + action='store_true', + help=argparse.SUPPRESS, # deprecated, kept for now (with a warning) for backwards compatibility + ) + + parser.add_argument( + '--docker-network', + metavar='NET', + help='run using the specified network', + ) + + parser.add_argument( + '--docker-terminate', + metavar='T', + default=TerminateMode.ALWAYS, + type=TerminateMode, + action=EnumAction, + help='terminate the container: %(choices)s (default: %(default)s)', + ) + + parser.add_argument( + '--prime-containers', + action='store_true', + help='download containers without running tests', + ) + + # Docker support isn't related to ansible-core-ci. + # However, ansible-core-ci support is a reasonable indicator that the user may need the `--dev-*` options. + suppress = None if get_ci_provider().supports_core_ci_auth() else argparse.SUPPRESS + + parser.add_argument( + '--dev-systemd-debug', + action='store_true', + help=suppress or 'enable systemd debugging in containers', + ) + + parser.add_argument( + '--dev-probe-cgroups', + metavar='DIR', + nargs='?', + const='', + help=suppress or 'probe container cgroups, with optional log dir', + ) + + +def add_environment_docker( + exclusive_parser: argparse.ArgumentParser, + environments_parser: argparse.ArgumentParser, + target_mode: TargetMode, +) -> None: + """Add environment arguments for running in docker containers.""" + if target_mode in (TargetMode.POSIX_INTEGRATION, TargetMode.SHELL): + docker_images = sorted(filter_completion(docker_completion())) + else: + docker_images = sorted(filter_completion(docker_completion(), controller_only=True)) + + register_completer(exclusive_parser.add_argument( + '--docker', + metavar='IMAGE', + nargs='?', + const='default', + help='run from a docker container', + ), functools.partial(complete_choices, docker_images)) + + environments_parser.add_argument( + '--docker-privileged', + action='store_true', + help='run docker container in privileged mode', + ) + + environments_parser.add_argument( + '--docker-seccomp', + metavar='SC', + choices=SECCOMP_CHOICES, + help='set seccomp confinement for the test container: %(choices)s', + ) + + environments_parser.add_argument( + '--docker-memory', + metavar='INT', + type=int, + help='memory limit for docker in bytes', + ) + + +def add_global_remote( + parser: argparse.ArgumentParser, + controller_mode: ControllerMode, +) -> None: + """Add global options for remote instances.""" + if controller_mode != ControllerMode.DELEGATED: + parser.set_defaults( + remote_stage=None, + remote_endpoint=None, + remote_terminate=None, + ) + + return + + suppress = None if get_ci_provider().supports_core_ci_auth() else argparse.SUPPRESS + + register_completer(parser.add_argument( + '--remote-stage', + metavar='STAGE', + default='prod', + help=suppress or 'remote stage to use: prod, dev', + ), complete_remote_stage) + + parser.add_argument( + '--remote-endpoint', + metavar='EP', + help=suppress or 'remote provisioning endpoint to use', + ) + + parser.add_argument( + '--remote-terminate', + metavar='T', + default=TerminateMode.NEVER, + type=TerminateMode, + action=EnumAction, + help=suppress or 'terminate the remote instance: %(choices)s (default: %(default)s)', + ) + + +def add_environment_remote( + exclusive_parser: argparse.ArgumentParser, + environments_parser: argparse.ArgumentParser, + target_mode: TargetMode, +) -> None: + """Add environment arguments for running in ansible-core-ci provisioned remote virtual machines.""" + if target_mode == TargetMode.POSIX_INTEGRATION: + remote_platforms = get_remote_platform_choices() + elif target_mode == TargetMode.SHELL: + remote_platforms = sorted(set(get_remote_platform_choices()) | set(get_windows_platform_choices())) + else: + remote_platforms = get_remote_platform_choices(True) + + suppress = None if get_ci_provider().supports_core_ci_auth() else argparse.SUPPRESS + + register_completer(exclusive_parser.add_argument( + '--remote', + metavar='NAME', + help=suppress or 'run from a remote instance', + ), functools.partial(complete_choices, remote_platforms)) + + environments_parser.add_argument( + '--remote-provider', + metavar='PR', + choices=REMOTE_PROVIDERS, + help=suppress or 'remote provider to use: %(choices)s', + ) + + environments_parser.add_argument( + '--remote-arch', + metavar='ARCH', + choices=REMOTE_ARCHITECTURES, + help=suppress or 'remote arch to use: %(choices)s', + ) + + +def complete_remote_stage(prefix: str, **_) -> list[str]: + """Return a list of supported stages matching the given prefix.""" + return [stage for stage in ('prod', 'dev') if stage.startswith(prefix)] + + +def complete_windows(prefix: str, parsed_args: argparse.Namespace, **_) -> list[str]: + """Return a list of supported Windows versions matching the given prefix, excluding versions already parsed from the command line.""" + return [i for i in get_windows_version_choices() if i.startswith(prefix) and (not parsed_args.windows or i not in parsed_args.windows)] + + +def complete_network_platform(prefix: str, parsed_args: argparse.Namespace, **_) -> list[str]: + """Return a list of supported network platforms matching the given prefix, excluding platforms already parsed from the command line.""" + images = sorted(filter_completion(network_completion())) + + return [i for i in images if i.startswith(prefix) and (not parsed_args.platform or i not in parsed_args.platform)] + + +def complete_network_platform_collection(prefix: str, parsed_args: argparse.Namespace, **_) -> list[str]: + """Return a list of supported network platforms matching the given prefix, excluding collection platforms already parsed from the command line.""" + left = prefix.split('=')[0] + images = sorted(set(image.platform for image in filter_completion(network_completion()).values())) + + return [i + '=' for i in images if i.startswith(left) and (not parsed_args.platform_collection or i not in [x[0] for x in parsed_args.platform_collection])] + + +def complete_network_platform_connection(prefix: str, parsed_args: argparse.Namespace, **_) -> list[str]: + """Return a list of supported network platforms matching the given prefix, excluding connection platforms already parsed from the command line.""" + left = prefix.split('=')[0] + images = sorted(set(image.platform for image in filter_completion(network_completion()).values())) + + return [i + '=' for i in images if i.startswith(left) and (not parsed_args.platform_connection or i not in [x[0] for x in parsed_args.platform_connection])] + + +def get_remote_platform_choices(controller: bool = False) -> list[str]: + """Return a list of supported remote platforms matching the given prefix.""" + return sorted(filter_completion(remote_completion(), controller_only=controller)) + + +def get_windows_platform_choices() -> list[str]: + """Return a list of supported Windows versions matching the given prefix.""" + return sorted(f'windows/{windows.version}' for windows in filter_completion(windows_completion()).values()) + + +def get_windows_version_choices() -> list[str]: + """Return a list of supported Windows versions.""" + return sorted(windows.version for windows in filter_completion(windows_completion()).values()) |