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
|
import logging
import os
import re
import subprocess
import unittest
import lsb_release
import yaml
# Setup the logger
log = logging.getLogger('AzureTest')
logging.basicConfig(format='%(asctime)s %(levelname)s - %(message)s')
log.setLevel(logging.INFO)
def run_cmd(cmd):
try:
process = subprocess.Popen(
cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
return process.communicate()
except Exception:
log.critical("run_cmd: execution error", exc_info=True)
raise
class TestConfig(unittest.TestCase):
def setUp(self):
super(TestConfig, self).setUp()
self.ci_unsupported = ['quantal']
distinfo = lsb_release.get_distro_information()
self.suite = distinfo['CODENAME']
def is_cloud_init_supported(self):
if self.suite in self.ci_unsupported:
return False
return True
def test_suite(self):
"""
Test that LSB functions work
"""
log.info("Confirming that suite was detected")
assert self.suite is not None
log.info("Running test suite against %s" % self.suite)
def test_is_disk_big(self):
"""
Confirm disk size is 30GB
"""
cmd = ['df', '/']
out, _ = run_cmd(cmd)
log.info("Confirming root is proper size")
for line in out.splitlines():
_line = line.split()
if _line[-1] == "/":
log.info("Root disk size detected at %s" % _line[-2])
block_count = _line[1]
log.info("Root disk block count is %s" % block_count)
assert int(block_count) > 29800000
def test_ephemeral(self):
"""
Test that ephemeral is mounted to the right place
"""
mount = None
mt_pts = ["/mnt", "/mnt/resource"]
with open('/proc/mounts', 'r') as f:
for line in f.readlines():
dev, mnt, fs, _, _, _ = line.split()
if mnt in mt_pts:
mount = mnt
assert mount is not None
if self.is_cloud_init_supported():
self.assertEquals(mount, "/mnt")
else:
self.assertEquals(mount, "/mnt/resource")
def test_sudo(self):
"""
Test that passwordless sudo works for cloud-init supported releases
"""
if not self.is_cloud_init_supported():
return
log.info("Testing ability to call sudo")
returncode = subprocess.call(["/usr/bin/sudo", "/usr/bin/id", "-u"])
log.info("Got %s as the UID" % returncode)
assert int(returncode) == 0
def test_pkg_list(self):
"""
Confirm that required packages are installed
"""
log.info("Checking for dpkg output")
cmd = ['/usr/bin/dpkg-query', '-W']
(out, _err) = run_cmd(cmd)
log.info("Checking to make sure that output is good")
if not out:
log.critical(_err)
log.critical("dpkg-query failed")
assert out is not None
# Look for the kernel packages extra
packages = ['walinuxagent']
# Where are the KVP daemon tools?
if 'precise' in self.suite:
packages.append('linux-image-hwe-generic')
packages.append('linux-cloud-tools-generic-lts-trusty')
elif 'trusty' in self.suite:
packages.append('linux-image-virtual-lts-utopic')
packages.append('linux-lts-utopic-cloud-tools-common')
else:
packages.append('linux-cloud-tools-virtual')
log.info("Checking on installed packages")
for pkg in packages:
if pkg in out.decode():
log.info("Found package %s", pkg)
else:
log.critical("package {} was not found".format(pkg))
raise Exception("did not find kernel pkg {}".format(pkg))
def test_archive(self):
"""
Confirm archive configuration
Regular archive should have azure.archive.ubuntu.com/ubuntu
Security archive should have security.ubuntu.com/ubuntu
"""
consider_re = re.compile("^deb.*")
sec_arch = "%s-security" % self.suite
archive_url = "http://azure.archive.ubuntu.com/ubuntu"
security_url = "http://security.ubuntu.com/ubuntu"
log.info("Parsing /etc/apt/sources.lists")
with open('/etc/apt/sources.list', 'r') as sources:
for line in sources.readlines():
if not consider_re.match(line):
continue
_line = line.split()
if sec_arch in _line[2]:
log.info("Found security archive at %s" % _line[1])
assert security_url in _line[1]
else:
log.info("Found archive at %s" % _line[1])
assert archive_url in _line[1]
log.info("Finished archive parsing")
def test_cloud_init_datasources(self):
"""
Confirm Azure datasource supported releases
Should not exist: /var/lib/cloud/seed/nocloud-net/user-data
Should not exist: /var/lib/cloud/seed/nocloud-net/meta-data
File attributes for /etc/cloud/cloud.cfg.d/90_dpkg.cfg
Datasource should HAVE Azure in list
Datasource should NOT HAVE NoCloud and ConfigDrive
"""
if not self.is_cloud_init_supported():
log.info("Skipping test for cloud-init datasource")
return
log.info("Reading YAML of /etc/cloud/cloud.cfg.d/90_dpkg.cfg in")
cfg = None
with open('/etc/cloud/cloud.cfg.d/90_dpkg.cfg', 'r') as _cfg:
cfg = yaml.load(_cfg)
log.info("Checking that YAML read-in properly")
assert cfg is not None
log.info("Checking that datasource is a list")
assert isinstance(cfg['datasource_list'], list)
log.info("Checking that Azure is in datasource")
assert 'Azure' in cfg['datasource_list']
log.info("Making sure that legacy user-data is NOT present")
assert not os.path.exists('/var/lib/cloud/seed/nocloud-net/user-data')
log.info("Making sure that legacy meta-data is NOT present")
assert not os.path.exists('/var/lib/cloud/seed/nocloud-net/meta-data')
log.info("Checking that /tmp/cloud-init.done exists")
assert os.path.exists("/tmp/cloud-init.done")
log.info("Tests passed for cloud-init datasource")
def test_non_cloud_init_datasource(self):
"""
Confirm nocloud datasource releases
Should exist: /var/lib/cloud/seed/nocloud-net/user-data
Should exist: /var/lib/cloud/seed/nocloud-net/meta-data
File attributes for /etc/cloud/cloud.cfg.d/90_dpkg.cfg
Datasource should not have Azure in list
Datasource should have NoCloud and ConfigDrive
"""
if self.is_cloud_init_supported():
return
log.info("Reading YAML of /etc/cloud/cloud.cfg.d/90_dpkg.cfg in")
cfg = None
with open('/etc/cloud/cloud.cfg.d/90_dpkg.cfg', 'r') as _cfg:
cfg = yaml.load(_cfg)
log.info("Testing that datasource YAML read-in")
assert cfg is not None
log.info("Making sure that datasource is a list")
assert isinstance(cfg['datasource_list'], list)
log.info("Making sure that Azure is NOT in datasource list")
assert 'Azure' not in cfg['datasource_list']
log.info("Making sure that ConfigDrive is in datasource")
assert 'ConfigDrive' in cfg['datasource_list']
log.info("Mkaing sure that NoCloud is in datasource")
assert 'NoCloud' in cfg['datasource_list']
log.info("Checking to make sure that legacy user-data is present")
assert os.path.exists('/var/lib/cloud/seed/nocloud-net/user-data')
log.info("Checking to make sure that legacy meta-data is present")
assert os.path.exists('/var/lib/cloud/seed/nocloud-net/meta-data')
|