summaryrefslogtreecommitdiff
path: root/cloudinit/sources/helpers/tests/test_netlink.py
diff options
context:
space:
mode:
Diffstat (limited to 'cloudinit/sources/helpers/tests/test_netlink.py')
-rw-r--r--cloudinit/sources/helpers/tests/test_netlink.py373
1 files changed, 373 insertions, 0 deletions
diff --git a/cloudinit/sources/helpers/tests/test_netlink.py b/cloudinit/sources/helpers/tests/test_netlink.py
new file mode 100644
index 00000000..c2898a16
--- /dev/null
+++ b/cloudinit/sources/helpers/tests/test_netlink.py
@@ -0,0 +1,373 @@
+# Author: Tamilmani Manoharan <tamanoha@microsoft.com>
+#
+# This file is part of cloud-init. See LICENSE file for license information.
+
+from cloudinit.tests.helpers import CiTestCase, mock
+import socket
+import struct
+import codecs
+from cloudinit.sources.helpers.netlink import (
+ NetlinkCreateSocketError, create_bound_netlink_socket, read_netlink_socket,
+ read_rta_oper_state, unpack_rta_attr, wait_for_media_disconnect_connect,
+ OPER_DOWN, OPER_UP, OPER_DORMANT, OPER_LOWERLAYERDOWN, OPER_NOTPRESENT,
+ OPER_TESTING, OPER_UNKNOWN, RTATTR_START_OFFSET, RTM_NEWLINK, RTM_SETLINK,
+ RTM_GETLINK, MAX_SIZE)
+
+
+def int_to_bytes(i):
+ '''convert integer to binary: eg: 1 to \x01'''
+ hex_value = '{0:x}'.format(i)
+ hex_value = '0' * (len(hex_value) % 2) + hex_value
+ return codecs.decode(hex_value, 'hex_codec')
+
+
+class TestCreateBoundNetlinkSocket(CiTestCase):
+
+ @mock.patch('cloudinit.sources.helpers.netlink.socket.socket')
+ def test_socket_error_on_create(self, m_socket):
+ '''create_bound_netlink_socket catches socket creation exception'''
+
+ """NetlinkCreateSocketError is raised when socket creation errors."""
+ m_socket.side_effect = socket.error("Fake socket failure")
+ with self.assertRaises(NetlinkCreateSocketError) as ctx_mgr:
+ create_bound_netlink_socket()
+ self.assertEqual(
+ 'Exception during netlink socket create: Fake socket failure',
+ str(ctx_mgr.exception))
+
+
+class TestReadNetlinkSocket(CiTestCase):
+
+ @mock.patch('cloudinit.sources.helpers.netlink.socket.socket')
+ @mock.patch('cloudinit.sources.helpers.netlink.select.select')
+ def test_read_netlink_socket(self, m_select, m_socket):
+ '''read_netlink_socket able to receive data'''
+ data = 'netlinktest'
+ m_select.return_value = [m_socket], None, None
+ m_socket.recv.return_value = data
+ recv_data = read_netlink_socket(m_socket, 2)
+ m_select.assert_called_with([m_socket], [], [], 2)
+ m_socket.recv.assert_called_with(MAX_SIZE)
+ self.assertIsNotNone(recv_data)
+ self.assertEqual(recv_data, data)
+
+ @mock.patch('cloudinit.sources.helpers.netlink.socket.socket')
+ @mock.patch('cloudinit.sources.helpers.netlink.select.select')
+ def test_netlink_read_timeout(self, m_select, m_socket):
+ '''read_netlink_socket should timeout if nothing to read'''
+ m_select.return_value = [], None, None
+ data = read_netlink_socket(m_socket, 1)
+ m_select.assert_called_with([m_socket], [], [], 1)
+ self.assertEqual(m_socket.recv.call_count, 0)
+ self.assertIsNone(data)
+
+ def test_read_invalid_socket(self):
+ '''read_netlink_socket raises assert error if socket is invalid'''
+ socket = None
+ with self.assertRaises(AssertionError) as context:
+ read_netlink_socket(socket, 1)
+ self.assertTrue('netlink socket is none' in str(context.exception))
+
+
+class TestParseNetlinkMessage(CiTestCase):
+
+ def test_read_rta_oper_state(self):
+ '''read_rta_oper_state could parse netlink message and extract data'''
+ ifname = "eth0"
+ bytes = ifname.encode("utf-8")
+ buf = bytearray(48)
+ struct.pack_into("HH4sHHc", buf, RTATTR_START_OFFSET, 8, 3, bytes, 5,
+ 16, int_to_bytes(OPER_DOWN))
+ interface_state = read_rta_oper_state(buf)
+ self.assertEqual(interface_state.ifname, ifname)
+ self.assertEqual(interface_state.operstate, OPER_DOWN)
+
+ def test_read_none_data(self):
+ '''read_rta_oper_state raises assert error if data is none'''
+ data = None
+ with self.assertRaises(AssertionError) as context:
+ read_rta_oper_state(data)
+ self.assertTrue('data is none', str(context.exception))
+
+ def test_read_invalid_rta_operstate_none(self):
+ '''read_rta_oper_state returns none if operstate is none'''
+ ifname = "eth0"
+ buf = bytearray(40)
+ bytes = ifname.encode("utf-8")
+ struct.pack_into("HH4s", buf, RTATTR_START_OFFSET, 8, 3, bytes)
+ interface_state = read_rta_oper_state(buf)
+ self.assertIsNone(interface_state)
+
+ def test_read_invalid_rta_ifname_none(self):
+ '''read_rta_oper_state returns none if ifname is none'''
+ buf = bytearray(40)
+ struct.pack_into("HHc", buf, RTATTR_START_OFFSET, 5, 16,
+ int_to_bytes(OPER_DOWN))
+ interface_state = read_rta_oper_state(buf)
+ self.assertIsNone(interface_state)
+
+ def test_read_invalid_data_len(self):
+ '''raise assert error if data size is smaller than required size'''
+ buf = bytearray(32)
+ with self.assertRaises(AssertionError) as context:
+ read_rta_oper_state(buf)
+ self.assertTrue('length of data is smaller than RTATTR_START_OFFSET' in
+ str(context.exception))
+
+ def test_unpack_rta_attr_none_data(self):
+ '''unpack_rta_attr raises assert error if data is none'''
+ data = None
+ with self.assertRaises(AssertionError) as context:
+ unpack_rta_attr(data, RTATTR_START_OFFSET)
+ self.assertTrue('data is none' in str(context.exception))
+
+ def test_unpack_rta_attr_invalid_offset(self):
+ '''unpack_rta_attr raises assert error if offset is invalid'''
+ data = bytearray(48)
+ with self.assertRaises(AssertionError) as context:
+ unpack_rta_attr(data, "offset")
+ self.assertTrue('offset is not integer' in str(context.exception))
+ with self.assertRaises(AssertionError) as context:
+ unpack_rta_attr(data, 31)
+ self.assertTrue('rta offset is less than expected length' in
+ str(context.exception))
+
+
+@mock.patch('cloudinit.sources.helpers.netlink.socket.socket')
+@mock.patch('cloudinit.sources.helpers.netlink.read_netlink_socket')
+class TestWaitForMediaDisconnectConnect(CiTestCase):
+ with_logs = True
+
+ def _media_switch_data(self, ifname, msg_type, operstate):
+ '''construct netlink data with specified fields'''
+ if ifname and operstate is not None:
+ data = bytearray(48)
+ bytes = ifname.encode("utf-8")
+ struct.pack_into("HH4sHHc", data, RTATTR_START_OFFSET, 8, 3,
+ bytes, 5, 16, int_to_bytes(operstate))
+ elif ifname:
+ data = bytearray(40)
+ bytes = ifname.encode("utf-8")
+ struct.pack_into("HH4s", data, RTATTR_START_OFFSET, 8, 3, bytes)
+ elif operstate:
+ data = bytearray(40)
+ struct.pack_into("HHc", data, RTATTR_START_OFFSET, 5, 16,
+ int_to_bytes(operstate))
+ struct.pack_into("=LHHLL", data, 0, len(data), msg_type, 0, 0, 0)
+ return data
+
+ def test_media_down_up_scenario(self, m_read_netlink_socket,
+ m_socket):
+ '''Test for media down up sequence for required interface name'''
+ ifname = "eth0"
+ # construct data for Oper State down
+ data_op_down = self._media_switch_data(ifname, RTM_NEWLINK, OPER_DOWN)
+ # construct data for Oper State up
+ data_op_up = self._media_switch_data(ifname, RTM_NEWLINK, OPER_UP)
+ m_read_netlink_socket.side_effect = [data_op_down, data_op_up]
+ wait_for_media_disconnect_connect(m_socket, ifname)
+ self.assertEqual(m_read_netlink_socket.call_count, 2)
+
+ def test_wait_for_media_switch_diff_interface(self, m_read_netlink_socket,
+ m_socket):
+ '''wait_for_media_disconnect_connect ignores unexpected interfaces.
+
+ The first two messages are for other interfaces and last two are for
+ expected interface. So the function exit only after receiving last
+ 2 messages and therefore the call count for m_read_netlink_socket
+ has to be 4
+ '''
+ other_ifname = "eth1"
+ expected_ifname = "eth0"
+ data_op_down_eth1 = self._media_switch_data(
+ other_ifname, RTM_NEWLINK, OPER_DOWN)
+ data_op_up_eth1 = self._media_switch_data(
+ other_ifname, RTM_NEWLINK, OPER_UP)
+ data_op_down_eth0 = self._media_switch_data(
+ expected_ifname, RTM_NEWLINK, OPER_DOWN)
+ data_op_up_eth0 = self._media_switch_data(
+ expected_ifname, RTM_NEWLINK, OPER_UP)
+ m_read_netlink_socket.side_effect = [data_op_down_eth1,
+ data_op_up_eth1,
+ data_op_down_eth0,
+ data_op_up_eth0]
+ wait_for_media_disconnect_connect(m_socket, expected_ifname)
+ self.assertIn('Ignored netlink event on interface %s' % other_ifname,
+ self.logs.getvalue())
+ self.assertEqual(m_read_netlink_socket.call_count, 4)
+
+ def test_invalid_msgtype_getlink(self, m_read_netlink_socket, m_socket):
+ '''wait_for_media_disconnect_connect ignores GETLINK events.
+
+ The first two messages are for oper down and up for RTM_GETLINK type
+ which netlink module will ignore. The last 2 messages are RTM_NEWLINK
+ with oper state down and up messages. Therefore the call count for
+ m_read_netlink_socket has to be 4 ignoring first 2 messages
+ of RTM_GETLINK
+ '''
+ ifname = "eth0"
+ data_getlink_down = self._media_switch_data(
+ ifname, RTM_GETLINK, OPER_DOWN)
+ data_getlink_up = self._media_switch_data(
+ ifname, RTM_GETLINK, OPER_UP)
+ data_newlink_down = self._media_switch_data(
+ ifname, RTM_NEWLINK, OPER_DOWN)
+ data_newlink_up = self._media_switch_data(
+ ifname, RTM_NEWLINK, OPER_UP)
+ m_read_netlink_socket.side_effect = [data_getlink_down,
+ data_getlink_up,
+ data_newlink_down,
+ data_newlink_up]
+ wait_for_media_disconnect_connect(m_socket, ifname)
+ self.assertEqual(m_read_netlink_socket.call_count, 4)
+
+ def test_invalid_msgtype_setlink(self, m_read_netlink_socket, m_socket):
+ '''wait_for_media_disconnect_connect ignores SETLINK events.
+
+ The first two messages are for oper down and up for RTM_GETLINK type
+ which it will ignore. 3rd and 4th messages are RTM_NEWLINK with down
+ and up messages. This function should exit after 4th messages since it
+ sees down->up scenario. So the call count for m_read_netlink_socket
+ has to be 4 ignoring first 2 messages of RTM_GETLINK and
+ last 2 messages of RTM_NEWLINK
+ '''
+ ifname = "eth0"
+ data_setlink_down = self._media_switch_data(
+ ifname, RTM_SETLINK, OPER_DOWN)
+ data_setlink_up = self._media_switch_data(
+ ifname, RTM_SETLINK, OPER_UP)
+ data_newlink_down = self._media_switch_data(
+ ifname, RTM_NEWLINK, OPER_DOWN)
+ data_newlink_up = self._media_switch_data(
+ ifname, RTM_NEWLINK, OPER_UP)
+ m_read_netlink_socket.side_effect = [data_setlink_down,
+ data_setlink_up,
+ data_newlink_down,
+ data_newlink_up,
+ data_newlink_down,
+ data_newlink_up]
+ wait_for_media_disconnect_connect(m_socket, ifname)
+ self.assertEqual(m_read_netlink_socket.call_count, 4)
+
+ def test_netlink_invalid_switch_scenario(self, m_read_netlink_socket,
+ m_socket):
+ '''returns only if it receives UP event after a DOWN event'''
+ ifname = "eth0"
+ data_op_down = self._media_switch_data(ifname, RTM_NEWLINK, OPER_DOWN)
+ data_op_up = self._media_switch_data(ifname, RTM_NEWLINK, OPER_UP)
+ data_op_dormant = self._media_switch_data(ifname, RTM_NEWLINK,
+ OPER_DORMANT)
+ data_op_notpresent = self._media_switch_data(ifname, RTM_NEWLINK,
+ OPER_NOTPRESENT)
+ data_op_lowerdown = self._media_switch_data(ifname, RTM_NEWLINK,
+ OPER_LOWERLAYERDOWN)
+ data_op_testing = self._media_switch_data(ifname, RTM_NEWLINK,
+ OPER_TESTING)
+ data_op_unknown = self._media_switch_data(ifname, RTM_NEWLINK,
+ OPER_UNKNOWN)
+ m_read_netlink_socket.side_effect = [data_op_up, data_op_up,
+ data_op_dormant, data_op_up,
+ data_op_notpresent, data_op_up,
+ data_op_lowerdown, data_op_up,
+ data_op_testing, data_op_up,
+ data_op_unknown, data_op_up,
+ data_op_down, data_op_up]
+ wait_for_media_disconnect_connect(m_socket, ifname)
+ self.assertEqual(m_read_netlink_socket.call_count, 14)
+
+ def test_netlink_valid_inbetween_transitions(self, m_read_netlink_socket,
+ m_socket):
+ '''wait_for_media_disconnect_connect handles in between transitions'''
+ ifname = "eth0"
+ data_op_down = self._media_switch_data(ifname, RTM_NEWLINK, OPER_DOWN)
+ data_op_up = self._media_switch_data(ifname, RTM_NEWLINK, OPER_UP)
+ data_op_dormant = self._media_switch_data(ifname, RTM_NEWLINK,
+ OPER_DORMANT)
+ data_op_unknown = self._media_switch_data(ifname, RTM_NEWLINK,
+ OPER_UNKNOWN)
+ m_read_netlink_socket.side_effect = [data_op_down, data_op_dormant,
+ data_op_unknown, data_op_up]
+ wait_for_media_disconnect_connect(m_socket, ifname)
+ self.assertEqual(m_read_netlink_socket.call_count, 4)
+
+ def test_netlink_invalid_operstate(self, m_read_netlink_socket, m_socket):
+ '''wait_for_media_disconnect_connect should handle invalid operstates.
+
+ The function should not fail and return even if it receives invalid
+ operstates. It always should wait for down up sequence.
+ '''
+ ifname = "eth0"
+ data_op_down = self._media_switch_data(ifname, RTM_NEWLINK, OPER_DOWN)
+ data_op_up = self._media_switch_data(ifname, RTM_NEWLINK, OPER_UP)
+ data_op_invalid = self._media_switch_data(ifname, RTM_NEWLINK, 7)
+ m_read_netlink_socket.side_effect = [data_op_invalid, data_op_up,
+ data_op_down, data_op_invalid,
+ data_op_up]
+ wait_for_media_disconnect_connect(m_socket, ifname)
+ self.assertEqual(m_read_netlink_socket.call_count, 5)
+
+ def test_wait_invalid_socket(self, m_read_netlink_socket, m_socket):
+ '''wait_for_media_disconnect_connect handle none netlink socket.'''
+ socket = None
+ ifname = "eth0"
+ with self.assertRaises(AssertionError) as context:
+ wait_for_media_disconnect_connect(socket, ifname)
+ self.assertTrue('netlink socket is none' in str(context.exception))
+
+ def test_wait_invalid_ifname(self, m_read_netlink_socket, m_socket):
+ '''wait_for_media_disconnect_connect handle none interface name'''
+ ifname = None
+ with self.assertRaises(AssertionError) as context:
+ wait_for_media_disconnect_connect(m_socket, ifname)
+ self.assertTrue('interface name is none' in str(context.exception))
+ ifname = ""
+ with self.assertRaises(AssertionError) as context:
+ wait_for_media_disconnect_connect(m_socket, ifname)
+ self.assertTrue('interface name cannot be empty' in
+ str(context.exception))
+
+ def test_wait_invalid_rta_attr(self, m_read_netlink_socket, m_socket):
+ ''' wait_for_media_disconnect_connect handles invalid rta data'''
+ ifname = "eth0"
+ data_invalid1 = self._media_switch_data(None, RTM_NEWLINK, OPER_DOWN)
+ data_invalid2 = self._media_switch_data(ifname, RTM_NEWLINK, None)
+ data_op_down = self._media_switch_data(ifname, RTM_NEWLINK, OPER_DOWN)
+ data_op_up = self._media_switch_data(ifname, RTM_NEWLINK, OPER_UP)
+ m_read_netlink_socket.side_effect = [data_invalid1, data_invalid2,
+ data_op_down, data_op_up]
+ wait_for_media_disconnect_connect(m_socket, ifname)
+ self.assertEqual(m_read_netlink_socket.call_count, 4)
+
+ def test_read_multiple_netlink_msgs(self, m_read_netlink_socket, m_socket):
+ '''Read multiple messages in single receive call'''
+ ifname = "eth0"
+ bytes = ifname.encode("utf-8")
+ data = bytearray(96)
+ struct.pack_into("=LHHLL", data, 0, 48, RTM_NEWLINK, 0, 0, 0)
+ struct.pack_into("HH4sHHc", data, RTATTR_START_OFFSET, 8, 3,
+ bytes, 5, 16, int_to_bytes(OPER_DOWN))
+ struct.pack_into("=LHHLL", data, 48, 48, RTM_NEWLINK, 0, 0, 0)
+ struct.pack_into("HH4sHHc", data, 48 + RTATTR_START_OFFSET, 8,
+ 3, bytes, 5, 16, int_to_bytes(OPER_UP))
+ m_read_netlink_socket.return_value = data
+ wait_for_media_disconnect_connect(m_socket, ifname)
+ self.assertEqual(m_read_netlink_socket.call_count, 1)
+
+ def test_read_partial_netlink_msgs(self, m_read_netlink_socket, m_socket):
+ '''Read partial messages in receive call'''
+ ifname = "eth0"
+ bytes = ifname.encode("utf-8")
+ data1 = bytearray(112)
+ data2 = bytearray(32)
+ struct.pack_into("=LHHLL", data1, 0, 48, RTM_NEWLINK, 0, 0, 0)
+ struct.pack_into("HH4sHHc", data1, RTATTR_START_OFFSET, 8, 3,
+ bytes, 5, 16, int_to_bytes(OPER_DOWN))
+ struct.pack_into("=LHHLL", data1, 48, 48, RTM_NEWLINK, 0, 0, 0)
+ struct.pack_into("HH4sHHc", data1, 80, 8, 3, bytes, 5, 16,
+ int_to_bytes(OPER_DOWN))
+ struct.pack_into("=LHHLL", data1, 96, 48, RTM_NEWLINK, 0, 0, 0)
+ struct.pack_into("HH4sHHc", data2, 16, 8, 3, bytes, 5, 16,
+ int_to_bytes(OPER_UP))
+ m_read_netlink_socket.side_effect = [data1, data2]
+ wait_for_media_disconnect_connect(m_socket, ifname)
+ self.assertEqual(m_read_netlink_socket.call_count, 2)