hotplug unit tests

2 files changed, 222 insertions, 3 deletions

diff --git a/cloudinit/cmd/devel/hotplug_hook.py b/cloudinit/cmd/devel/hotplug_hook.py
index 51f618ae..92d50fc8 100644
--- a/cloudinit/cmd/devel/hotplug_hook.py
+++ b/cloudinit/cmd/devel/hotplug_hook.py
@@ -71,7 +71,7 @@ class UeventHandler(object):
         elif self.action == 'remove':
             detect_presence = False
         else:
-            raise ValueError('Cannot detect unknown action: %s' % self.action)
+            raise ValueError('Unknown action: %s' % self.action)
 
         if detect_presence != self.device_detected():
             raise RuntimeError(
@@ -178,10 +178,11 @@ def handle_hotplug(
             event_handler.success()
             break
         except Exception as e:
-            if attempt + 1 >= len(wait_times):
-                raise
             LOG.debug('Exception while processing hotplug event. %s' % e)
             time.sleep(wait)
+            last_exception = e
+    else:
+        raise last_exception  # type: ignore
 
 
 def handle_args(name, args):
diff --git a/tests/unittests/cmd/devel/test_hotplug_hook.py b/tests/unittests/cmd/devel/test_hotplug_hook.py
new file mode 100644
index 00000000..5e371a72
--- /dev/null
+++ b/tests/unittests/cmd/devel/test_hotplug_hook.py
@@ -0,0 +1,218 @@
+import pytest
+from collections import namedtuple
+from unittest import mock
+from unittest.mock import call
+
+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.yield_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
+    )
+
+    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()
+    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()
+    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
+        init.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(1), call(1), call(3), call(5)
+        ]