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
|
from mocker import MockerTestCase, ANY, ARGS, KWARGS
import os
from cloudinit import (partwalker_handle_handler, handler_handle_part,
handler_register)
from cloudinit.util import write_file, logexc
class TestPartwalkerHandleHandler(MockerTestCase):
def setUp(self):
self.data = {
"handlercount": 0,
"frequency": "?",
"handlerdir": "?",
"handlers": [],
"data": None}
self.expected_module_name = "part-handler-%03d" % (
self.data["handlercount"],)
expected_file_name = "%s.py" % self.expected_module_name
expected_file_fullname = os.path.join(self.data["handlerdir"],
expected_file_name)
self.module_fake = "fake module handle"
self.ctype = None
self.filename = None
self.payload = "dummy payload"
# Mock the write_file function
write_file_mock = self.mocker.replace(write_file, passthrough=False)
write_file_mock(expected_file_fullname, self.payload, 0600)
def test_no_errors(self):
"""Payload gets written to file and added to C{pdata}."""
# Mock the __import__ builtin
import_mock = self.mocker.replace("__builtin__.__import__")
import_mock(self.expected_module_name)
self.mocker.result(self.module_fake)
# Mock the handle_register function
handle_reg_mock = self.mocker.replace(handler_register,
passthrough=False)
handle_reg_mock(self.module_fake, self.data["handlers"],
self.data["data"], self.data["frequency"])
# Activate mocks
self.mocker.replay()
partwalker_handle_handler(self.data, self.ctype, self.filename,
self.payload)
self.assertEqual(1, self.data["handlercount"])
def test_import_error(self):
"""Module import errors are logged. No handler added to C{pdata}"""
# Mock the __import__ builtin
import_mock = self.mocker.replace("__builtin__.__import__")
import_mock(self.expected_module_name)
self.mocker.throw(ImportError())
# Mock log function
logexc_mock = self.mocker.replace(logexc, passthrough=False)
logexc_mock(ANY)
# Mock the print_exc function
print_exc_mock = self.mocker.replace("traceback.print_exc",
passthrough=False)
print_exc_mock(ARGS, KWARGS)
# Activate mocks
self.mocker.replay()
partwalker_handle_handler(self.data, self.ctype, self.filename,
self.payload)
self.assertEqual(0, self.data["handlercount"])
def test_attribute_error(self):
"""Attribute errors are logged. No handler added to C{pdata}"""
# Mock the __import__ builtin
import_mock = self.mocker.replace("__builtin__.__import__")
import_mock(self.expected_module_name)
self.mocker.result(self.module_fake)
# Mock the handle_register function
handle_reg_mock = self.mocker.replace(handler_register,
passthrough=False)
handle_reg_mock(self.module_fake, self.data["handlers"],
self.data["data"], self.data["frequency"])
self.mocker.throw(AttributeError())
# Mock log function
logexc_mock = self.mocker.replace(logexc, passthrough=False)
logexc_mock(ANY)
# Mock the print_exc function
print_exc_mock = self.mocker.replace("traceback.print_exc",
passthrough=False)
print_exc_mock(ARGS, KWARGS)
# Activate mocks
self.mocker.replay()
partwalker_handle_handler(self.data, self.ctype, self.filename,
self.payload)
self.assertEqual(0, self.data["handlercount"])
class TestHandlerHandlePart(MockerTestCase):
def setUp(self):
self.data = "fake data"
self.ctype = "fake ctype"
self.filename = "fake filename"
self.payload = "fake payload"
self.frequency = "once-per-instance"
def test_normal_version_1(self):
"""
C{handle_part} is called without C{frequency} for
C{handler_version} == 1.
"""
# Build a mock part-handler module
mod_mock = self.mocker.mock()
getattr(mod_mock, "frequency")
self.mocker.result("once-per-instance")
getattr(mod_mock, "handler_version")
self.mocker.result(1)
mod_mock.handle_part(self.data, self.ctype, self.filename,
self.payload)
self.mocker.replay()
handler_handle_part(mod_mock, self.data, self.ctype, self.filename,
self.payload, self.frequency)
def test_normal_version_2(self):
"""
C{handle_part} is called with C{frequency} for
C{handler_version} == 2.
"""
# Build a mock part-handler module
mod_mock = self.mocker.mock()
getattr(mod_mock, "frequency")
self.mocker.result("once-per-instance")
getattr(mod_mock, "handler_version")
self.mocker.result(2)
mod_mock.handle_part(self.data, self.ctype, self.filename,
self.payload, self.frequency)
self.mocker.replay()
handler_handle_part(mod_mock, self.data, self.ctype, self.filename,
self.payload, self.frequency)
def test_modfreq_per_always(self):
"""
C{handle_part} is called regardless of frequency if nofreq is always.
"""
self.frequency = "once"
# Build a mock part-handler module
mod_mock = self.mocker.mock()
getattr(mod_mock, "frequency")
self.mocker.result("always")
getattr(mod_mock, "handler_version")
self.mocker.result(1)
mod_mock.handle_part(self.data, self.ctype, self.filename,
self.payload)
self.mocker.replay()
handler_handle_part(mod_mock, self.data, self.ctype, self.filename,
self.payload, self.frequency)
def test_no_handle_when_modfreq_once(self):
"""C{handle_part} is not called if frequency is once"""
self.frequency = "once"
# Build a mock part-handler module
mod_mock = self.mocker.mock()
getattr(mod_mock, "frequency")
self.mocker.result("once-per-instance")
self.mocker.replay()
handler_handle_part(mod_mock, self.data, self.ctype, self.filename,
self.payload, self.frequency)
def test_exception_is_caught(self):
"""Exceptions within C{handle_part} are caught and logged."""
# Build a mock part-handler module
mod_mock = self.mocker.mock()
getattr(mod_mock, "frequency")
self.mocker.result("once-per-instance")
getattr(mod_mock, "handler_version")
self.mocker.result(1)
mod_mock.handle_part(self.data, self.ctype, self.filename,
self.payload)
self.mocker.throw(Exception())
# Mock log function
logexc_mock = self.mocker.replace(logexc, passthrough=False)
logexc_mock(ANY)
# Mock the print_exc function
print_exc_mock = self.mocker.replace("traceback.print_exc",
passthrough=False)
print_exc_mock(ARGS, KWARGS)
self.mocker.replay()
handler_handle_part(mod_mock, self.data, self.ctype, self.filename,
self.payload, self.frequency)
|
