1
from tempfile import mkdtemp
2
from shutil import rmtree
4
2
from StringIO import StringIO
5
3
from copy import copy
6
from cloudinit.DataSourceMAAS import (
5
from cloudinit import util
6
from cloudinit import url_helper
7
from cloudinit.sources import DataSourceMAAS
12
9
from mocker import MockerTestCase
18
15
super(TestMAASDataSource, self).setUp()
19
16
# Make a temp directoy for tests to use.
20
self.tmp = mkdtemp(prefix="unittest_")
23
super(TestMAASDataSource, self).tearDown()
24
# Clean up temp directory
17
self.tmp = self.makeDir()
27
19
def test_seed_dir_valid(self):
28
20
"""Verify a valid seeddir is read as such"""
35
27
my_d = os.path.join(self.tmp, "valid")
36
28
populate_dir(my_d, data)
38
(userdata, metadata) = read_maas_seed_dir(my_d)
30
(userdata, metadata) = DataSourceMAAS.read_maas_seed_dir(my_d)
40
32
self.assertEqual(userdata, data['user-data'])
41
33
for key in ('instance-id', 'local-hostname'):
54
46
my_d = os.path.join(self.tmp, "valid_extra")
55
47
populate_dir(my_d, data)
57
(userdata, metadata) = read_maas_seed_dir(my_d)
49
(userdata, metadata) = DataSourceMAAS.read_maas_seed_dir(my_d)
59
51
self.assertEqual(userdata, data['user-data'])
60
52
for key in ('instance-id', 'local-hostname'):
76
68
invalid_data = copy(valid)
77
69
del invalid_data['local-hostname']
78
70
populate_dir(my_d, invalid_data)
79
self.assertRaises(MAASSeedDirMalformed, read_maas_seed_dir, my_d)
71
self.assertRaises(DataSourceMAAS.MAASSeedDirMalformed,
72
DataSourceMAAS.read_maas_seed_dir, my_d)
81
74
# missing 'instance-id'
82
75
my_d = "%s-02" % my_based
83
76
invalid_data = copy(valid)
84
77
del invalid_data['instance-id']
85
78
populate_dir(my_d, invalid_data)
86
self.assertRaises(MAASSeedDirMalformed, read_maas_seed_dir, my_d)
79
self.assertRaises(DataSourceMAAS.MAASSeedDirMalformed,
80
DataSourceMAAS.read_maas_seed_dir, my_d)
88
82
def test_seed_dir_none(self):
89
83
"""Verify that empty seed_dir raises MAASSeedDirNone"""
91
85
my_d = os.path.join(self.tmp, "valid_empty")
92
self.assertRaises(MAASSeedDirNone, read_maas_seed_dir, my_d)
86
self.assertRaises(DataSourceMAAS.MAASSeedDirNone,
87
DataSourceMAAS.read_maas_seed_dir, my_d)
94
89
def test_seed_dir_missing(self):
95
90
"""Verify that missing seed_dir raises MAASSeedDirNone"""
96
self.assertRaises(MAASSeedDirNone, read_maas_seed_dir,
91
self.assertRaises(DataSourceMAAS.MAASSeedDirNone,
92
DataSourceMAAS.read_maas_seed_dir,
97
93
os.path.join(self.tmp, "nonexistantdirectory"))
99
95
def test_seed_url_valid(self):
102
98
'meta-data/local-hostname': 'test-hostname',
103
99
'meta-data/public-keys': 'test-hostname',
104
100
'user-data': 'foodata'}
102
'meta-data/local-hostname',
103
'meta-data/instance-id',
104
'meta-data/public-keys',
106
107
my_seed = "http://example.com/xmeta"
107
108
my_ver = "1999-99-99"
108
109
my_headers = {'header1': 'value1', 'header2': 'value2'}
110
111
def my_headers_cb(url):
113
mock_request = self.mocker.replace("urllib2.Request",
115
mock_urlopen = self.mocker.replace("urllib2.urlopen",
118
for (key, val) in valid.iteritems():
119
mock_request("%s/%s/%s" % (my_seed, my_ver, key),
120
data=None, headers=my_headers)
122
self.mocker.result("fake-request-%s" % key)
123
mock_urlopen("fake-request-%s" % key, timeout=None)
124
self.mocker.result(StringIO(val))
114
mock_request = self.mocker.replace(url_helper.readurl,
117
for key in valid_order:
118
url = "%s/%s/%s" % (my_seed, my_ver, key)
119
mock_request(url, headers=my_headers, timeout=None)
120
resp = valid.get(key)
121
self.mocker.result(url_helper.UrlResponse(200, resp))
126
122
self.mocker.replay()
128
(userdata, metadata) = read_maas_seed_url(my_seed,
124
(userdata, metadata) = DataSourceMAAS.read_maas_seed_url(my_seed,
129
125
header_cb=my_headers_cb, version=my_ver)
131
127
self.assertEqual("foodata", userdata)