summaryrefslogtreecommitdiff
path: root/test/lib/ansible_test/_internal/units/__init__.py
blob: 2214543111752df838230fbaf32ceb0f6ebda7f1 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
"""Execute unit tests using pytest."""
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type

import os
import sys

from ..util import (
    ANSIBLE_TEST_DATA_ROOT,
    display,
    get_available_python_versions,
    is_subdir,
    SubprocessError,
    REMOTE_ONLY_PYTHON_VERSIONS,
)

from ..util_common import (
    intercept_command,
    ResultType,
    handle_layout_messages,
)

from ..ansible_util import (
    ansible_environment,
    check_pyyaml,
)

from ..target import (
    walk_internal_targets,
    walk_units_targets,
)

from ..config import (
    UnitsConfig,
)

from ..coverage_util import (
    coverage_context,
)

from ..data import (
    data_context,
)

from ..executor import (
    AllTargetsSkipped,
    Delegate,
    get_changes_filter,
    install_command_requirements,
    SUPPORTED_PYTHON_VERSIONS,
)


def command_units(args):
    """
    :type args: UnitsConfig
    """
    handle_layout_messages(data_context().content.unit_messages)

    changes = get_changes_filter(args)
    require = args.require + changes
    include = walk_internal_targets(walk_units_targets(), args.include, args.exclude, require)

    paths = [target.path for target in include]
    remote_paths = [path for path in paths
                    if is_subdir(path, data_context().content.unit_module_path)
                    or is_subdir(path, data_context().content.unit_module_utils_path)]

    if not paths:
        raise AllTargetsSkipped()

    if args.python and args.python in REMOTE_ONLY_PYTHON_VERSIONS and not remote_paths:
        raise AllTargetsSkipped()

    if args.delegate:
        raise Delegate(require=changes, exclude=args.exclude)

    version_commands = []

    available_versions = sorted(get_available_python_versions(list(SUPPORTED_PYTHON_VERSIONS)).keys())

    for version in SUPPORTED_PYTHON_VERSIONS:
        # run all versions unless version given, in which case run only that version
        if args.python and version != args.python_version:
            continue

        if not args.python and version not in available_versions:
            display.warning("Skipping unit tests on Python %s due to missing interpreter." % version)
            continue

        if args.requirements_mode != 'skip':
            install_command_requirements(args, version)

        env = ansible_environment(args)

        cmd = [
            'pytest',
            '--boxed',
            '-r', 'a',
            '-n', str(args.num_workers) if args.num_workers else 'auto',
            '--color',
            'yes' if args.color else 'no',
            '-p', 'no:cacheprovider',
            '-c', os.path.join(ANSIBLE_TEST_DATA_ROOT, 'pytest.ini'),
            '--junit-xml', os.path.join(ResultType.JUNIT.path, 'python%s-units.xml' % version),
        ]

        if not data_context().content.collection:
            cmd.append('--durations=25')

        if version != '2.6':
            # added in pytest 4.5.0, which requires python 2.7+
            cmd.append('--strict-markers')

        plugins = []

        if args.coverage:
            plugins.append('ansible_pytest_coverage')

        if data_context().content.collection:
            plugins.append('ansible_pytest_collections')

        if plugins:
            env['PYTHONPATH'] += ':%s' % os.path.join(ANSIBLE_TEST_DATA_ROOT, 'pytest/plugins')
            env['PYTEST_PLUGINS'] = ','.join(plugins)

        if args.collect_only:
            cmd.append('--collect-only')

        if args.verbosity:
            cmd.append('-' + ('v' * args.verbosity))

        if version in REMOTE_ONLY_PYTHON_VERSIONS:
            test_paths = remote_paths
        else:
            test_paths = paths

        if not test_paths:
            continue

        cmd.extend(test_paths)

        version_commands.append((version, cmd, env))

    if args.requirements_mode == 'only':
        sys.exit()

    for version, command, env in version_commands:
        check_pyyaml(args, version)

        display.info('Unit test with Python %s' % version)

        try:
            with coverage_context(args):
                intercept_command(args, command, target_name='units', env=env, python_version=version)
        except SubprocessError as ex:
            # pytest exits with status code 5 when all tests are skipped, which isn't an error for our use case
            if ex.status != 5:
                raise