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
|
# This file is part of cloud-init. See LICENSE file for license information.
import logging
import os
import pytest
import sys
from contextlib import contextmanager
from tests.integration_tests import integration_settings
from tests.integration_tests.clouds import (
Ec2Cloud,
GceCloud,
AzureCloud,
OciCloud,
LxdContainerCloud,
)
log = logging.getLogger('integration_testing')
log.addHandler(logging.StreamHandler(sys.stdout))
log.setLevel(logging.INFO)
platforms = {
'ec2': Ec2Cloud,
'gce': GceCloud,
'azure': AzureCloud,
'oci': OciCloud,
'lxd_container': LxdContainerCloud,
}
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()
supported_platforms = set(all_platforms).intersection(
mark.name for mark in item.iter_markers())
current_platform = integration_settings.PLATFORM
if supported_platforms and current_platform not in supported_platforms:
pytest.skip('Cannot run on platform {}'.format(current_platform))
# 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.yield_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]()
yield cloud
cloud.destroy()
@pytest.fixture(scope='session', autouse=True)
def setup_image(session_cloud):
"""Setup the target environment with the correct version of cloud-init.
So we can launch instances / run tests with the correct image
"""
client = None
log.info('Setting up environment for %s', session_cloud.datasource)
if integration_settings.CLOUD_INIT_SOURCE == 'NONE':
pass # that was easy
elif integration_settings.CLOUD_INIT_SOURCE == 'IN_PLACE':
if session_cloud.datasource != 'lxd_container':
raise ValueError(
'IN_PLACE as CLOUD_INIT_SOURCE only works for LXD')
# The mount needs to happen after the instance is launched, so
# no further action needed here
elif integration_settings.CLOUD_INIT_SOURCE == 'PROPOSED':
client = session_cloud.launch()
client.install_proposed_image()
elif integration_settings.CLOUD_INIT_SOURCE.startswith('ppa:'):
client = session_cloud.launch()
client.install_ppa(integration_settings.CLOUD_INIT_SOURCE)
elif os.path.isfile(str(integration_settings.CLOUD_INIT_SOURCE)):
client = session_cloud.launch()
client.install_deb()
else:
raise ValueError(
'Invalid value for CLOUD_INIT_SOURCE setting: {}'.format(
integration_settings.CLOUD_INIT_SOURCE))
if client:
# 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')
@contextmanager
def _client(request, fixture_utils, session_cloud):
"""Fixture implementation for the client fixtures.
Launch the dynamic IntegrationClient instance using any provided
userdata, yield to the test, then cleanup
"""
user_data = fixture_utils.closest_marker_first_arg_or(
request, 'user_data', None)
name = fixture_utils.closest_marker_first_arg_or(
request, 'instance_name', None
)
launch_kwargs = {}
if name is not None:
launch_kwargs = {"name": name}
with session_cloud.launch(
user_data=user_data, launch_kwargs=launch_kwargs
) as instance:
yield instance
@pytest.yield_fixture
def client(request, fixture_utils, session_cloud):
"""Provide a client that runs for every test."""
with _client(request, fixture_utils, session_cloud) as client:
yield client
@pytest.yield_fixture(scope='module')
def module_client(request, fixture_utils, session_cloud):
"""Provide a client that runs once per module."""
with _client(request, fixture_utils, session_cloud) as client:
yield client
@pytest.yield_fixture(scope='class')
def class_client(request, fixture_utils, session_cloud):
"""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
|
