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
|
# Author: Mathieu Corbin <mathieu.corbin@exoscale.com>
# Author: Christopher Glass <christopher.glass@exoscale.com>
#
# This file is part of cloud-init. See LICENSE file for license information.
from cloudinit import helpers
from cloudinit.sources.DataSourceExoscale import (
API_VERSION,
DataSourceExoscale,
METADATA_URL,
get_password,
PASSWORD_SERVER_PORT,
read_metadata)
from cloudinit.tests.helpers import HttprettyTestCase, mock
from cloudinit import util
import httpretty
import os
import requests
TEST_PASSWORD_URL = "{}:{}/{}/".format(METADATA_URL,
PASSWORD_SERVER_PORT,
API_VERSION)
TEST_METADATA_URL = "{}/{}/meta-data/".format(METADATA_URL,
API_VERSION)
TEST_USERDATA_URL = "{}/{}/user-data".format(METADATA_URL,
API_VERSION)
@httpretty.activate
class TestDatasourceExoscale(HttprettyTestCase):
def setUp(self):
super(TestDatasourceExoscale, self).setUp()
self.tmp = self.tmp_dir()
self.password_url = TEST_PASSWORD_URL
self.metadata_url = TEST_METADATA_URL
self.userdata_url = TEST_USERDATA_URL
def test_password_saved(self):
"""The password is not set when it is not found
in the metadata service."""
httpretty.register_uri(httpretty.GET,
self.password_url,
body="saved_password")
self.assertFalse(get_password())
def test_password_empty(self):
"""No password is set if the metadata service returns
an empty string."""
httpretty.register_uri(httpretty.GET,
self.password_url,
body="")
self.assertFalse(get_password())
def test_password(self):
"""The password is set to what is found in the metadata
service."""
expected_password = "p@ssw0rd"
httpretty.register_uri(httpretty.GET,
self.password_url,
body=expected_password)
password = get_password()
self.assertEqual(expected_password, password)
def test_activate_removes_set_passwords_semaphore(self):
"""Allow set_passwords to run every boot by removing the semaphore."""
path = helpers.Paths({'cloud_dir': self.tmp})
sem_dir = self.tmp_path('instance/sem', dir=self.tmp)
util.ensure_dir(sem_dir)
sem_file = os.path.join(sem_dir, 'config_set_passwords')
with open(sem_file, 'w') as stream:
stream.write('')
ds = DataSourceExoscale({}, None, path)
ds.activate(None, None)
self.assertFalse(os.path.exists(sem_file))
def test_get_data(self):
"""The datasource conforms to expected behavior when supplied
full test data."""
path = helpers.Paths({'run_dir': self.tmp})
ds = DataSourceExoscale({}, None, path)
ds._is_platform_viable = lambda: True
expected_password = "p@ssw0rd"
expected_id = "12345"
expected_hostname = "myname"
expected_userdata = "#cloud-config"
httpretty.register_uri(httpretty.GET,
self.userdata_url,
body=expected_userdata)
httpretty.register_uri(httpretty.GET,
self.password_url,
body=expected_password)
httpretty.register_uri(httpretty.GET,
self.metadata_url,
body="instance-id\nlocal-hostname")
httpretty.register_uri(httpretty.GET,
"{}local-hostname".format(self.metadata_url),
body=expected_hostname)
httpretty.register_uri(httpretty.GET,
"{}instance-id".format(self.metadata_url),
body=expected_id)
self.assertTrue(ds._get_data())
self.assertEqual(ds.userdata_raw.decode("utf-8"), "#cloud-config")
self.assertEqual(ds.metadata, {"instance-id": expected_id,
"local-hostname": expected_hostname})
self.assertEqual(ds.get_config_obj(),
{'ssh_pwauth': True,
'password': expected_password,
'chpasswd': {
'expire': False,
}})
def test_get_data_saved_password(self):
"""The datasource conforms to expected behavior when saved_password is
returned by the password server."""
path = helpers.Paths({'run_dir': self.tmp})
ds = DataSourceExoscale({}, None, path)
ds._is_platform_viable = lambda: True
expected_answer = "saved_password"
expected_id = "12345"
expected_hostname = "myname"
expected_userdata = "#cloud-config"
httpretty.register_uri(httpretty.GET,
self.userdata_url,
body=expected_userdata)
httpretty.register_uri(httpretty.GET,
self.password_url,
body=expected_answer)
httpretty.register_uri(httpretty.GET,
self.metadata_url,
body="instance-id\nlocal-hostname")
httpretty.register_uri(httpretty.GET,
"{}local-hostname".format(self.metadata_url),
body=expected_hostname)
httpretty.register_uri(httpretty.GET,
"{}instance-id".format(self.metadata_url),
body=expected_id)
self.assertTrue(ds._get_data())
self.assertEqual(ds.userdata_raw.decode("utf-8"), "#cloud-config")
self.assertEqual(ds.metadata, {"instance-id": expected_id,
"local-hostname": expected_hostname})
self.assertEqual(ds.get_config_obj(), {})
def test_get_data_no_password(self):
"""The datasource conforms to expected behavior when no password is
returned by the password server."""
path = helpers.Paths({'run_dir': self.tmp})
ds = DataSourceExoscale({}, None, path)
ds._is_platform_viable = lambda: True
expected_answer = ""
expected_id = "12345"
expected_hostname = "myname"
expected_userdata = "#cloud-config"
httpretty.register_uri(httpretty.GET,
self.userdata_url,
body=expected_userdata)
httpretty.register_uri(httpretty.GET,
self.password_url,
body=expected_answer)
httpretty.register_uri(httpretty.GET,
self.metadata_url,
body="instance-id\nlocal-hostname")
httpretty.register_uri(httpretty.GET,
"{}local-hostname".format(self.metadata_url),
body=expected_hostname)
httpretty.register_uri(httpretty.GET,
"{}instance-id".format(self.metadata_url),
body=expected_id)
self.assertTrue(ds._get_data())
self.assertEqual(ds.userdata_raw.decode("utf-8"), "#cloud-config")
self.assertEqual(ds.metadata, {"instance-id": expected_id,
"local-hostname": expected_hostname})
self.assertEqual(ds.get_config_obj(), {})
@mock.patch('cloudinit.sources.DataSourceExoscale.get_password')
def test_read_metadata_when_password_server_unreachable(self, m_password):
"""The read_metadata function returns partial results in case the
password server (only) is unreachable."""
expected_id = "12345"
expected_hostname = "myname"
expected_userdata = "#cloud-config"
m_password.side_effect = requests.Timeout('Fake Connection Timeout')
httpretty.register_uri(httpretty.GET,
self.userdata_url,
body=expected_userdata)
httpretty.register_uri(httpretty.GET,
self.metadata_url,
body="instance-id\nlocal-hostname")
httpretty.register_uri(httpretty.GET,
"{}local-hostname".format(self.metadata_url),
body=expected_hostname)
httpretty.register_uri(httpretty.GET,
"{}instance-id".format(self.metadata_url),
body=expected_id)
result = read_metadata()
self.assertIsNone(result.get("password"))
self.assertEqual(result.get("user-data").decode("utf-8"),
expected_userdata)
def test_non_viable_platform(self):
"""The datasource fails fast when the platform is not viable."""
path = helpers.Paths({'run_dir': self.tmp})
ds = DataSourceExoscale({}, None, path)
ds._is_platform_viable = lambda: False
self.assertFalse(ds._get_data())
|
