487 lines
16 KiB
Python
487 lines
16 KiB
Python
"""Common utility code that depends on CommonConfig."""
|
|
from __future__ import (absolute_import, division, print_function)
|
|
__metaclass__ = type
|
|
|
|
import atexit
|
|
import contextlib
|
|
import os
|
|
import shutil
|
|
import sys
|
|
import tempfile
|
|
import textwrap
|
|
|
|
from . import types as t
|
|
|
|
from .encoding import (
|
|
to_bytes,
|
|
)
|
|
|
|
from .util import (
|
|
common_environment,
|
|
COVERAGE_CONFIG_NAME,
|
|
display,
|
|
find_python,
|
|
remove_tree,
|
|
MODE_DIRECTORY,
|
|
MODE_FILE_EXECUTE,
|
|
PYTHON_PATHS,
|
|
raw_command,
|
|
read_lines_without_comments,
|
|
ANSIBLE_TEST_DATA_ROOT,
|
|
ApplicationError,
|
|
)
|
|
|
|
from .io import (
|
|
write_text_file,
|
|
write_json_file,
|
|
)
|
|
|
|
from .data import (
|
|
data_context,
|
|
)
|
|
|
|
from .provider.layout import (
|
|
LayoutMessages,
|
|
)
|
|
|
|
DOCKER_COMPLETION = {} # type: t.Dict[str, t.Dict[str, str]]
|
|
REMOTE_COMPLETION = {} # type: t.Dict[str, t.Dict[str, str]]
|
|
NETWORK_COMPLETION = {} # type: t.Dict[str, t.Dict[str, str]]
|
|
|
|
|
|
class ResultType:
|
|
"""Test result type."""
|
|
BOT = None # type: ResultType
|
|
COVERAGE = None # type: ResultType
|
|
DATA = None # type: ResultType
|
|
JUNIT = None # type: ResultType
|
|
LOGS = None # type: ResultType
|
|
REPORTS = None # type: ResultType
|
|
TMP = None # type: ResultType
|
|
|
|
@staticmethod
|
|
def _populate():
|
|
ResultType.BOT = ResultType('bot')
|
|
ResultType.COVERAGE = ResultType('coverage')
|
|
ResultType.DATA = ResultType('data')
|
|
ResultType.JUNIT = ResultType('junit')
|
|
ResultType.LOGS = ResultType('logs')
|
|
ResultType.REPORTS = ResultType('reports')
|
|
ResultType.TMP = ResultType('.tmp')
|
|
|
|
def __init__(self, name): # type: (str) -> None
|
|
self.name = name
|
|
|
|
@property
|
|
def relative_path(self): # type: () -> str
|
|
"""The content relative path to the results."""
|
|
return os.path.join(data_context().content.results_path, self.name)
|
|
|
|
@property
|
|
def path(self): # type: () -> str
|
|
"""The absolute path to the results."""
|
|
return os.path.join(data_context().content.root, self.relative_path)
|
|
|
|
def __str__(self): # type: () -> str
|
|
return self.name
|
|
|
|
|
|
# noinspection PyProtectedMember
|
|
ResultType._populate() # pylint: disable=protected-access
|
|
|
|
|
|
class CommonConfig:
|
|
"""Configuration common to all commands."""
|
|
def __init__(self, args, command):
|
|
"""
|
|
:type args: any
|
|
:type command: str
|
|
"""
|
|
self.command = command
|
|
|
|
self.color = args.color # type: bool
|
|
self.explain = args.explain # type: bool
|
|
self.verbosity = args.verbosity # type: int
|
|
self.debug = args.debug # type: bool
|
|
self.truncate = args.truncate # type: int
|
|
self.redact = args.redact # type: bool
|
|
|
|
self.info_stderr = False # type: bool
|
|
|
|
self.cache = {}
|
|
|
|
def get_ansible_config(self): # type: () -> str
|
|
"""Return the path to the Ansible config for the given config."""
|
|
return os.path.join(ANSIBLE_TEST_DATA_ROOT, 'ansible.cfg')
|
|
|
|
|
|
class NetworkPlatformSettings:
|
|
"""Settings required for provisioning a network platform."""
|
|
def __init__(self, collection, inventory_vars): # type: (str, t.Type[str, str]) -> None
|
|
self.collection = collection
|
|
self.inventory_vars = inventory_vars
|
|
|
|
|
|
def get_docker_completion():
|
|
"""
|
|
:rtype: dict[str, dict[str, str]]
|
|
"""
|
|
return get_parameterized_completion(DOCKER_COMPLETION, 'docker')
|
|
|
|
|
|
def get_remote_completion():
|
|
"""
|
|
:rtype: dict[str, dict[str, str]]
|
|
"""
|
|
return get_parameterized_completion(REMOTE_COMPLETION, 'remote')
|
|
|
|
|
|
def get_network_completion():
|
|
"""
|
|
:rtype: dict[str, dict[str, str]]
|
|
"""
|
|
return get_parameterized_completion(NETWORK_COMPLETION, 'network')
|
|
|
|
|
|
def get_parameterized_completion(cache, name):
|
|
"""
|
|
:type cache: dict[str, dict[str, str]]
|
|
:type name: str
|
|
:rtype: dict[str, dict[str, str]]
|
|
"""
|
|
if not cache:
|
|
if data_context().content.collection:
|
|
context = 'collection'
|
|
else:
|
|
context = 'ansible-base'
|
|
|
|
images = read_lines_without_comments(os.path.join(ANSIBLE_TEST_DATA_ROOT, 'completion', '%s.txt' % name), remove_blank_lines=True)
|
|
|
|
cache.update(dict(kvp for kvp in [parse_parameterized_completion(i) for i in images] if kvp and kvp[1].get('context', context) == context))
|
|
|
|
return cache
|
|
|
|
|
|
def parse_parameterized_completion(value): # type: (str) -> t.Optional[t.Tuple[str, t.Dict[str, str]]]
|
|
"""Parse the given completion entry, returning the entry name and a dictionary of key/value settings."""
|
|
values = value.split()
|
|
|
|
if not values:
|
|
return None
|
|
|
|
name = values[0]
|
|
data = dict((kvp[0], kvp[1] if len(kvp) > 1 else '') for kvp in [item.split('=', 1) for item in values[1:]])
|
|
|
|
return name, data
|
|
|
|
|
|
def docker_qualify_image(name):
|
|
"""
|
|
:type name: str
|
|
:rtype: str
|
|
"""
|
|
config = get_docker_completion().get(name, {})
|
|
|
|
return config.get('name', name)
|
|
|
|
|
|
def get_network_settings(args, platform, version): # type: (NetworkIntegrationConfig, str, str) -> NetworkPlatformSettings
|
|
"""Returns settings for the given network platform and version."""
|
|
platform_version = '%s/%s' % (platform, version)
|
|
completion = get_network_completion().get(platform_version, {})
|
|
collection = args.platform_collection.get(platform, completion.get('collection'))
|
|
|
|
settings = NetworkPlatformSettings(
|
|
collection,
|
|
dict(
|
|
ansible_connection=args.platform_connection.get(platform, completion.get('connection')),
|
|
ansible_network_os='%s.%s' % (collection, platform) if collection else platform,
|
|
)
|
|
)
|
|
|
|
return settings
|
|
|
|
|
|
def handle_layout_messages(messages): # type: (t.Optional[LayoutMessages]) -> None
|
|
"""Display the given layout messages."""
|
|
if not messages:
|
|
return
|
|
|
|
for message in messages.info:
|
|
display.info(message, verbosity=1)
|
|
|
|
for message in messages.warning:
|
|
display.warning(message)
|
|
|
|
if messages.error:
|
|
raise ApplicationError('\n'.join(messages.error))
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def named_temporary_file(args, prefix, suffix, directory, content):
|
|
"""
|
|
:param args: CommonConfig
|
|
:param prefix: str
|
|
:param suffix: str
|
|
:param directory: str
|
|
:param content: str | bytes | unicode
|
|
:rtype: str
|
|
"""
|
|
if args.explain:
|
|
yield os.path.join(directory, '%stemp%s' % (prefix, suffix))
|
|
else:
|
|
with tempfile.NamedTemporaryFile(prefix=prefix, suffix=suffix, dir=directory) as tempfile_fd:
|
|
tempfile_fd.write(to_bytes(content))
|
|
tempfile_fd.flush()
|
|
|
|
yield tempfile_fd.name
|
|
|
|
|
|
def write_json_test_results(category, # type: ResultType
|
|
name, # type: str
|
|
content, # type: t.Union[t.List[t.Any], t.Dict[str, t.Any]]
|
|
formatted=True, # type: bool
|
|
encoder=None, # type: t.Optional[t.Callable[[t.Any], t.Any]]
|
|
): # type: (...) -> None
|
|
"""Write the given json content to the specified test results path, creating directories as needed."""
|
|
path = os.path.join(category.path, name)
|
|
write_json_file(path, content, create_directories=True, formatted=formatted, encoder=encoder)
|
|
|
|
|
|
def write_text_test_results(category, name, content): # type: (ResultType, str, str) -> None
|
|
"""Write the given text content to the specified test results path, creating directories as needed."""
|
|
path = os.path.join(category.path, name)
|
|
write_text_file(path, content, create_directories=True)
|
|
|
|
|
|
def get_python_path(args, interpreter):
|
|
"""
|
|
:type args: TestConfig
|
|
:type interpreter: str
|
|
:rtype: str
|
|
"""
|
|
python_path = PYTHON_PATHS.get(interpreter)
|
|
|
|
if python_path:
|
|
return python_path
|
|
|
|
prefix = 'python-'
|
|
suffix = '-ansible'
|
|
|
|
root_temp_dir = '/tmp'
|
|
|
|
if args.explain:
|
|
return os.path.join(root_temp_dir, ''.join((prefix, 'temp', suffix)))
|
|
|
|
python_path = tempfile.mkdtemp(prefix=prefix, suffix=suffix, dir=root_temp_dir)
|
|
injected_interpreter = os.path.join(python_path, 'python')
|
|
|
|
# A symlink is faster than the execv wrapper, but isn't compatible with virtual environments.
|
|
# Attempt to detect when it is safe to use a symlink by checking the real path of the interpreter.
|
|
use_symlink = os.path.dirname(os.path.realpath(interpreter)) == os.path.dirname(interpreter)
|
|
|
|
if use_symlink:
|
|
display.info('Injecting "%s" as a symlink to the "%s" interpreter.' % (injected_interpreter, interpreter), verbosity=1)
|
|
|
|
os.symlink(interpreter, injected_interpreter)
|
|
else:
|
|
display.info('Injecting "%s" as a execv wrapper for the "%s" interpreter.' % (injected_interpreter, interpreter), verbosity=1)
|
|
|
|
create_interpreter_wrapper(interpreter, injected_interpreter)
|
|
|
|
os.chmod(python_path, MODE_DIRECTORY)
|
|
|
|
if not PYTHON_PATHS:
|
|
atexit.register(cleanup_python_paths)
|
|
|
|
PYTHON_PATHS[interpreter] = python_path
|
|
|
|
return python_path
|
|
|
|
|
|
def create_temp_dir(prefix=None, suffix=None, base_dir=None): # type: (t.Optional[str], t.Optional[str], t.Optional[str]) -> str
|
|
"""Create a temporary directory that persists until the current process exits."""
|
|
temp_path = tempfile.mkdtemp(prefix=prefix or 'tmp', suffix=suffix or '', dir=base_dir)
|
|
atexit.register(remove_tree, temp_path)
|
|
return temp_path
|
|
|
|
|
|
def create_interpreter_wrapper(interpreter, injected_interpreter): # type: (str, str) -> None
|
|
"""Create a wrapper for the given Python interpreter at the specified path."""
|
|
# sys.executable is used for the shebang to guarantee it is a binary instead of a script
|
|
# injected_interpreter could be a script from the system or our own wrapper created for the --venv option
|
|
shebang_interpreter = sys.executable
|
|
|
|
code = textwrap.dedent('''
|
|
#!%s
|
|
|
|
from __future__ import absolute_import
|
|
|
|
from os import execv
|
|
from sys import argv
|
|
|
|
python = '%s'
|
|
|
|
execv(python, [python] + argv[1:])
|
|
''' % (shebang_interpreter, interpreter)).lstrip()
|
|
|
|
write_text_file(injected_interpreter, code)
|
|
|
|
os.chmod(injected_interpreter, MODE_FILE_EXECUTE)
|
|
|
|
|
|
def cleanup_python_paths():
|
|
"""Clean up all temporary python directories."""
|
|
for path in sorted(PYTHON_PATHS.values()):
|
|
display.info('Cleaning up temporary python directory: %s' % path, verbosity=2)
|
|
shutil.rmtree(path)
|
|
|
|
|
|
def get_coverage_environment(args, target_name, version, temp_path, module_coverage, remote_temp_path=None):
|
|
"""
|
|
:type args: TestConfig
|
|
:type target_name: str
|
|
:type version: str
|
|
:type temp_path: str
|
|
:type module_coverage: bool
|
|
:type remote_temp_path: str | None
|
|
:rtype: dict[str, str]
|
|
"""
|
|
if temp_path:
|
|
# integration tests (both localhost and the optional testhost)
|
|
# config and results are in a temporary directory
|
|
coverage_config_base_path = temp_path
|
|
coverage_output_base_path = temp_path
|
|
elif args.coverage_config_base_path:
|
|
# unit tests, sanity tests and other special cases (localhost only)
|
|
# config is in a temporary directory
|
|
# results are in the source tree
|
|
coverage_config_base_path = args.coverage_config_base_path
|
|
coverage_output_base_path = os.path.join(data_context().content.root, data_context().content.results_path)
|
|
else:
|
|
raise Exception('No temp path and no coverage config base path. Check for missing coverage_context usage.')
|
|
|
|
config_file = os.path.join(coverage_config_base_path, COVERAGE_CONFIG_NAME)
|
|
coverage_file = os.path.join(coverage_output_base_path, ResultType.COVERAGE.name, '%s=%s=%s=%s=coverage' % (
|
|
args.command, target_name, args.coverage_label or 'local-%s' % version, 'python-%s' % version))
|
|
|
|
if not args.explain and not os.path.exists(config_file):
|
|
raise Exception('Missing coverage config file: %s' % config_file)
|
|
|
|
if args.coverage_check:
|
|
# cause the 'coverage' module to be found, but not imported or enabled
|
|
coverage_file = ''
|
|
|
|
# Enable code coverage collection on local Python programs (this does not include Ansible modules).
|
|
# Used by the injectors to support code coverage.
|
|
# Used by the pytest unit test plugin to support code coverage.
|
|
# The COVERAGE_FILE variable is also used directly by the 'coverage' module.
|
|
env = dict(
|
|
COVERAGE_CONF=config_file,
|
|
COVERAGE_FILE=coverage_file,
|
|
)
|
|
|
|
if module_coverage:
|
|
# Enable code coverage collection on Ansible modules (both local and remote).
|
|
# Used by the AnsiballZ wrapper generator in lib/ansible/executor/module_common.py to support code coverage.
|
|
env.update(dict(
|
|
_ANSIBLE_COVERAGE_CONFIG=config_file,
|
|
_ANSIBLE_COVERAGE_OUTPUT=coverage_file,
|
|
))
|
|
|
|
if remote_temp_path:
|
|
# Include the command, target and label so the remote host can create a filename with that info. The remote
|
|
# is responsible for adding '={language version}=coverage.{hostname}.{pid}.{id}'
|
|
env['_ANSIBLE_COVERAGE_REMOTE_OUTPUT'] = os.path.join(remote_temp_path, '%s=%s=%s' % (
|
|
args.command, target_name, args.coverage_label or 'remote'))
|
|
env['_ANSIBLE_COVERAGE_REMOTE_PATH_FILTER'] = os.path.join(data_context().content.root, '*')
|
|
|
|
return env
|
|
|
|
|
|
def intercept_command(args, cmd, target_name, env, capture=False, data=None, cwd=None, python_version=None, temp_path=None, module_coverage=True,
|
|
virtualenv=None, disable_coverage=False, remote_temp_path=None):
|
|
"""
|
|
:type args: TestConfig
|
|
:type cmd: collections.Iterable[str]
|
|
:type target_name: str
|
|
:type env: dict[str, str]
|
|
:type capture: bool
|
|
:type data: str | None
|
|
:type cwd: str | None
|
|
:type python_version: str | None
|
|
:type temp_path: str | None
|
|
:type module_coverage: bool
|
|
:type virtualenv: str | None
|
|
:type disable_coverage: bool
|
|
:type remote_temp_path: str | None
|
|
:rtype: str | None, str | None
|
|
"""
|
|
if not env:
|
|
env = common_environment()
|
|
else:
|
|
env = env.copy()
|
|
|
|
cmd = list(cmd)
|
|
version = python_version or args.python_version
|
|
interpreter = virtualenv or find_python(version)
|
|
inject_path = os.path.join(ANSIBLE_TEST_DATA_ROOT, 'injector')
|
|
|
|
if not virtualenv:
|
|
# injection of python into the path is required when not activating a virtualenv
|
|
# otherwise scripts may find the wrong interpreter or possibly no interpreter
|
|
python_path = get_python_path(args, interpreter)
|
|
inject_path = python_path + os.path.pathsep + inject_path
|
|
|
|
env['PATH'] = inject_path + os.path.pathsep + env['PATH']
|
|
env['ANSIBLE_TEST_PYTHON_VERSION'] = version
|
|
env['ANSIBLE_TEST_PYTHON_INTERPRETER'] = interpreter
|
|
|
|
if args.coverage and not disable_coverage:
|
|
# add the necessary environment variables to enable code coverage collection
|
|
env.update(get_coverage_environment(args, target_name, version, temp_path, module_coverage,
|
|
remote_temp_path=remote_temp_path))
|
|
|
|
return run_command(args, cmd, capture=capture, env=env, data=data, cwd=cwd)
|
|
|
|
|
|
def resolve_csharp_ps_util(import_name, path):
|
|
"""
|
|
:type import_name: str
|
|
:type path: str
|
|
"""
|
|
if data_context().content.is_ansible or not import_name.startswith('.'):
|
|
# We don't support relative paths for builtin utils, there's no point.
|
|
return import_name
|
|
|
|
packages = import_name.split('.')
|
|
module_packages = path.split(os.path.sep)
|
|
|
|
for package in packages:
|
|
if not module_packages or package:
|
|
break
|
|
del module_packages[-1]
|
|
|
|
return 'ansible_collections.%s%s' % (data_context().content.prefix,
|
|
'.'.join(module_packages + [p for p in packages if p]))
|
|
|
|
|
|
def run_command(args, cmd, capture=False, env=None, data=None, cwd=None, always=False, stdin=None, stdout=None,
|
|
cmd_verbosity=1, str_errors='strict'):
|
|
"""
|
|
:type args: CommonConfig
|
|
:type cmd: collections.Iterable[str]
|
|
:type capture: bool
|
|
:type env: dict[str, str] | None
|
|
:type data: str | None
|
|
:type cwd: str | None
|
|
:type always: bool
|
|
:type stdin: file | None
|
|
:type stdout: file | None
|
|
:type cmd_verbosity: int
|
|
:type str_errors: str
|
|
:rtype: str | None, str | None
|
|
"""
|
|
explain = args.explain and not always
|
|
return raw_command(cmd, capture=capture, env=env, data=data, cwd=cwd, explain=explain, stdin=stdin, stdout=stdout,
|
|
cmd_verbosity=cmd_verbosity, str_errors=str_errors)
|