6
from cloudinit.sources import DataSourceMAAS
7
from cloudinit import url_helper
8
from ..helpers import TestCase, populate_dir
11
from unittest import mock
16
class TestMAASDataSource(TestCase):
19
super(TestMAASDataSource, self).setUp()
20
# Make a temp directoy for tests to use.
21
self.tmp = tempfile.mkdtemp()
22
self.addCleanup(shutil.rmtree, self.tmp)
24
def test_seed_dir_valid(self):
25
"""Verify a valid seeddir is read as such."""
27
data = {'instance-id': 'i-valid01',
28
'local-hostname': 'valid01-hostname',
29
'user-data': b'valid01-userdata',
30
'public-keys': 'ssh-rsa AAAAB3Nz...aC1yc2E= keyname'}
32
my_d = os.path.join(self.tmp, "valid")
33
populate_dir(my_d, data)
35
(userdata, metadata) = DataSourceMAAS.read_maas_seed_dir(my_d)
37
self.assertEqual(userdata, data['user-data'])
38
for key in ('instance-id', 'local-hostname'):
39
self.assertEqual(data[key], metadata[key])
41
# verify that 'userdata' is not returned as part of the metadata
42
self.assertFalse(('user-data' in metadata))
44
def test_seed_dir_valid_extra(self):
45
"""Verify extra files do not affect seed_dir validity."""
47
data = {'instance-id': 'i-valid-extra',
48
'local-hostname': 'valid-extra-hostname',
49
'user-data': b'valid-extra-userdata', 'foo': 'bar'}
51
my_d = os.path.join(self.tmp, "valid_extra")
52
populate_dir(my_d, data)
54
(userdata, metadata) = DataSourceMAAS.read_maas_seed_dir(my_d)
56
self.assertEqual(userdata, data['user-data'])
57
for key in ('instance-id', 'local-hostname'):
58
self.assertEqual(data[key], metadata[key])
60
# additional files should not just appear as keys in metadata atm
61
self.assertFalse(('foo' in metadata))
63
def test_seed_dir_invalid(self):
64
"""Verify that invalid seed_dir raises MAASSeedDirMalformed."""
66
valid = {'instance-id': 'i-instanceid',
67
'local-hostname': 'test-hostname', 'user-data': ''}
69
my_based = os.path.join(self.tmp, "valid_extra")
71
# missing 'userdata' file
72
my_d = "%s-01" % my_based
73
invalid_data = copy(valid)
74
del invalid_data['local-hostname']
75
populate_dir(my_d, invalid_data)
76
self.assertRaises(DataSourceMAAS.MAASSeedDirMalformed,
77
DataSourceMAAS.read_maas_seed_dir, my_d)
79
# missing 'instance-id'
80
my_d = "%s-02" % my_based
81
invalid_data = copy(valid)
82
del invalid_data['instance-id']
83
populate_dir(my_d, invalid_data)
84
self.assertRaises(DataSourceMAAS.MAASSeedDirMalformed,
85
DataSourceMAAS.read_maas_seed_dir, my_d)
87
def test_seed_dir_none(self):
88
"""Verify that empty seed_dir raises MAASSeedDirNone."""
90
my_d = os.path.join(self.tmp, "valid_empty")
91
self.assertRaises(DataSourceMAAS.MAASSeedDirNone,
92
DataSourceMAAS.read_maas_seed_dir, my_d)
94
def test_seed_dir_missing(self):
95
"""Verify that missing seed_dir raises MAASSeedDirNone."""
96
self.assertRaises(DataSourceMAAS.MAASSeedDirNone,
97
DataSourceMAAS.read_maas_seed_dir,
98
os.path.join(self.tmp, "nonexistantdirectory"))
100
def test_seed_url_valid(self):
101
"""Verify that valid seed_url is read as such."""
103
'meta-data/instance-id': 'i-instanceid',
104
'meta-data/local-hostname': 'test-hostname',
105
'meta-data/public-keys': 'test-hostname',
106
'user-data': b'foodata',
109
'meta-data/local-hostname',
110
'meta-data/instance-id',
111
'meta-data/public-keys',
114
my_seed = "http://example.com/xmeta"
115
my_ver = "1999-99-99"
116
my_headers = {'header1': 'value1', 'header2': 'value2'}
118
def my_headers_cb(url):
121
# Each time url_helper.readurl() is called, something different is
122
# returned based on the canned data above. We need to build up a list
123
# of side effect return values, which the mock will return. At the
124
# same time, we'll build up a list of expected call arguments for
125
# asserting after the code under test is run.
129
for key in valid_order:
130
resp = valid.get(key)
131
url = "%s/%s/%s" % (my_seed, my_ver, key)
133
mock.call(url, headers=None, timeout=mock.ANY,
134
data=mock.ANY, sec_between=mock.ANY,
135
ssl_details=mock.ANY, retries=mock.ANY,
136
headers_cb=my_headers_cb,
137
exception_cb=mock.ANY))
138
yield url_helper.StringResponse(resp)
140
# Now do the actual call of the code under test.
141
with mock.patch.object(url_helper, 'readurl',
142
side_effect=side_effect()) as mockobj:
143
userdata, metadata = DataSourceMAAS.read_maas_seed_url(
144
my_seed, version=my_ver)
146
self.assertEqual(b"foodata", userdata)
147
self.assertEqual(metadata['instance-id'],
148
valid['meta-data/instance-id'])
149
self.assertEqual(metadata['local-hostname'],
150
valid['meta-data/local-hostname'])
152
mockobj.has_calls(calls)
154
def test_seed_url_invalid(self):
155
"""Verify that invalid seed_url raises MAASSeedDirMalformed."""
158
def test_seed_url_missing(self):
159
"""Verify seed_url with no found entries raises MAASSeedDirNone."""
added
removed
tests/unittests/test_datasource/test_maas.py
