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
|
# This file is part of cloud-init. See LICENSE file for license information.
import logging
import os
import uuid
from enum import Enum
from tempfile import NamedTemporaryFile
from pycloudlib.instance import BaseInstance
from pycloudlib.result import Result
from tests.integration_tests import integration_settings
from tests.integration_tests.decorators import retry
try:
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from tests.integration_tests.clouds import ( # noqa: F401
IntegrationCloud,
)
except ImportError:
pass
log = logging.getLogger("integration_testing")
def _get_tmp_path():
tmp_filename = str(uuid.uuid4())
return "/var/tmp/{}.tmp".format(tmp_filename)
class CloudInitSource(Enum):
"""Represents the cloud-init image source setting as a defined value.
Values here represent all possible values for CLOUD_INIT_SOURCE in
tests/integration_tests/integration_settings.py. See that file for an
explanation of these values. If the value set there can't be parsed into
one of these values, an exception will be raised
"""
NONE = 1
IN_PLACE = 2
PROPOSED = 3
PPA = 4
DEB_PACKAGE = 5
UPGRADE = 6
def installs_new_version(self):
return self.name not in [self.NONE.name, self.IN_PLACE.name]
class IntegrationInstance:
def __init__(
self,
cloud: "IntegrationCloud",
instance: BaseInstance,
settings=integration_settings,
):
self.cloud = cloud
self.instance = instance
self.settings = settings
self._ip = ""
def destroy(self):
self.instance.delete()
def restart(self):
"""Restart this instance (via cloud mechanism) and wait for boot.
This wraps pycloudlib's `BaseInstance.restart`
"""
log.info("Restarting instance and waiting for boot")
self.instance.restart()
def execute(self, command, *, use_sudo=True) -> Result:
if self.instance.username == "root" and use_sudo is False:
raise RuntimeError("Root user cannot run unprivileged")
return self.instance.execute(command, use_sudo=use_sudo)
def pull_file(self, remote_path, local_path):
# First copy to a temporary directory because of permissions issues
tmp_path = _get_tmp_path()
self.instance.execute("cp {} {}".format(str(remote_path), tmp_path))
self.instance.pull_file(tmp_path, str(local_path))
def push_file(self, local_path, remote_path):
# First push to a temporary directory because of permissions issues
tmp_path = _get_tmp_path()
self.instance.push_file(str(local_path), tmp_path)
assert self.execute("mv {} {}".format(tmp_path, str(remote_path))).ok
def read_from_file(self, remote_path) -> str:
result = self.execute("cat {}".format(remote_path))
if result.failed:
# TODO: Raise here whatever pycloudlib raises when it has
# a consistent error response
raise IOError(
"Failed reading remote file via cat: {}\n"
"Return code: {}\n"
"Stderr: {}\n"
"Stdout: {}".format(
remote_path,
result.return_code,
result.stderr,
result.stdout,
)
)
return result.stdout
def write_to_file(self, remote_path, contents: str):
# Writes file locally and then pushes it rather
# than writing the file directly on the instance
with NamedTemporaryFile("w", delete=False) as tmp_file:
tmp_file.write(contents)
try:
self.push_file(tmp_file.name, remote_path)
finally:
os.unlink(tmp_file.name)
def snapshot(self):
image_id = self.cloud.snapshot(self.instance)
log.info("Created new image: %s", image_id)
return image_id
def install_new_cloud_init(
self,
source: CloudInitSource,
take_snapshot=True,
clean=True,
):
if source == CloudInitSource.DEB_PACKAGE:
self.install_deb()
elif source == CloudInitSource.PPA:
self.install_ppa()
elif source == CloudInitSource.PROPOSED:
self.install_proposed_image()
elif source == CloudInitSource.UPGRADE:
self.upgrade_cloud_init()
else:
raise RuntimeError(
"Specified to install {} which isn't supported here".format(
source
)
)
version = self.execute("cloud-init -v").split()[-1]
log.info("Installed cloud-init version: %s", version)
if clean:
self.instance.clean()
if take_snapshot:
snapshot_id = self.snapshot()
self.cloud.snapshot_id = snapshot_id
# assert with retry because we can compete with apt already running in the
# background and get: E: Could not get lock /var/lib/apt/lists/lock - open
# (11: Resource temporarily unavailable)
@retry(tries=30, delay=1)
def install_proposed_image(self):
log.info("Installing proposed image")
assert self.execute(
'echo deb "http://archive.ubuntu.com/ubuntu '
'$(lsb_release -sc)-proposed main" >> '
"/etc/apt/sources.list.d/proposed.list"
).ok
assert self.execute("apt-get update -q").ok
assert self.execute("apt-get install -qy cloud-init").ok
@retry(tries=30, delay=1)
def install_ppa(self):
log.info("Installing PPA")
assert self.execute(
"add-apt-repository {} -y".format(self.settings.CLOUD_INIT_SOURCE)
).ok
assert self.execute("apt-get update -q").ok
assert self.execute("apt-get install -qy cloud-init").ok
@retry(tries=30, delay=1)
def install_deb(self):
log.info("Installing deb package")
deb_path = integration_settings.CLOUD_INIT_SOURCE
deb_name = os.path.basename(deb_path)
remote_path = "/var/tmp/{}".format(deb_name)
self.push_file(
local_path=integration_settings.CLOUD_INIT_SOURCE,
remote_path=remote_path,
)
assert self.execute("dpkg -i {path}".format(path=remote_path)).ok
@retry(tries=30, delay=1)
def upgrade_cloud_init(self):
log.info("Upgrading cloud-init to latest version in archive")
assert self.execute("apt-get update -q").ok
assert self.execute("apt-get install -qy cloud-init").ok
def ip(self) -> str:
if self._ip:
return self._ip
try:
self._ip = self.instance.ip
except NotImplementedError:
self._ip = "Unknown"
return self._ip
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
if not self.settings.KEEP_INSTANCE:
self.destroy()
else:
log.info("Keeping Instance, public ip: %s", self.ip())
|
