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
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
|
from collections import namedtuple
from unittest import mock
from unittest.mock import call
import pytest
from cloudinit.cmd.devel.hotplug_hook import handle_hotplug
from cloudinit.distros import Distro
from cloudinit.event import EventType
from cloudinit.net.activators import NetworkActivator
from cloudinit.net.network_state import NetworkState
from cloudinit.sources import DataSource
from cloudinit.stages import Init
hotplug_args = namedtuple("hotplug_args", "udevaction, subsystem, devpath")
FAKE_MAC = "11:22:33:44:55:66"
@pytest.fixture
def mocks():
m_init = mock.MagicMock(spec=Init)
m_distro = mock.MagicMock(spec=Distro)
m_datasource = mock.MagicMock(spec=DataSource)
m_datasource.distro = m_distro
m_init.datasource = m_datasource
m_init.fetch.return_value = m_datasource
read_sys_net = mock.patch(
"cloudinit.cmd.devel.hotplug_hook.read_sys_net_safe",
return_value=FAKE_MAC,
)
update_event_enabled = mock.patch(
"cloudinit.stages.update_event_enabled",
return_value=True,
)
m_network_state = mock.MagicMock(spec=NetworkState)
parse_net = mock.patch(
"cloudinit.cmd.devel.hotplug_hook.parse_net_config_data",
return_value=m_network_state,
)
m_activator = mock.MagicMock(spec=NetworkActivator)
select_activator = mock.patch(
"cloudinit.cmd.devel.hotplug_hook.activators.select_activator",
return_value=m_activator,
)
sleep = mock.patch("time.sleep")
read_sys_net.start()
update_event_enabled.start()
parse_net.start()
select_activator.start()
m_sleep = sleep.start()
yield namedtuple("mocks", "m_init m_network_state m_activator m_sleep")(
m_init=m_init,
m_network_state=m_network_state,
m_activator=m_activator,
m_sleep=m_sleep,
)
read_sys_net.stop()
update_event_enabled.stop()
parse_net.stop()
select_activator.stop()
sleep.stop()
class TestUnsupportedActions:
def test_unsupported_subsystem(self, mocks):
with pytest.raises(
Exception, match="cannot handle events for subsystem: not_real"
):
handle_hotplug(
hotplug_init=mocks.m_init,
devpath="/dev/fake",
subsystem="not_real",
udevaction="add",
)
def test_unsupported_udevaction(self, mocks):
with pytest.raises(ValueError, match="Unknown action: not_real"):
handle_hotplug(
hotplug_init=mocks.m_init,
devpath="/dev/fake",
udevaction="not_real",
subsystem="net",
)
class TestHotplug:
def test_succcessful_add(self, mocks):
init = mocks.m_init
mocks.m_network_state.iter_interfaces.return_value = [
{
"mac_address": FAKE_MAC,
}
]
handle_hotplug(
hotplug_init=init,
devpath="/dev/fake",
udevaction="add",
subsystem="net",
)
init.datasource.update_metadata_if_supported.assert_called_once_with(
[EventType.HOTPLUG]
)
mocks.m_activator.bring_up_interface.assert_called_once_with("fake")
mocks.m_activator.bring_down_interface.assert_not_called()
init._write_to_cache.assert_called_once_with()
def test_successful_remove(self, mocks):
init = mocks.m_init
mocks.m_network_state.iter_interfaces.return_value = [{}]
handle_hotplug(
hotplug_init=init,
devpath="/dev/fake",
udevaction="remove",
subsystem="net",
)
init.datasource.update_metadata_if_supported.assert_called_once_with(
[EventType.HOTPLUG]
)
mocks.m_activator.bring_down_interface.assert_called_once_with("fake")
mocks.m_activator.bring_up_interface.assert_not_called()
init._write_to_cache.assert_called_once_with()
def test_update_event_disabled(self, mocks, caplog):
init = mocks.m_init
with mock.patch(
"cloudinit.stages.update_event_enabled", return_value=False
):
handle_hotplug(
hotplug_init=init,
devpath="/dev/fake",
udevaction="remove",
subsystem="net",
)
assert "hotplug not enabled for event of type" in caplog.text
init.datasource.update_metadata_if_supported.assert_not_called()
mocks.m_activator.bring_up_interface.assert_not_called()
mocks.m_activator.bring_down_interface.assert_not_called()
init._write_to_cache.assert_not_called()
def test_update_metadata_failed(self, mocks):
mocks.m_init.datasource.update_metadata_if_supported.return_value = (
False
)
with pytest.raises(
RuntimeError, match="Datasource .* not updated for event hotplug"
):
handle_hotplug(
hotplug_init=mocks.m_init,
devpath="/dev/fake",
udevaction="remove",
subsystem="net",
)
def test_detect_hotplugged_device_not_detected_on_add(self, mocks):
mocks.m_network_state.iter_interfaces.return_value = [{}]
with pytest.raises(
RuntimeError,
match="Failed to detect {} in updated metadata".format(FAKE_MAC),
):
handle_hotplug(
hotplug_init=mocks.m_init,
devpath="/dev/fake",
udevaction="add",
subsystem="net",
)
def test_detect_hotplugged_device_detected_on_remove(self, mocks):
mocks.m_network_state.iter_interfaces.return_value = [
{
"mac_address": FAKE_MAC,
}
]
with pytest.raises(
RuntimeError, match="Failed to detect .* in updated metadata"
):
handle_hotplug(
hotplug_init=mocks.m_init,
devpath="/dev/fake",
udevaction="remove",
subsystem="net",
)
def test_apply_failed_on_add(self, mocks):
mocks.m_network_state.iter_interfaces.return_value = [
{
"mac_address": FAKE_MAC,
}
]
mocks.m_activator.bring_up_interface.return_value = False
with pytest.raises(
RuntimeError, match="Failed to bring up device: /dev/fake"
):
handle_hotplug(
hotplug_init=mocks.m_init,
devpath="/dev/fake",
udevaction="add",
subsystem="net",
)
def test_apply_failed_on_remove(self, mocks):
mocks.m_network_state.iter_interfaces.return_value = [{}]
mocks.m_activator.bring_down_interface.return_value = False
with pytest.raises(
RuntimeError, match="Failed to bring down device: /dev/fake"
):
handle_hotplug(
hotplug_init=mocks.m_init,
devpath="/dev/fake",
udevaction="remove",
subsystem="net",
)
def test_retry(self, mocks):
with pytest.raises(RuntimeError):
handle_hotplug(
hotplug_init=mocks.m_init,
devpath="/dev/fake",
udevaction="add",
subsystem="net",
)
assert mocks.m_sleep.call_count == 5
assert mocks.m_sleep.call_args_list == [
call(1),
call(3),
call(5),
call(10),
call(30),
]
|
