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
|
import json
import logging
import os
import pytest
from pycloudlib.cloud import ImageType
from tests.integration_tests.clouds import ImageSpecification, IntegrationCloud
from tests.integration_tests.conftest import get_validated_source
from tests.integration_tests.instances import (
CloudInitSource,
IntegrationInstance,
)
from tests.integration_tests.util import verify_clean_log
LOG = logging.getLogger("integration_testing.test_ubuntu_advantage")
CLOUD_INIT_UA_TOKEN = os.environ.get("CLOUD_INIT_UA_TOKEN")
ATTACH_FALLBACK = """\
#cloud-config
ubuntu_advantage:
features:
disable_auto_attach: true
token: {token}
"""
ATTACH = """\
#cloud-config
ubuntu_advantage:
token: {token}
enable:
- esm-infra
"""
PRO_AUTO_ATTACH_DISABLED = """\
#cloud-config
ubuntu_advantage:
features:
disable_auto_attach: true
"""
PRO_DAEMON_DISABLED = """\
#cloud-config
# Disable UA daemon (only needed in GCE)
ubuntu_advantage:
features:
disable_auto_attach: true
bootcmd:
- sudo systemctl mask ubuntu-advantage.service
"""
AUTO_ATTACH_CUSTOM_SERVICES = """\
#cloud-config
ubuntu_advantage:
enable:
- livepatch
"""
def did_ua_service_noop(client: IntegrationInstance) -> bool:
ua_log = client.read_from_file("/var/log/ubuntu-advantage.log")
return (
"Skipping auto-attach and deferring to cloud-init to setup and"
" configure auto-attach" in ua_log
)
def is_attached(client: IntegrationInstance) -> bool:
status_resp = client.execute("sudo pro status --format json")
assert status_resp.ok
status = json.loads(status_resp.stdout)
return bool(status.get("attached"))
def get_services_status(client: IntegrationInstance) -> dict:
"""Creates a map of service -> is_enable.
pro status --format json contains a key with list of service objects like:
{
...
"services":[
{
"available":"yes",
"blocked_by":[
],
"description":"Common Criteria EAL2 Provisioning Packages",
"description_override":null,
"entitled":"yes",
"name":"cc-eal",
"status":"disabled",
"status_details":"CC EAL2 is not configured"
},
...
]
}
:return: Dict where the keys are ua service names and the values
are booleans representing if the service is enable or not.
"""
status_resp = client.execute("sudo pro status --format json")
assert status_resp.ok
status = json.loads(status_resp.stdout)
return {
svc["name"]: svc["status"] == "enabled" for svc in status["services"]
}
@pytest.mark.adhoc
@pytest.mark.ubuntu
@pytest.mark.skipif(
not CLOUD_INIT_UA_TOKEN, reason="CLOUD_INIT_UA_TOKEN env var not provided"
)
class TestUbuntuAdvantage:
@pytest.mark.user_data(ATTACH_FALLBACK.format(token=CLOUD_INIT_UA_TOKEN))
def test_valid_token(self, client: IntegrationInstance):
log = client.read_from_file("/var/log/cloud-init.log")
verify_clean_log(log)
assert is_attached(client)
@pytest.mark.user_data(ATTACH.format(token=CLOUD_INIT_UA_TOKEN))
def test_idempotency(self, client: IntegrationInstance):
log = client.read_from_file("/var/log/cloud-init.log")
verify_clean_log(log)
assert is_attached(client)
# Clean reboot to change instance-id and trigger cc_ua in next boot
assert client.execute("cloud-init clean --logs").ok
client.restart()
log = client.read_from_file("/var/log/cloud-init.log")
verify_clean_log(log)
assert is_attached(client)
def maybe_install_cloud_init(session_cloud: IntegrationCloud):
cfg_image_spec = ImageSpecification.from_os_image()
source = get_validated_source(session_cloud)
launch_kwargs = {
"image_id": session_cloud.cloud_instance.daily_image(
cfg_image_spec.image_id, image_type=ImageType.PRO
)
}
if source is CloudInitSource.NONE:
LOG.info(
"No need to customize cloud-init version. Return without spawning"
" an extra instance"
)
return launch_kwargs
user_data = (
PRO_DAEMON_DISABLED
if session_cloud.settings.PLATFORM == "gce"
else PRO_AUTO_ATTACH_DISABLED
)
with session_cloud.launch(
user_data=user_data,
launch_kwargs=launch_kwargs,
) as client:
# TODO: Re-enable this check after cloud images contain
# cloud-init 23.4.
# Explanation: We have to include something under
# user-data.ubuntu_advantage to skip the automatic auto-attach
# (driven by ua-auto-attach.service and/or ubuntu-advantage.service)
# while customizing the instance but in cloud-init < 23.4,
# user-data.ubuntu_advantage requires a token key.
# log = client.read_from_file("/var/log/cloud-init.log")
# verify_clean_log(log)
assert not is_attached(
client
), "Test precondition error. Instance is auto-attached."
if session_cloud.settings.PLATFORM == "gce":
LOG.info(
"Restore `ubuntu-advantage.service` original status for next"
" boot"
)
assert client.execute(
"sudo systemctl unmask ubuntu-advantage.service"
).ok
client.install_new_cloud_init(source)
client.destroy()
return {"image_id": session_cloud.snapshot_id}
@pytest.mark.azure
@pytest.mark.ec2
@pytest.mark.gce
@pytest.mark.ubuntu
class TestUbuntuAdvantagePro:
def test_custom_services(self, session_cloud: IntegrationCloud):
release = ImageSpecification.from_os_image().release
if release not in {"bionic", "focal", "jammy"}:
pytest.skip(f"Cannot run on non LTS release: {release}")
launch_kwargs = maybe_install_cloud_init(session_cloud)
with session_cloud.launch(
user_data=AUTO_ATTACH_CUSTOM_SERVICES,
launch_kwargs=launch_kwargs,
) as client:
log = client.read_from_file("/var/log/cloud-init.log")
verify_clean_log(log)
assert did_ua_service_noop(client)
assert is_attached(client)
services_status = get_services_status(client)
assert services_status.pop(
"livepatch"
), "livepatch expected to be enabled"
enabled_services = {
svc for svc, status in services_status.items() if status
}
assert (
not enabled_services
), f"Only livepatch must be enabled. Found: {enabled_services}"
|
