diff options
Diffstat (limited to 'tests/unittests/test_net_activators.py')
| -rw-r--r-- | tests/unittests/test_net_activators.py | 232 |
1 files changed, 232 insertions, 0 deletions
diff --git a/tests/unittests/test_net_activators.py b/tests/unittests/test_net_activators.py new file mode 100644 index 00000000..db825c35 --- /dev/null +++ b/tests/unittests/test_net_activators.py @@ -0,0 +1,232 @@ +from collections import namedtuple +from unittest.mock import patch + +import pytest + +from cloudinit.net.activators import ( + DEFAULT_PRIORITY, + search_activator, + select_activator, +) +from cloudinit.net.activators import ( + IfUpDownActivator, + NetplanActivator, + NetworkManagerActivator +) +from cloudinit.net.network_state import parse_net_config_data +from cloudinit.safeyaml import load + + +V1_CONFIG = """\ +version: 1 +config: +- type: physical + name: eth0 +- type: physical + name: eth1 +""" + +V2_CONFIG = """\ +version: 2 +ethernets: + eth0: + dhcp4: true + eth1: + dhcp4: true +""" + +NETPLAN_CALL_LIST = [ + ((['netplan', 'apply'], ), {}), +] + + +@pytest.yield_fixture +def available_mocks(): + mocks = namedtuple('Mocks', 'm_which, m_file') + with patch('cloudinit.subp.which', return_value=True) as m_which: + with patch('os.path.isfile', return_value=True) as m_file: + yield mocks(m_which, m_file) + + +@pytest.yield_fixture +def unavailable_mocks(): + mocks = namedtuple('Mocks', 'm_which, m_file') + with patch('cloudinit.subp.which', return_value=False) as m_which: + with patch('os.path.isfile', return_value=False) as m_file: + yield mocks(m_which, m_file) + + +class TestSearchAndSelect: + def test_defaults(self, available_mocks): + resp = search_activator() + assert resp == DEFAULT_PRIORITY + + activator = select_activator() + assert activator == DEFAULT_PRIORITY[0] + + def test_priority(self, available_mocks): + new_order = [NetplanActivator, NetworkManagerActivator] + resp = search_activator(priority=new_order) + assert resp == new_order + + activator = select_activator(priority=new_order) + assert activator == new_order[0] + + def test_target(self, available_mocks): + search_activator(target='/tmp') + assert '/tmp' == available_mocks.m_which.call_args[1]['target'] + + select_activator(target='/tmp') + assert '/tmp' == available_mocks.m_which.call_args[1]['target'] + + @patch('cloudinit.net.activators.IfUpDownActivator.available', + return_value=False) + def test_first_not_available(self, m_available, available_mocks): + resp = search_activator() + assert resp == DEFAULT_PRIORITY[1:] + + resp = select_activator() + assert resp == DEFAULT_PRIORITY[1] + + def test_priority_not_exist(self, available_mocks): + with pytest.raises(ValueError): + search_activator(priority=['spam', 'eggs']) + with pytest.raises(ValueError): + select_activator(priority=['spam', 'eggs']) + + def test_none_available(self, unavailable_mocks): + resp = search_activator() + assert resp == [] + + with pytest.raises(RuntimeError): + select_activator() + + +IF_UP_DOWN_AVAILABLE_CALLS = [ + (('ifquery',), {'search': ['/sbin', '/usr/sbin'], 'target': None}), + (('ifup',), {'search': ['/sbin', '/usr/sbin'], 'target': None}), + (('ifdown',), {'search': ['/sbin', '/usr/sbin'], 'target': None}), +] + +NETPLAN_AVAILABLE_CALLS = [ + (('netplan',), {'search': ['/usr/sbin', '/sbin'], 'target': None}), +] + +NETWORK_MANAGER_AVAILABLE_CALLS = [ + (('nmcli',), {'target': None}), +] + + +@pytest.mark.parametrize('activator, available_calls', [ + (IfUpDownActivator, IF_UP_DOWN_AVAILABLE_CALLS), + (NetplanActivator, NETPLAN_AVAILABLE_CALLS), + (NetworkManagerActivator, NETWORK_MANAGER_AVAILABLE_CALLS), +]) +class TestActivatorsAvailable: + def test_available( + self, activator, available_calls, available_mocks + ): + activator.available() + assert available_mocks.m_which.call_args_list == available_calls + + +IF_UP_DOWN_BRING_UP_CALL_LIST = [ + ((['ifup', 'eth0'], ), {}), + ((['ifup', 'eth1'], ), {}), +] + +NETWORK_MANAGER_BRING_UP_CALL_LIST = [ + ((['nmcli', 'connection', 'up', 'ifname', 'eth0'], ), {}), + ((['nmcli', 'connection', 'up', 'ifname', 'eth1'], ), {}), +] + + +@pytest.mark.parametrize('activator, expected_call_list', [ + (IfUpDownActivator, IF_UP_DOWN_BRING_UP_CALL_LIST), + (NetplanActivator, NETPLAN_CALL_LIST), + (NetworkManagerActivator, NETWORK_MANAGER_BRING_UP_CALL_LIST), +]) +class TestActivatorsBringUp: + @patch('cloudinit.subp.subp', return_value=('', '')) + def test_bring_up_interface( + self, m_subp, activator, expected_call_list, available_mocks + ): + activator.bring_up_interface('eth0') + assert len(m_subp.call_args_list) == 1 + assert m_subp.call_args_list[0] == expected_call_list[0] + + @patch('cloudinit.subp.subp', return_value=('', '')) + def test_bring_up_interfaces( + self, m_subp, activator, expected_call_list, available_mocks + ): + activator.bring_up_interfaces(['eth0', 'eth1']) + assert expected_call_list == m_subp.call_args_list + + @patch('cloudinit.subp.subp', return_value=('', '')) + def test_bring_up_all_interfaces_v1( + self, m_subp, activator, expected_call_list, available_mocks + ): + network_state = parse_net_config_data(load(V1_CONFIG)) + activator.bring_up_all_interfaces(network_state) + for call in m_subp.call_args_list: + assert call in expected_call_list + + @patch('cloudinit.subp.subp', return_value=('', '')) + def test_bring_up_all_interfaces_v2( + self, m_subp, activator, expected_call_list, available_mocks + ): + network_state = parse_net_config_data(load(V2_CONFIG)) + activator.bring_up_all_interfaces(network_state) + for call in m_subp.call_args_list: + assert call in expected_call_list + + +IF_UP_DOWN_BRING_DOWN_CALL_LIST = [ + ((['ifdown', 'eth0'], ), {}), + ((['ifdown', 'eth1'], ), {}), +] + +NETWORK_MANAGER_BRING_DOWN_CALL_LIST = [ + ((['nmcli', 'connection', 'down', 'eth0'], ), {}), + ((['nmcli', 'connection', 'down', 'eth1'], ), {}), +] + + +@pytest.mark.parametrize('activator, expected_call_list', [ + (IfUpDownActivator, IF_UP_DOWN_BRING_DOWN_CALL_LIST), + (NetplanActivator, NETPLAN_CALL_LIST), + (NetworkManagerActivator, NETWORK_MANAGER_BRING_DOWN_CALL_LIST), +]) +class TestActivatorsBringDown: + @patch('cloudinit.subp.subp', return_value=('', '')) + def test_bring_down_interface( + self, m_subp, activator, expected_call_list, available_mocks + ): + activator.bring_down_interface('eth0') + assert len(m_subp.call_args_list) == 1 + assert m_subp.call_args_list[0] == expected_call_list[0] + + @patch('cloudinit.subp.subp', return_value=('', '')) + def test_bring_down_interfaces( + self, m_subp, activator, expected_call_list, available_mocks + ): + activator.bring_down_interfaces(['eth0', 'eth1']) + assert expected_call_list == m_subp.call_args_list + + @patch('cloudinit.subp.subp', return_value=('', '')) + def test_bring_down_all_interfaces_v1( + self, m_subp, activator, expected_call_list, available_mocks + ): + network_state = parse_net_config_data(load(V1_CONFIG)) + activator.bring_down_all_interfaces(network_state) + for call in m_subp.call_args_list: + assert call in expected_call_list + + @patch('cloudinit.subp.subp', return_value=('', '')) + def test_bring_down_all_interfaces_v2( + self, m_subp, activator, expected_call_list, available_mocks + ): + network_state = parse_net_config_data(load(V2_CONFIG)) + activator.bring_down_all_interfaces(network_state) + for call in m_subp.call_args_list: + assert call in expected_call_list |