summaryrefslogtreecommitdiff
path: root/tests/integration_tests/conftest.py
blob: a90a5d49bfdf5b1f8a76814c1981bdb7af955a73 (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
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
# This file is part of cloud-init. See LICENSE file for license information.
import datetime
import functools
import logging
import os
import sys
from contextlib import contextmanager
from pathlib import Path
from tarfile import TarFile
from typing import Dict, Type

import pytest
from pycloudlib.lxd.instance import LXDInstance

from tests.integration_tests import integration_settings
from tests.integration_tests.clouds import (
    AzureCloud,
    Ec2Cloud,
    GceCloud,
    ImageSpecification,
    IntegrationCloud,
    LxdContainerCloud,
    LxdVmCloud,
    OciCloud,
    OpenstackCloud,
    _LxdIntegrationCloud,
)
from tests.integration_tests.instances import (
    CloudInitSource,
    IntegrationInstance,
)

log = logging.getLogger("integration_testing")
log.addHandler(logging.StreamHandler(sys.stdout))
log.setLevel(logging.INFO)

platforms: Dict[str, Type[IntegrationCloud]] = {
    "ec2": Ec2Cloud,
    "gce": GceCloud,
    "azure": AzureCloud,
    "oci": OciCloud,
    "lxd_container": LxdContainerCloud,
    "lxd_vm": LxdVmCloud,
    "openstack": OpenstackCloud,
}
os_list = ["ubuntu"]

session_start_time = datetime.datetime.now().strftime("%y%m%d%H%M%S")


def pytest_runtest_setup(item):
    """Skip tests on unsupported clouds.

    A test can take any number of marks to specify the platforms it can
    run on. If a platform(s) is specified and we're not running on that
    platform, then skip the test. If platform specific marks are not
    specified, then we assume the test can be run anywhere.
    """
    all_platforms = platforms.keys()
    test_marks = [mark.name for mark in item.iter_markers()]
    supported_platforms = set(all_platforms).intersection(test_marks)
    current_platform = integration_settings.PLATFORM
    unsupported_message = "Cannot run on platform {}".format(current_platform)
    if "no_container" in test_marks:
        if "lxd_container" in test_marks:
            raise Exception(
                "lxd_container and no_container marks simultaneously set "
                "on test"
            )
        if current_platform == "lxd_container":
            pytest.skip(unsupported_message)
    if supported_platforms and current_platform not in supported_platforms:
        pytest.skip(unsupported_message)

    image = ImageSpecification.from_os_image()
    current_os = image.os
    supported_os_set = set(os_list).intersection(test_marks)
    if current_os and supported_os_set and current_os not in supported_os_set:
        pytest.skip("Cannot run on OS {}".format(current_os))
    if "unstable" in test_marks and not integration_settings.RUN_UNSTABLE:
        pytest.skip("Test marked unstable. Manually remove mark to run it")

    current_release = image.release
    if "not_{}".format(current_release) in test_marks:
        pytest.skip("Cannot run on release {}".format(current_release))


# disable_subp_usage is defined at a higher level, but we don't
# want it applied here
@pytest.fixture()
def disable_subp_usage(request):
    pass


@pytest.fixture(scope="session")
def session_cloud():
    if integration_settings.PLATFORM not in platforms.keys():
        raise ValueError(
            "{} is an invalid PLATFORM specified in settings. "
            "Must be one of {}".format(
                integration_settings.PLATFORM, list(platforms.keys())
            )
        )

    cloud = platforms[integration_settings.PLATFORM]()
    cloud.emit_settings_to_log()

    yield cloud

    cloud.destroy()


def get_validated_source(
    session_cloud: IntegrationCloud,
    source=integration_settings.CLOUD_INIT_SOURCE,
) -> CloudInitSource:
    if source == "NONE":
        return CloudInitSource.NONE
    elif source == "IN_PLACE":
        if session_cloud.datasource not in ["lxd_container", "lxd_vm"]:
            raise ValueError(
                "IN_PLACE as CLOUD_INIT_SOURCE only works for LXD"
            )
        return CloudInitSource.IN_PLACE
    elif source == "PROPOSED":
        return CloudInitSource.PROPOSED
    elif source.startswith("ppa:"):
        return CloudInitSource.PPA
    elif os.path.isfile(str(source)):
        return CloudInitSource.DEB_PACKAGE
    elif source == "UPGRADE":
        return CloudInitSource.UPGRADE
    raise ValueError(
        "Invalid value for CLOUD_INIT_SOURCE setting: {}".format(source)
    )


@pytest.fixture(scope="session")
def setup_image(session_cloud: IntegrationCloud, request):
    """Setup the target environment with the correct version of cloud-init.

    So we can launch instances / run tests with the correct image
    """

    source = get_validated_source(session_cloud)
    if not source.installs_new_version():
        return
    log.info("Setting up environment for %s", session_cloud.datasource)
    client = session_cloud.launch()
    client.install_new_cloud_init(source)
    # Even if we're keeping instances, we don't want to keep this
    # one around as it was just for image creation
    client.destroy()
    log.info("Done with environment setup")

    # For some reason a yield here raises a
    # ValueError: setup_image did not yield a value
    # during setup so use a finalizer instead.
    request.addfinalizer(session_cloud.delete_snapshot)


def _collect_logs(
    instance: IntegrationInstance, node_id: str, test_failed: bool
):
    """Collect logs from remote instance.

    Args:
        instance: The current IntegrationInstance to collect logs from
        node_id: The pytest representation of this test, E.g.:
            tests/integration_tests/test_example.py::TestExample.test_example
        test_failed: If test failed or not
    """
    if any(
        [
            integration_settings.COLLECT_LOGS == "NEVER",
            integration_settings.COLLECT_LOGS == "ON_ERROR"
            and not test_failed,
        ]
    ):
        return
    instance.execute(
        "cloud-init collect-logs -u -t /var/tmp/cloud-init.tar.gz"
    )
    node_id_path = Path(
        node_id.replace(
            ".py", ""
        )  # Having a directory with '.py' would be weird
        .replace("::", os.path.sep)  # Turn classes/tests into paths
        .replace("[", "-")  # For parametrized names
        .replace("]", "")  # For parameterized names
    )
    log_dir = (
        Path(integration_settings.LOCAL_LOG_PATH)
        / session_start_time
        / node_id_path
    )
    log.info("Writing logs to %s", log_dir)

    if not log_dir.exists():
        log_dir.mkdir(parents=True)

    # Add a symlink to the latest log output directory
    last_symlink = Path(integration_settings.LOCAL_LOG_PATH) / "last"
    if os.path.islink(last_symlink):
        os.unlink(last_symlink)
    os.symlink(log_dir.parent, last_symlink)

    tarball_path = log_dir / "cloud-init.tar.gz"
    try:
        instance.pull_file("/var/tmp/cloud-init.tar.gz", tarball_path)
    except Exception as e:
        log.error("Failed to pull logs: %s", e)
        return

    tarball = TarFile.open(str(tarball_path))
    tarball.extractall(path=str(log_dir))
    tarball_path.unlink()


@contextmanager
def _client(request, fixture_utils, session_cloud: IntegrationCloud):
    """Fixture implementation for the client fixtures.

    Launch the dynamic IntegrationClient instance using any provided
    userdata, yield to the test, then cleanup
    """
    getter = functools.partial(
        fixture_utils.closest_marker_first_arg_or, request, default=None
    )
    user_data = getter("user_data")
    name = getter("instance_name")
    lxd_config_dict = getter("lxd_config_dict")
    lxd_setup = getter("lxd_setup")
    lxd_use_exec = fixture_utils.closest_marker_args_or(
        request, "lxd_use_exec", None
    )

    launch_kwargs = {}
    if name is not None:
        launch_kwargs["name"] = name
    if lxd_config_dict is not None:
        if not isinstance(session_cloud, _LxdIntegrationCloud):
            pytest.skip("lxd_config_dict requires LXD")
        launch_kwargs["config_dict"] = lxd_config_dict
    if lxd_use_exec is not None:
        if not isinstance(session_cloud, _LxdIntegrationCloud):
            pytest.skip("lxd_use_exec requires LXD")
        launch_kwargs["execute_via_ssh"] = False
    local_launch_kwargs = {}
    if lxd_setup is not None:
        if not isinstance(session_cloud, _LxdIntegrationCloud):
            pytest.skip("lxd_setup requires LXD")
        local_launch_kwargs["lxd_setup"] = lxd_setup

    with session_cloud.launch(
        user_data=user_data, launch_kwargs=launch_kwargs, **local_launch_kwargs
    ) as instance:
        if lxd_use_exec is not None and isinstance(
            instance.instance, LXDInstance
        ):
            # Existing instances are not affected by the launch kwargs, so
            # ensure it here; we still need the launch kwarg so waiting works
            instance.instance.execute_via_ssh = False
        previous_failures = request.session.testsfailed
        yield instance
        test_failed = request.session.testsfailed - previous_failures > 0
        _collect_logs(instance, request.node.nodeid, test_failed)


@pytest.fixture
def client(request, fixture_utils, session_cloud, setup_image):
    """Provide a client that runs for every test."""
    with _client(request, fixture_utils, session_cloud) as client:
        yield client


@pytest.fixture(scope="module")
def module_client(request, fixture_utils, session_cloud, setup_image):
    """Provide a client that runs once per module."""
    with _client(request, fixture_utils, session_cloud) as client:
        yield client


@pytest.fixture(scope="class")
def class_client(request, fixture_utils, session_cloud, setup_image):
    """Provide a client that runs once per class."""
    with _client(request, fixture_utils, session_cloud) as client:
        yield client


def pytest_assertrepr_compare(op, left, right):
    """Custom integration test assertion explanations.

    See
    https://docs.pytest.org/en/stable/assert.html#defining-your-own-explanation-for-failed-assertions
    for pytest's documentation.
    """
    if op == "not in" and isinstance(left, str) and isinstance(right, str):
        # This stanza emits an improved assertion message if we're testing for
        # the presence of a string within a cloud-init log: it will report only
        # the specific lines containing the string (instead of the full log,
        # the default behaviour).
        potential_log_lines = right.splitlines()
        first_line = potential_log_lines[0]
        if "DEBUG" in first_line and "Cloud-init" in first_line:
            # We are looking at a cloud-init log, so just pick out the relevant
            # lines
            found_lines = [
                line for line in potential_log_lines if left in line
            ]
            return [
                '"{}" not in cloud-init.log string; unexpectedly found on'
                " these lines:".format(left)
            ] + found_lines


def pytest_configure(config):
    """Perform initial configuration, before the test runs start.

    This hook is only called if integration tests are being executed, so we can
    use it to configure defaults for integration testing that differ from the
    rest of the tests in the codebase.

    See
    https://docs.pytest.org/en/latest/reference.html#_pytest.hookspec.pytest_configure
    for pytest's documentation.
    """
    if "log_cli_level" in config.option and not config.option.log_cli_level:
        # If log_cli_level is available in this version of pytest and not set
        # to anything, set it to INFO.
        config.option.log_cli_level = "INFO"