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
|
# This file is part of cloud-init. See LICENSE file for license information.
import logging
import os
from tempfile import NamedTemporaryFile
from pycloudlib.instance import BaseInstance
import cloudinit
from cloudinit.subp import subp
from tests.integration_tests import integration_settings
try:
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from tests.integration_tests.clouds import IntegrationCloud
except ImportError:
pass
log = logging.getLogger('integration_testing')
class IntegrationInstance:
use_sudo = True
def __init__(self, cloud: 'IntegrationCloud', instance: BaseInstance,
settings=integration_settings):
self.cloud = cloud
self.instance = instance
self.settings = settings
def emit_settings_to_log(self) -> None:
log.info(
"\n".join(
["Settings:"]
+ [
"{}={}".format(key, getattr(self.settings, key))
for key in sorted(self.settings.current_settings)
]
)
)
def destroy(self):
self.instance.delete()
def execute(self, command):
return self.instance.execute(command)
def pull_file(self, remote_file, local_file):
self.instance.pull_file(remote_file, local_file)
def push_file(self, local_path, remote_path):
self.instance.push_file(local_path, remote_path)
def read_from_file(self, remote_path) -> str:
tmp_file = NamedTemporaryFile('r')
self.pull_file(remote_path, tmp_file.name)
with tmp_file as f:
contents = f.read()
return contents
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):
return self.cloud.snapshot(self.instance)
def _install_new_cloud_init(self, remote_script):
self.execute(remote_script)
version = self.execute('cloud-init -v').split()[-1]
log.info('Installed cloud-init version: %s', version)
self.instance.clean()
image_id = self.snapshot()
log.info('Created new image: %s', image_id)
self.cloud.image_id = image_id
def install_proposed_image(self):
log.info('Installing proposed image')
remote_script = (
'{sudo} echo deb "http://archive.ubuntu.com/ubuntu '
'$(lsb_release -sc)-proposed main" | '
'{sudo} tee /etc/apt/sources.list.d/proposed.list\n'
'{sudo} apt-get update -q\n'
'{sudo} apt-get install -qy cloud-init'
).format(sudo='sudo' if self.use_sudo else '')
self._install_new_cloud_init(remote_script)
def install_ppa(self, repo):
log.info('Installing PPA')
remote_script = (
'{sudo} add-apt-repository {repo} -y && '
'{sudo} apt-get update -q && '
'{sudo} apt-get install -qy cloud-init'
).format(sudo='sudo' if self.use_sudo else '', repo=repo)
self._install_new_cloud_init(remote_script)
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)
remote_script = '{sudo} dpkg -i {path}'.format(
sudo='sudo' if self.use_sudo else '', path=remote_path)
self._install_new_cloud_init(remote_script)
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
if not self.settings.KEEP_INSTANCE:
self.destroy()
class IntegrationEc2Instance(IntegrationInstance):
pass
class IntegrationGceInstance(IntegrationInstance):
pass
class IntegrationAzureInstance(IntegrationInstance):
pass
class IntegrationOciInstance(IntegrationInstance):
pass
class IntegrationLxdContainerInstance(IntegrationInstance):
use_sudo = False
def __init__(self, cloud: 'IntegrationCloud', instance: BaseInstance,
settings=integration_settings):
super().__init__(cloud, instance, settings)
if self.settings.CLOUD_INIT_SOURCE == 'IN_PLACE':
self._mount_source()
def _mount_source(self):
command = (
'lxc config device add {name} host-cloud-init disk '
'source={cloudinit_path} '
'path=/usr/lib/python3/dist-packages/cloudinit'
).format(
name=self.instance.name, cloudinit_path=cloudinit.__path__[0])
subp(command.split())
|
