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
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
|
# This file is part of cloud-init. See LICENSE file for license information.
from cloudinit.sources import DataSourceIBMCloud as ibm
from cloudinit.tests import helpers as test_helpers
import base64
import copy
import json
import mock
from textwrap import dedent
D_PATH = "cloudinit.sources.DataSourceIBMCloud."
class TestIBMCloud(test_helpers.CiTestCase):
"""Test the datasource."""
def setUp(self):
super(TestIBMCloud, self).setUp()
pass
@mock.patch(D_PATH + "_is_xen", return_value=True)
@mock.patch(D_PATH + "_is_ibm_provisioning")
@mock.patch(D_PATH + "util.blkid")
class TestGetIBMPlatform(test_helpers.CiTestCase):
"""Test the get_ibm_platform helper."""
blkid_base = {
"/dev/xvda1": {
"DEVNAME": "/dev/xvda1", "LABEL": "cloudimg-bootfs",
"TYPE": "ext3"},
"/dev/xvda2": {
"DEVNAME": "/dev/xvda2", "LABEL": "cloudimg-rootfs",
"TYPE": "ext4"},
}
blkid_metadata_disk = {
"/dev/xvdh1": {
"DEVNAME": "/dev/xvdh1", "LABEL": "METADATA", "TYPE": "vfat",
"SEC_TYPE": "msdos", "UUID": "681B-8C5D",
"PARTUUID": "3d631e09-01"},
}
blkid_oscode_disk = {
"/dev/xvdh": {
"DEVNAME": "/dev/xvdh", "LABEL": "config-2", "TYPE": "vfat",
"SEC_TYPE": "msdos", "UUID": ibm.IBM_CONFIG_UUID}
}
def setUp(self):
self.blkid_metadata = copy.deepcopy(self.blkid_base)
self.blkid_metadata.update(copy.deepcopy(self.blkid_metadata_disk))
self.blkid_oscode = copy.deepcopy(self.blkid_base)
self.blkid_oscode.update(copy.deepcopy(self.blkid_oscode_disk))
def test_id_template_live_metadata(self, m_blkid, m_is_prov, _m_xen):
"""identify TEMPLATE_LIVE_METADATA."""
m_blkid.return_value = self.blkid_metadata
m_is_prov.return_value = False
self.assertEqual(
(ibm.Platforms.TEMPLATE_LIVE_METADATA, "/dev/xvdh1"),
ibm.get_ibm_platform())
def test_id_template_prov_metadata(self, m_blkid, m_is_prov, _m_xen):
"""identify TEMPLATE_PROVISIONING_METADATA."""
m_blkid.return_value = self.blkid_metadata
m_is_prov.return_value = True
self.assertEqual(
(ibm.Platforms.TEMPLATE_PROVISIONING_METADATA, "/dev/xvdh1"),
ibm.get_ibm_platform())
def test_id_template_prov_nodata(self, m_blkid, m_is_prov, _m_xen):
"""identify TEMPLATE_PROVISIONING_NODATA."""
m_blkid.return_value = self.blkid_base
m_is_prov.return_value = True
self.assertEqual(
(ibm.Platforms.TEMPLATE_PROVISIONING_NODATA, None),
ibm.get_ibm_platform())
def test_id_os_code(self, m_blkid, m_is_prov, _m_xen):
"""Identify OS_CODE."""
m_blkid.return_value = self.blkid_oscode
m_is_prov.return_value = False
self.assertEqual((ibm.Platforms.OS_CODE, "/dev/xvdh"),
ibm.get_ibm_platform())
def test_id_os_code_must_match_uuid(self, m_blkid, m_is_prov, _m_xen):
"""Test against false positive on openstack with non-ibm UUID."""
blkid = self.blkid_oscode
blkid["/dev/xvdh"]["UUID"] = "9999-9999"
m_blkid.return_value = blkid
m_is_prov.return_value = False
self.assertEqual((None, None), ibm.get_ibm_platform())
@mock.patch(D_PATH + "_read_system_uuid", return_value=None)
@mock.patch(D_PATH + "get_ibm_platform")
class TestReadMD(test_helpers.CiTestCase):
"""Test the read_datasource helper."""
template_md = {
"files": [],
"network_config": {"content_path": "/content/interfaces"},
"hostname": "ci-fond-ram",
"name": "ci-fond-ram",
"domain": "testing.ci.cloud-init.org",
"meta": {"dsmode": "net"},
"uuid": "8e636730-9f5d-c4a5-327c-d7123c46e82f",
"public_keys": {"1091307": "ssh-rsa AAAAB3NzaC1...Hw== ci-pubkey"},
}
oscode_md = {
"hostname": "ci-grand-gannet.testing.ci.cloud-init.org",
"name": "ci-grand-gannet",
"uuid": "2f266908-8e6c-4818-9b5c-42e9cc66a785",
"random_seed": "bm90LXJhbmRvbQo=",
"crypt_key": "ssh-rsa AAAAB3NzaC1yc2..n6z/",
"configuration_token": "eyJhbGciOi..M3ZA",
"public_keys": {"1091307": "ssh-rsa AAAAB3N..Hw== ci-pubkey"},
}
content_interfaces = dedent("""\
auto lo
iface lo inet loopback
auto eth0
allow-hotplug eth0
iface eth0 inet static
address 10.82.43.5
netmask 255.255.255.192
""")
userdata = b"#!/bin/sh\necho hi mom\n"
# meta.js file gets json encoded userdata as a list.
meta_js = '["#!/bin/sh\necho hi mom\n"]'
vendor_data = {
"cloud-init": "#!/bin/bash\necho 'root:$6$5ab01p1m1' | chpasswd -e"}
network_data = {
"links": [
{"id": "interface_29402281", "name": "eth0", "mtu": None,
"type": "phy", "ethernet_mac_address": "06:00:f1:bd:da:25"},
{"id": "interface_29402279", "name": "eth1", "mtu": None,
"type": "phy", "ethernet_mac_address": "06:98:5e:d0:7f:86"}
],
"networks": [
{"id": "network_109887563", "link": "interface_29402281",
"type": "ipv4", "ip_address": "10.82.43.2",
"netmask": "255.255.255.192",
"routes": [
{"network": "10.0.0.0", "netmask": "255.0.0.0",
"gateway": "10.82.43.1"},
{"network": "161.26.0.0", "netmask": "255.255.0.0",
"gateway": "10.82.43.1"}]},
{"id": "network_109887551", "link": "interface_29402279",
"type": "ipv4", "ip_address": "108.168.194.252",
"netmask": "255.255.255.248",
"routes": [
{"network": "0.0.0.0", "netmask": "0.0.0.0",
"gateway": "108.168.194.249"}]}
],
"services": [
{"type": "dns", "address": "10.0.80.11"},
{"type": "dns", "address": "10.0.80.12"}
],
}
sysuuid = '7f79ebf5-d791-43c3-a723-854e8389d59f'
def _get_expected_metadata(self, os_md):
"""return expected 'metadata' for data loaded from meta_data.json."""
os_md = copy.deepcopy(os_md)
renames = (
('hostname', 'local-hostname'),
('uuid', 'instance-id'),
('public_keys', 'public-keys'))
ret = {}
for osname, mdname in renames:
if osname in os_md:
ret[mdname] = os_md[osname]
if 'random_seed' in os_md:
ret['random_seed'] = base64.b64decode(os_md['random_seed'])
return ret
def test_provisioning_md(self, m_platform, m_sysuuid):
"""Provisioning env with a metadata disk should return None."""
m_platform.return_value = (
ibm.Platforms.TEMPLATE_PROVISIONING_METADATA, "/dev/xvdh")
self.assertIsNone(ibm.read_md())
def test_provisioning_no_metadata(self, m_platform, m_sysuuid):
"""Provisioning env with no metadata disk should return None."""
m_platform.return_value = (
ibm.Platforms.TEMPLATE_PROVISIONING_NODATA, None)
self.assertIsNone(ibm.read_md())
def test_provisioning_not_ibm(self, m_platform, m_sysuuid):
"""Provisioning env but not identified as IBM should return None."""
m_platform.return_value = (None, None)
self.assertIsNone(ibm.read_md())
def test_template_live(self, m_platform, m_sysuuid):
"""Template live environment should be identified."""
tmpdir = self.tmp_dir()
m_platform.return_value = (
ibm.Platforms.TEMPLATE_LIVE_METADATA, tmpdir)
m_sysuuid.return_value = self.sysuuid
test_helpers.populate_dir(tmpdir, {
'openstack/latest/meta_data.json': json.dumps(self.template_md),
'openstack/latest/user_data': self.userdata,
'openstack/content/interfaces': self.content_interfaces,
'meta.js': self.meta_js})
ret = ibm.read_md()
self.assertEqual(ibm.Platforms.TEMPLATE_LIVE_METADATA,
ret['platform'])
self.assertEqual(tmpdir, ret['source'])
self.assertEqual(self.userdata, ret['userdata'])
self.assertEqual(self._get_expected_metadata(self.template_md),
ret['metadata'])
self.assertEqual(self.sysuuid, ret['system-uuid'])
def test_os_code_live(self, m_platform, m_sysuuid):
"""Verify an os_code metadata path."""
tmpdir = self.tmp_dir()
m_platform.return_value = (ibm.Platforms.OS_CODE, tmpdir)
netdata = json.dumps(self.network_data)
test_helpers.populate_dir(tmpdir, {
'openstack/latest/meta_data.json': json.dumps(self.oscode_md),
'openstack/latest/user_data': self.userdata,
'openstack/latest/vendor_data.json': json.dumps(self.vendor_data),
'openstack/latest/network_data.json': netdata,
})
ret = ibm.read_md()
self.assertEqual(ibm.Platforms.OS_CODE, ret['platform'])
self.assertEqual(tmpdir, ret['source'])
self.assertEqual(self.userdata, ret['userdata'])
self.assertEqual(self._get_expected_metadata(self.oscode_md),
ret['metadata'])
def test_os_code_live_no_userdata(self, m_platform, m_sysuuid):
"""Verify os_code without user-data."""
tmpdir = self.tmp_dir()
m_platform.return_value = (ibm.Platforms.OS_CODE, tmpdir)
test_helpers.populate_dir(tmpdir, {
'openstack/latest/meta_data.json': json.dumps(self.oscode_md),
'openstack/latest/vendor_data.json': json.dumps(self.vendor_data),
})
ret = ibm.read_md()
self.assertEqual(ibm.Platforms.OS_CODE, ret['platform'])
self.assertEqual(tmpdir, ret['source'])
self.assertIsNone(ret['userdata'])
self.assertEqual(self._get_expected_metadata(self.oscode_md),
ret['metadata'])
# vi: ts=4 expandtab
|
