#!/usr/bin/env python # PYTHON_ARGCOMPLETE_OK """Test runner for all Ansible tests.""" from __future__ import absolute_import, print_function import errno import os import sys from lib.util import ( ApplicationError, display, raw_command, ) from lib.delegation import ( delegate, ) from lib.executor import ( command_posix_integration, command_network_integration, command_windows_integration, command_units, command_compile, command_shell, SUPPORTED_PYTHON_VERSIONS, COMPILE_PYTHON_VERSIONS, ApplicationWarning, Delegate, generate_pip_install, check_startup, ) from lib.config import ( PosixIntegrationConfig, WindowsIntegrationConfig, NetworkIntegrationConfig, SanityConfig, UnitsConfig, CompileConfig, ShellConfig, ) from lib.sanity import ( command_sanity, sanity_init, sanity_get_tests, ) from lib.target import ( find_target_completion, walk_posix_integration_targets, walk_network_integration_targets, walk_windows_integration_targets, walk_units_targets, walk_compile_targets, walk_sanity_targets, ) from lib.core_ci import ( AWS_ENDPOINTS, ) from lib.cloud import ( initialize_cloud_plugins, ) import lib.cover def main(): """Main program function.""" try: git_root = os.path.abspath(os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', '..')) os.chdir(git_root) initialize_cloud_plugins() sanity_init() args = parse_args() config = args.config(args) display.verbosity = config.verbosity display.color = config.color display.info_stderr = isinstance(config, SanityConfig) and config.lint check_startup() try: args.func(config) except Delegate as ex: delegate(config, ex.exclude, ex.require) display.review_warnings() except ApplicationWarning as ex: display.warning(str(ex)) exit(0) except ApplicationError as ex: display.error(str(ex)) exit(1) except KeyboardInterrupt: exit(2) except IOError as ex: if ex.errno == errno.EPIPE: exit(3) raise def parse_args(): """Parse command line arguments.""" try: import argparse except ImportError: if '--requirements' not in sys.argv: raise raw_command(generate_pip_install('ansible-test')) import argparse try: import argcomplete except ImportError: argcomplete = None if argcomplete: epilog = 'Tab completion available using the "argcomplete" python package.' else: epilog = 'Install the "argcomplete" python package to enable tab completion.' parser = argparse.ArgumentParser(epilog=epilog) common = argparse.ArgumentParser(add_help=False) common.add_argument('-e', '--explain', action='store_true', help='explain commands that would be executed') common.add_argument('-v', '--verbose', dest='verbosity', action='count', default=0, help='display more output') common.add_argument('--color', metavar='COLOR', nargs='?', help='generate color output: %(choices)s', choices=('yes', 'no', 'auto'), const='yes', default='auto') common.add_argument('--debug', action='store_true', help='run ansible commands in debug mode') test = argparse.ArgumentParser(add_help=False, parents=[common]) test.add_argument('include', metavar='TARGET', nargs='*', help='test the specified target').completer = complete_target test.add_argument('--exclude', metavar='TARGET', action='append', help='exclude the specified target').completer = complete_target test.add_argument('--require', metavar='TARGET', action='append', help='require the specified target').completer = complete_target test.add_argument('--coverage', action='store_true', help='analyze code coverage when running tests') test.add_argument('--coverage-label', default='', help='label to include in coverage output file names') test.add_argument('--metadata', help=argparse.SUPPRESS) add_changes(test, argparse) add_environments(test) integration = argparse.ArgumentParser(add_help=False, parents=[test]) integration.add_argument('--python', metavar='VERSION', choices=SUPPORTED_PYTHON_VERSIONS + ('default',), help='python version: %s' % ', '.join(SUPPORTED_PYTHON_VERSIONS)) integration.add_argument('--start-at', metavar='TARGET', help='start at the specified target').completer = complete_target integration.add_argument('--start-at-task', metavar='TASK', help='start at the specified task') integration.add_argument('--tags', metavar='TAGS', help='only run plays and tasks tagged with these values') integration.add_argument('--skip-tags', metavar='TAGS', help='only run plays and tasks whose tags do not match these values') integration.add_argument('--diff', action='store_true', help='show diff output') integration.add_argument('--allow-destructive', action='store_true', help='allow destructive tests (--local and --tox only)') integration.add_argument('--retry-on-error', action='store_true', help='retry failed test with increased verbosity') integration.add_argument('--continue-on-error', action='store_true', help='continue after failed test') integration.add_argument('--debug-strategy', action='store_true', help='run test playbooks using the debug strategy') integration.add_argument('--changed-all-target', metavar='TARGET', default='all', help='target to run when all tests are needed') subparsers = parser.add_subparsers(metavar='COMMAND') subparsers.required = True # work-around for python 3 bug which makes subparsers optional posix_integration = subparsers.add_parser('integration', parents=[integration], help='posix integration tests') posix_integration.set_defaults(func=command_posix_integration, targets=walk_posix_integration_targets, config=PosixIntegrationConfig) add_extra_docker_options(posix_integration) network_integration = subparsers.add_parser('network-integration', parents=[integration], help='network integration tests') network_integration.set_defaults(func=command_network_integration, targets=walk_network_integration_targets, config=NetworkIntegrationConfig) network_integration.add_argument('--platform', metavar='PLATFORM', action='append', help='network platform/version').completer = complete_network_platform network_integration.add_argument('--inventory', metavar='PATH', help='path to inventory used for tests') windows_integration = subparsers.add_parser('windows-integration', parents=[integration], help='windows integration tests') windows_integration.set_defaults(func=command_windows_integration, targets=walk_windows_integration_targets, config=WindowsIntegrationConfig) windows_integration.add_argument('--windows', metavar='VERSION', action='append', help='windows version').completer = complete_windows units = subparsers.add_parser('units', parents=[test], help='unit tests') units.set_defaults(func=command_units, targets=walk_units_targets, config=UnitsConfig) units.add_argument('--python', metavar='VERSION', choices=SUPPORTED_PYTHON_VERSIONS + ('default',), help='python version: %s' % ', '.join(SUPPORTED_PYTHON_VERSIONS)) units.add_argument('--collect-only', action='store_true', help='collect tests but do not execute them') add_extra_docker_options(units, integration=False) compiler = subparsers.add_parser('compile', parents=[test], help='compile tests') compiler.set_defaults(func=command_compile, targets=walk_compile_targets, config=CompileConfig) compiler.add_argument('--python', metavar='VERSION', choices=COMPILE_PYTHON_VERSIONS + ('default',), help='python version: %s' % ', '.join(COMPILE_PYTHON_VERSIONS)) add_lint(compiler) add_extra_docker_options(compiler, integration=False) sanity = subparsers.add_parser('sanity', parents=[test], help='sanity tests') sanity.set_defaults(func=command_sanity, targets=walk_sanity_targets, config=SanityConfig) sanity.add_argument('--test', metavar='TEST', action='append', choices=[test.name for test in sanity_get_tests()], help='tests to run').completer = complete_sanity_test sanity.add_argument('--skip-test', metavar='TEST', action='append', choices=[test.name for test in sanity_get_tests()], help='tests to skip').completer = complete_sanity_test sanity.add_argument('--list-tests', action='store_true', help='list available tests') sanity.add_argument('--python', metavar='VERSION', choices=SUPPORTED_PYTHON_VERSIONS + ('default',), help='python version: %s' % ', '.join(SUPPORTED_PYTHON_VERSIONS)) sanity.add_argument('--base-branch', help=argparse.SUPPRESS) add_lint(sanity) add_extra_docker_options(sanity, integration=False) shell = subparsers.add_parser('shell', parents=[common], help='open an interactive shell') shell.set_defaults(func=command_shell, config=ShellConfig) add_environments(shell, tox_version=True) add_extra_docker_options(shell) coverage_common = argparse.ArgumentParser(add_help=False, parents=[common]) add_environments(coverage_common, tox_version=True, tox_only=True) coverage = subparsers.add_parser('coverage', help='code coverage management and reporting') coverage_subparsers = coverage.add_subparsers(metavar='COMMAND') coverage_subparsers.required = True # work-around for python 3 bug which makes subparsers optional coverage_combine = coverage_subparsers.add_parser('combine', parents=[coverage_common], help='combine coverage data and rewrite remote paths') coverage_combine.set_defaults(func=lib.cover.command_coverage_combine, config=lib.cover.CoverageConfig) add_extra_coverage_options(coverage_combine) coverage_erase = coverage_subparsers.add_parser('erase', parents=[coverage_common], help='erase coverage data files') coverage_erase.set_defaults(func=lib.cover.command_coverage_erase, config=lib.cover.CoverageConfig) coverage_report = coverage_subparsers.add_parser('report', parents=[coverage_common], help='generate console coverage report') coverage_report.set_defaults(func=lib.cover.command_coverage_report, config=lib.cover.CoverageReportConfig) coverage_report.add_argument('--show-missing', action='store_true', help='show line numbers of statements not executed') add_extra_coverage_options(coverage_report) coverage_html = coverage_subparsers.add_parser('html', parents=[coverage_common], help='generate html coverage report') coverage_html.set_defaults(func=lib.cover.command_coverage_html, config=lib.cover.CoverageConfig) add_extra_coverage_options(coverage_html) coverage_xml = coverage_subparsers.add_parser('xml', parents=[coverage_common], help='generate xml coverage report') coverage_xml.set_defaults(func=lib.cover.command_coverage_xml, config=lib.cover.CoverageConfig) add_extra_coverage_options(coverage_xml) if argcomplete: argcomplete.autocomplete(parser, always_complete_options=False, validator=lambda i, k: True) args = parser.parse_args() if args.explain and not args.verbosity: args.verbosity = 1 if args.color == 'yes': args.color = True elif args.color == 'no': args.color = False else: args.color = sys.stdout.isatty() return args def add_lint(parser): """ :type parser: argparse.ArgumentParser """ parser.add_argument('--lint', action='store_true', help='write lint output to stdout, everything else stderr') parser.add_argument('--junit', action='store_true', help='write test failures to junit xml files') parser.add_argument('--failure-ok', action='store_true', help='exit successfully on failed tests after saving results') def add_changes(parser, argparse): """ :type parser: argparse.ArgumentParser :type argparse: argparse """ parser.add_argument('--changed', action='store_true', help='limit targets based on changes') changes = parser.add_argument_group(title='change detection arguments') changes.add_argument('--tracked', action='store_true', help=argparse.SUPPRESS) changes.add_argument('--untracked', action='store_true', help='include untracked files') changes.add_argument('--ignore-committed', dest='committed', action='store_false', help='exclude committed files') changes.add_argument('--ignore-staged', dest='staged', action='store_false', help='exclude staged files') changes.add_argument('--ignore-unstaged', dest='unstaged', action='store_false', help='exclude unstaged files') changes.add_argument('--changed-from', metavar='PATH', help=argparse.SUPPRESS) changes.add_argument('--changed-path', metavar='PATH', action='append', help=argparse.SUPPRESS) def add_environments(parser, tox_version=False, tox_only=False): """ :type parser: argparse.ArgumentParser :type tox_version: bool :type tox_only: bool """ parser.add_argument('--requirements', action='store_true', help='install command requirements') environments = parser.add_mutually_exclusive_group() environments.add_argument('--local', action='store_true', help='run from the local environment') if tox_version: environments.add_argument('--tox', metavar='VERSION', nargs='?', default=None, const='.'.join(str(i) for i in sys.version_info[:2]), choices=SUPPORTED_PYTHON_VERSIONS, help='run from a tox virtualenv: %s' % ', '.join(SUPPORTED_PYTHON_VERSIONS)) else: environments.add_argument('--tox', action='store_true', help='run from a tox virtualenv') tox = parser.add_argument_group(title='tox arguments') tox.add_argument('--tox-sitepackages', action='store_true', help='allow access to globally installed packages') if tox_only: environments.set_defaults( docker=None, remote=None, remote_stage=None, remote_aws_region=None, remote_terminate=None, ) return environments.add_argument('--docker', metavar='IMAGE', nargs='?', default=None, const='ubuntu1604', help='run from a docker container').completer = complete_docker environments.add_argument('--remote', metavar='PLATFORM', default=None, help='run from a remote instance').completer = complete_remote remote = parser.add_argument_group(title='remote arguments') remote.add_argument('--remote-stage', metavar='STAGE', help='remote stage to use: %(choices)s', choices=['prod', 'dev'], default='prod') remote.add_argument('--remote-aws-region', metavar='REGION', help='remote aws region to use: %(choices)s (default: auto)', choices=sorted(AWS_ENDPOINTS), default=None) remote.add_argument('--remote-terminate', metavar='WHEN', help='terminate remote instance: %(choices)s (default: %(default)s)', choices=['never', 'always', 'success'], default='never') def add_extra_coverage_options(parser): """ :type parser: argparse.ArgumentParser """ parser.add_argument('--group-by', metavar='GROUP', action='append', choices=lib.cover.COVERAGE_GROUPS, help='group output by: %s' % ', '.join(lib.cover.COVERAGE_GROUPS)) parser.add_argument('--all', action='store_true', help='include all python source files') parser.add_argument('--stub', action='store_true', help='generate empty report of all python source files') def add_extra_docker_options(parser, integration=True): """ :type parser: argparse.ArgumentParser :type integration: bool """ docker = parser.add_argument_group(title='docker arguments') docker.add_argument('--docker-no-pull', action='store_false', dest='docker_pull', help='do not explicitly pull the latest docker images') if not integration: return docker.add_argument('--docker-util', metavar='IMAGE', default='httptester', help='docker utility image to provide test services') docker.add_argument('--docker-privileged', action='store_true', help='run docker container in privileged mode') def complete_target(prefix, parsed_args, **_): """ :type prefix: unicode :type parsed_args: any :rtype: list[str] """ return find_target_completion(parsed_args.targets, prefix) def complete_remote(prefix, parsed_args, **_): """ :type prefix: unicode :type parsed_args: any :rtype: list[str] """ del parsed_args with open('test/runner/completion/remote.txt', 'r') as completion_fd: images = completion_fd.read().splitlines() return [i for i in images if i.startswith(prefix)] def complete_docker(prefix, parsed_args, **_): """ :type prefix: unicode :type parsed_args: any :rtype: list[str] """ del parsed_args with open('test/runner/completion/docker.txt', 'r') as completion_fd: images = completion_fd.read().splitlines() return [i for i in images if i.startswith(prefix)] def complete_windows(prefix, parsed_args, **_): """ :type prefix: unicode :type parsed_args: any :rtype: list[str] """ with open('test/runner/completion/windows.txt', 'r') as completion_fd: images = completion_fd.read().splitlines() return [i for i in images if i.startswith(prefix) and (not parsed_args.windows or i not in parsed_args.windows)] def complete_network_platform(prefix, parsed_args, **_): """ :type prefix: unicode :type parsed_args: any :rtype: list[str] """ with open('test/runner/completion/network.txt', 'r') as completion_fd: images = completion_fd.read().splitlines() return [i for i in images if i.startswith(prefix) and (not parsed_args.platform or i not in parsed_args.platform)] def complete_sanity_test(prefix, parsed_args, **_): """ :type prefix: unicode :type parsed_args: any :rtype: list[str] """ del parsed_args tests = sorted(t.name for t in sanity_get_tests()) return [i for i in tests if i.startswith(prefix)] if __name__ == '__main__': main()