25
class TestWalkerHandleHandler(MockerTestCase):
36
class TestWalkerHandleHandler(TestCase):
29
MockerTestCase.setUp(self)
39
super(TestWalkerHandleHandler, self).setUp()
40
tmpdir = tempfile.mkdtemp()
41
self.addCleanup(shutil.rmtree, tmpdir)
34
"handlerdir": self.makeDir(),
35
47
"handlers": helpers.ContentHandlers(),
38
50
self.expected_module_name = "part-handler-%03d" % (
39
51
self.data["handlercount"],)
40
52
expected_file_name = "%s.py" % self.expected_module_name
41
expected_file_fullname = os.path.join(self.data["handlerdir"],
53
self.expected_file_fullname = os.path.join(
54
self.data["handlerdir"], expected_file_name)
43
55
self.module_fake = FakeModule()
45
57
self.filename = None
46
58
self.payload = "dummy payload"
48
# Mock the write_file function
49
write_file_mock = self.mocker.replace(util.write_file,
51
write_file_mock(expected_file_fullname, self.payload, 0600)
60
# Mock the write_file() function. We'll assert that it got called as
61
# expected in each of the individual tests.
62
resources = ExitStack()
63
self.addCleanup(resources.close)
64
self.write_file_mock = resources.enter_context(
65
mock.patch('cloudinit.util.write_file'))
53
67
def test_no_errors(self):
54
68
"""Payload gets written to file and added to C{pdata}."""
55
import_mock = self.mocker.replace(importer.import_module,
57
import_mock(self.expected_module_name)
58
self.mocker.result(self.module_fake)
61
handlers.walker_handle_handler(self.data, self.ctype, self.filename,
64
self.assertEqual(1, self.data["handlercount"])
69
with mock.patch('cloudinit.importer.import_module',
70
return_value=self.module_fake) as mockobj:
71
handlers.walker_handle_handler(self.data, self.ctype,
72
self.filename, self.payload)
73
mockobj.assert_called_with_once(self.expected_module_name)
74
self.write_file_mock.assert_called_with_once(
75
self.expected_file_fullname, self.payload, 0o600)
76
self.assertEqual(self.data['handlercount'], 1)
66
78
def test_import_error(self):
67
79
"""Module import errors are logged. No handler added to C{pdata}."""
68
import_mock = self.mocker.replace(importer.import_module,
70
import_mock(self.expected_module_name)
71
self.mocker.throw(ImportError())
74
handlers.walker_handle_handler(self.data, self.ctype, self.filename,
77
self.assertEqual(0, self.data["handlercount"])
80
with mock.patch('cloudinit.importer.import_module',
81
side_effect=ImportError) as mockobj:
82
handlers.walker_handle_handler(self.data, self.ctype,
83
self.filename, self.payload)
84
mockobj.assert_called_with_once(self.expected_module_name)
85
self.write_file_mock.assert_called_with_once(
86
self.expected_file_fullname, self.payload, 0o600)
87
self.assertEqual(self.data['handlercount'], 0)
79
89
def test_attribute_error(self):
80
90
"""Attribute errors are logged. No handler added to C{pdata}."""
81
import_mock = self.mocker.replace(importer.import_module,
83
import_mock(self.expected_module_name)
84
self.mocker.result(self.module_fake)
85
self.mocker.throw(AttributeError())
88
handlers.walker_handle_handler(self.data, self.ctype, self.filename,
91
self.assertEqual(0, self.data["handlercount"])
94
class TestHandlerHandlePart(MockerTestCase):
91
with mock.patch('cloudinit.importer.import_module',
92
side_effect=AttributeError,
93
return_value=self.module_fake) as mockobj:
94
handlers.walker_handle_handler(self.data, self.ctype,
95
self.filename, self.payload)
96
mockobj.assert_called_with_once(self.expected_module_name)
97
self.write_file_mock.assert_called_with_once(
98
self.expected_file_fullname, self.payload, 0o600)
99
self.assertEqual(self.data['handlercount'], 0)
102
class TestHandlerHandlePart(unittest.TestCase):
97
105
self.data = "fake data"
108
116
C{handle_part} is called without C{frequency} for
109
117
C{handler_version} == 1.
111
mod_mock = self.mocker.mock()
112
getattr(mod_mock, "frequency")
113
self.mocker.result(settings.PER_INSTANCE)
114
getattr(mod_mock, "handler_version")
115
self.mocker.result(1)
116
mod_mock.handle_part(self.data, self.ctype, self.filename,
120
handlers.run_part(mod_mock, self.data, self.filename,
121
self.payload, self.frequency, self.headers)
119
mod_mock = mock.Mock(frequency=settings.PER_INSTANCE,
121
handlers.run_part(mod_mock, self.data, self.filename, self.payload,
122
self.frequency, self.headers)
123
# Assert that the handle_part() method of the mock object got
124
# called with the expected arguments.
125
mod_mock.handle_part.assert_called_with_once(
126
self.data, self.ctype, self.filename, self.payload)
123
128
def test_normal_version_2(self):
125
130
C{handle_part} is called with C{frequency} for
126
131
C{handler_version} == 2.
128
mod_mock = self.mocker.mock()
129
getattr(mod_mock, "frequency")
130
self.mocker.result(settings.PER_INSTANCE)
131
getattr(mod_mock, "handler_version")
132
self.mocker.result(2)
133
mod_mock.handle_part(self.data, self.ctype, self.filename,
134
self.payload, self.frequency)
137
handlers.run_part(mod_mock, self.data, self.filename,
138
self.payload, self.frequency, self.headers)
133
mod_mock = mock.Mock(frequency=settings.PER_INSTANCE,
135
handlers.run_part(mod_mock, self.data, self.filename, self.payload,
136
self.frequency, self.headers)
137
# Assert that the handle_part() method of the mock object got
138
# called with the expected arguments.
139
mod_mock.handle_part.assert_called_with_once(
140
self.data, self.ctype, self.filename, self.payload)
140
142
def test_modfreq_per_always(self):
142
144
C{handle_part} is called regardless of frequency if nofreq is always.
144
146
self.frequency = "once"
145
mod_mock = self.mocker.mock()
146
getattr(mod_mock, "frequency")
147
self.mocker.result(settings.PER_ALWAYS)
148
getattr(mod_mock, "handler_version")
149
self.mocker.result(1)
150
mod_mock.handle_part(self.data, self.ctype, self.filename,
154
handlers.run_part(mod_mock, self.data, self.filename,
155
self.payload, self.frequency, self.headers)
147
mod_mock = mock.Mock(frequency=settings.PER_ALWAYS,
149
handlers.run_part(mod_mock, self.data, self.filename, self.payload,
150
self.frequency, self.headers)
151
# Assert that the handle_part() method of the mock object got
152
# called with the expected arguments.
153
mod_mock.handle_part.assert_called_with_once(
154
self.data, self.ctype, self.filename, self.payload)
157
156
def test_no_handle_when_modfreq_once(self):
158
157
"""C{handle_part} is not called if frequency is once."""
159
158
self.frequency = "once"
160
mod_mock = self.mocker.mock()
161
getattr(mod_mock, "frequency")
162
self.mocker.result(settings.PER_ONCE)
165
handlers.run_part(mod_mock, self.data, self.filename,
166
self.payload, self.frequency, self.headers)
159
mod_mock = mock.Mock(frequency=settings.PER_ONCE)
160
handlers.run_part(mod_mock, self.data, self.filename, self.payload,
161
self.frequency, self.headers)
162
# Assert that the handle_part() method of the mock object got
163
# called with the expected arguments.
164
mod_mock.handle_part.assert_called_with_once(
165
self.data, self.ctype, self.filename, self.payload)
168
167
def test_exception_is_caught(self):
169
168
"""Exceptions within C{handle_part} are caught and logged."""
170
mod_mock = self.mocker.mock()
171
getattr(mod_mock, "frequency")
172
self.mocker.result(settings.PER_INSTANCE)
173
getattr(mod_mock, "handler_version")
174
self.mocker.result(1)
175
mod_mock.handle_part(self.data, self.ctype, self.filename,
177
self.mocker.throw(Exception())
180
handlers.run_part(mod_mock, self.data, self.filename,
181
self.payload, self.frequency, self.headers)
184
class TestCmdlineUrl(MockerTestCase):
169
mod_mock = mock.Mock(frequency=settings.PER_INSTANCE,
171
handlers.run_part(mod_mock, self.data, self.filename, self.payload,
172
self.frequency, self.headers)
173
mod_mock.handle_part.side_effect = Exception
174
handlers.run_part(mod_mock, self.data, self.filename, self.payload,
175
self.frequency, self.headers)
176
mod_mock.handle_part.assert_called_with_once(
177
self.data, self.ctype, self.filename, self.payload)
180
class TestCmdlineUrl(unittest.TestCase):
185
181
def test_invalid_content(self):
186
182
url = "http://example.com/foo"
189
185
cmdline = "ro %s=%s bar=1" % (key, url)
191
mock_readurl = self.mocker.replace(url_helper.readurl,
193
mock_readurl(url, ARGS, KWARGS)
194
self.mocker.result(url_helper.StringResponse(payload))
197
self.assertEqual((key, url, None),
198
util.get_cmdline_url(names=[key], starts="xxxxxx",
187
with mock.patch('cloudinit.url_helper.readurl',
188
return_value=url_helper.StringResponse(payload)):
190
util.get_cmdline_url(names=[key], starts="xxxxxx",
201
194
def test_valid_content(self):
202
195
url = "http://example.com/foo"
204
197
payload = "xcloud-config\nmydata: foo\nbar: wark\n"
205
198
cmdline = "ro %s=%s bar=1" % (key, url)
207
mock_readurl = self.mocker.replace(url_helper.readurl,
209
mock_readurl(url, ARGS, KWARGS)
210
self.mocker.result(url_helper.StringResponse(payload))
213
self.assertEqual((key, url, payload),
214
util.get_cmdline_url(names=[key], starts="xcloud-config",
200
with mock.patch('cloudinit.url_helper.readurl',
201
return_value=url_helper.StringResponse(payload)):
203
util.get_cmdline_url(names=[key], starts="xcloud-config",
217
207
def test_no_key_found(self):
218
208
url = "http://example.com/foo"
220
210
cmdline = "ro %s=%s bar=1" % (key, url)
222
self.mocker.replace(url_helper.readurl, passthrough=False)
223
self.mocker.result(url_helper.StringResponse(""))
212
with mock.patch('cloudinit.url_helper.readurl',
213
return_value=url_helper.StringResponse('')):
215
util.get_cmdline_url(names=["does-not-appear"],
216
starts="#cloud-config", cmdline=cmdline),
226
self.assertEqual((None, None, None),
227
util.get_cmdline_url(names=["does-not-appear"],
228
starts="#cloud-config", cmdline=cmdline))
230
220
# vi: ts=4 expandtab
added
removed
tests/unittests/test__init__.py
