# Licensed under the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. You may obtain # a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations # under the License. import glob import mock import os.path import testtools from os_vif.internal.command import ip as ip_lib from oslo_concurrency import processutils from six.moves import builtins from vif_plug_ovs import constants from vif_plug_ovs import exception from vif_plug_ovs import linux_net from vif_plug_ovs import privsep class LinuxNetTest(testtools.TestCase): def setUp(self): super(LinuxNetTest, self).setUp() privsep.vif_plug.set_client_mode(False) @mock.patch.object(ip_lib, "set") @mock.patch.object(linux_net, "device_exists", return_value=True) def test_ensure_bridge_exists(self, mock_dev_exists, mock_ip_set): linux_net.ensure_bridge("br0") mock_ip_set.assert_called_once_with('br0', state='up', check_exit_code=[0, 2, 254]) mock_dev_exists.assert_has_calls([mock.call("br0")]) @mock.patch.object(linux_net, "_arp_filtering") @mock.patch.object(ip_lib, "set") @mock.patch.object(os.path, "exists", return_value=False) @mock.patch.object(processutils, "execute") @mock.patch.object(linux_net, "device_exists", return_value=False) def test_ensure_bridge_new_ipv4(self, mock_dev_exists, mock_execute, mock_path_exists, mock_ip_set, *args): linux_net.ensure_bridge("br0") calls = [ mock.call('brctl', 'addbr', 'br0'), mock.call('brctl', 'setfd', 'br0', 0), mock.call('brctl', 'stp', 'br0', "off"), mock.call('brctl', 'setageing', 'br0', 0), mock.call('tee', '/sys/class/net/br0/bridge/multicast_snooping', check_exit_code=[0, 1], process_input='0'), ] mock_execute.assert_has_calls(calls) mock_dev_exists.assert_has_calls([mock.call("br0")]) mock_ip_set.assert_called_once_with('br0', state='up', check_exit_code=[0, 2, 254]) @mock.patch.object(linux_net, "_arp_filtering") @mock.patch.object(ip_lib, "set") @mock.patch.object(os.path, "exists", return_value=True) @mock.patch.object(processutils, "execute") @mock.patch.object(linux_net, "device_exists", return_value=False) def test_ensure_bridge_new_ipv6(self, mock_dev_exists, mock_execute, mock_path_exists, mock_ip_set, *args): linux_net.ensure_bridge("br0") calls = [ mock.call('brctl', 'addbr', 'br0'), mock.call('brctl', 'setfd', 'br0', 0), mock.call('brctl', 'stp', 'br0', "off"), mock.call('brctl', 'setageing', 'br0', 0), mock.call('tee', '/sys/class/net/br0/bridge/multicast_snooping', check_exit_code=[0, 1], process_input='0'), mock.call('tee', '/proc/sys/net/ipv6/conf/br0/disable_ipv6', check_exit_code=[0, 1], process_input='1'), ] mock_execute.assert_has_calls(calls) mock_dev_exists.assert_has_calls([mock.call("br0")]) mock_ip_set.assert_called_once_with('br0', state='up', check_exit_code=[0, 2, 254]) @mock.patch.object(processutils, "execute") @mock.patch.object(linux_net, "device_exists", return_value=False) @mock.patch.object(linux_net, "interface_in_bridge", return_value=False) def test_delete_bridge_none(self, mock_interface_br, mock_dev_exists, mock_execute,): linux_net.delete_bridge("br0", "vnet1") mock_execute.assert_not_called() mock_dev_exists.assert_has_calls([mock.call("br0")]) mock_interface_br.assert_not_called() @mock.patch.object(ip_lib, "set") @mock.patch.object(processutils, "execute") @mock.patch.object(linux_net, "device_exists", return_value=True) @mock.patch.object(linux_net, "interface_in_bridge", return_value=True) def test_delete_bridge_exists(self, mock_interface_br, mock_dev_exists, mock_execute, mock_ip_set): linux_net.delete_bridge("br0", "vnet1") calls = [ mock.call('brctl', 'delif', 'br0', 'vnet1'), mock.call('brctl', 'delbr', 'br0')] mock_execute.assert_has_calls(calls) mock_dev_exists.assert_has_calls([mock.call("br0")]) mock_interface_br.assert_called_once_with("br0", "vnet1") mock_ip_set.assert_called_once_with('br0', state='down') @mock.patch.object(ip_lib, "set") @mock.patch.object(processutils, "execute") @mock.patch.object(linux_net, "device_exists", return_value=True) @mock.patch.object(linux_net, "interface_in_bridge", return_value=False) def test_delete_interface_not_present(self, mock_interface_br, mock_dev_exists, mock_execute, mock_ip_set): linux_net.delete_bridge("br0", "vnet1") mock_execute.assert_called_once_with('brctl', 'delbr', 'br0') mock_dev_exists.assert_has_calls([mock.call("br0")]) mock_interface_br.assert_called_once_with("br0", "vnet1") mock_ip_set.assert_called_once_with('br0', state='down') @mock.patch.object(processutils, "execute") def test_add_bridge_port(self, mock_execute): linux_net.add_bridge_port("br0", "vnet1") mock_execute.assert_has_calls([ mock.call('brctl', 'addif', 'br0', 'vnet1')]) def test_ovs_vif_port_cmd(self): expected = ['--', '--may-exist', 'add-port', 'fake-bridge', 'fake-dev', '--', 'set', 'Interface', 'fake-dev', 'external-ids:iface-id=fake-iface-id', 'external-ids:iface-status=active', 'external-ids:attached-mac=fake-mac', 'external-ids:vm-uuid=fake-instance-uuid'] cmd = linux_net._create_ovs_vif_cmd('fake-bridge', 'fake-dev', 'fake-iface-id', 'fake-mac', 'fake-instance-uuid') self.assertEqual(expected, cmd) expected += ['type=fake-type'] cmd = linux_net._create_ovs_vif_cmd('fake-bridge', 'fake-dev', 'fake-iface-id', 'fake-mac', 'fake-instance-uuid', 'fake-type') self.assertEqual(expected, cmd) expected += ['options:vhost-server-path=/fake/path'] cmd = linux_net._create_ovs_vif_cmd('fake-bridge', 'fake-dev', 'fake-iface-id', 'fake-mac', 'fake-instance-uuid', 'fake-type', vhost_server_path='/fake/path') self.assertEqual(expected, cmd) @mock.patch.object(linux_net, '_create_ovs_bridge_cmd') @mock.patch.object(linux_net, '_ovs_vsctl') def test_ensure_ovs_bridge(self, mock_vsctl, mock_create_ovs_bridge): bridge = 'fake-bridge' dp_type = 'fake-type' linux_net.ensure_ovs_bridge(bridge, dp_type) mock_create_ovs_bridge.assert_called_once_with(bridge, dp_type) self.assertTrue(mock_vsctl.called) def test_create_ovs_bridge_cmd(self): bridge = 'fake-bridge' dp_type = 'fake-type' expected = ['--', '--may-exist', 'add-br', bridge, '--', 'set', 'Bridge', bridge, 'datapath_type=%s' % dp_type] actual = linux_net._create_ovs_bridge_cmd(bridge, dp_type) self.assertEqual(expected, actual) @mock.patch.object(linux_net, '_ovs_supports_mtu_requests') @mock.patch.object(linux_net, '_ovs_vsctl') @mock.patch.object(linux_net, '_create_ovs_vif_cmd') @mock.patch.object(linux_net, '_set_device_mtu') def test_ovs_vif_port_with_type_vhostuser(self, mock_set_device_mtu, mock_create_cmd, mock_vsctl, mock_ovs_supports_mtu_requests): mock_ovs_supports_mtu_requests.return_value = True linux_net.create_ovs_vif_port( 'fake-bridge', 'fake-dev', 'fake-iface-id', 'fake-mac', "fake-instance-uuid", mtu=1500, interface_type=constants.OVS_VHOSTUSER_INTERFACE_TYPE) mock_create_cmd.assert_called_once_with('fake-bridge', 'fake-dev', 'fake-iface-id', 'fake-mac', "fake-instance-uuid", constants.OVS_VHOSTUSER_INTERFACE_TYPE, None) self.assertFalse(mock_set_device_mtu.called) self.assertTrue(mock_vsctl.called) @mock.patch.object(linux_net, '_ovs_supports_mtu_requests') @mock.patch.object(linux_net, '_ovs_vsctl') @mock.patch.object(linux_net, '_create_ovs_vif_cmd') @mock.patch.object(linux_net, '_set_device_mtu') def test_ovs_vif_port_with_type_vhostuserclient(self, mock_set_device_mtu, mock_create_cmd, mock_vsctl, mock_ovs_supports_mtu_requests): mock_ovs_supports_mtu_requests.return_value = True linux_net.create_ovs_vif_port( 'fake-bridge', 'fake-dev', 'fake-iface-id', 'fake-mac', "fake-instance-uuid", mtu=1500, interface_type=constants.OVS_VHOSTUSER_CLIENT_INTERFACE_TYPE, vhost_server_path="/fake/path") mock_create_cmd.assert_called_once_with('fake-bridge', 'fake-dev', 'fake-iface-id', 'fake-mac', "fake-instance-uuid", constants.OVS_VHOSTUSER_CLIENT_INTERFACE_TYPE, "/fake/path") self.assertFalse(mock_set_device_mtu.called) self.assertTrue(mock_vsctl.called) @mock.patch.object(linux_net, '_ovs_supports_mtu_requests') @mock.patch.object(linux_net, '_ovs_vsctl') @mock.patch.object(linux_net, '_create_ovs_vif_cmd') @mock.patch.object(linux_net, '_set_device_mtu') def test_ovs_vif_port_with_no_mtu(self, mock_set_device_mtu, mock_create_cmd, mock_vsctl, mock_ovs_supports_mtu_requests): mock_ovs_supports_mtu_requests.return_value = True linux_net.create_ovs_vif_port( 'fake-bridge', 'fake-dev', 'fake-iface-id', 'fake-mac', "fake-instance-uuid") mock_create_cmd.assert_called_once_with('fake-bridge', 'fake-dev', 'fake-iface-id', 'fake-mac', "fake-instance-uuid", None, None) self.assertFalse(mock_set_device_mtu.called) self.assertTrue(mock_vsctl.called) @mock.patch.object(linux_net, '_ovs_supports_mtu_requests') @mock.patch.object(linux_net, '_set_mtu_request') @mock.patch.object(linux_net, '_ovs_vsctl') @mock.patch.object(linux_net, '_create_ovs_vif_cmd', return_value='ovs_command') @mock.patch.object(linux_net, '_set_device_mtu') def test_ovs_vif_port_with_timeout(self, mock_set_device_mtu, mock_create_cmd, mock_vsctl, mock_set_mtu_request, mock_ovs_supports_mtu_requests): mock_ovs_supports_mtu_requests.return_value = True linux_net.create_ovs_vif_port( 'fake-bridge', 'fake-dev', 'fake-iface-id', 'fake-mac', "fake-instance-uuid", timeout=42) self.assertTrue(mock_create_cmd.called) self.assertFalse(mock_set_device_mtu.called) mock_vsctl.assert_called_with('ovs_command', timeout=42) @mock.patch.object(linux_net, '_ovs_supports_mtu_requests') @mock.patch.object(linux_net, '_set_mtu_request') @mock.patch.object(linux_net, '_ovs_vsctl') @mock.patch.object(linux_net, '_create_ovs_vif_cmd', return_value='ovs_command') @mock.patch.object(linux_net, '_set_device_mtu') def test_ovs_vif_port_with_no_timeout(self, mock_set_device_mtu, mock_create_cmd, mock_vsctl, mock_set_mtu_request, mock_ovs_supports_mtu_requests): mock_ovs_supports_mtu_requests.return_value = True linux_net.create_ovs_vif_port( 'fake-bridge', 'fake-dev', 'fake-iface-id', 'fake-mac', "fake-instance-uuid") self.assertTrue(mock_create_cmd.called) self.assertFalse(mock_set_device_mtu.called) mock_vsctl.assert_called_with('ovs_command', timeout=None) @mock.patch.object(processutils, "execute") def test_ovs_vsctl(self, mock_execute): args = ['fake-args', 42] timeout = 42 linux_net._ovs_vsctl(args) linux_net._ovs_vsctl(args, timeout=timeout) mock_execute.assert_has_calls([ mock.call('ovs-vsctl', *args), mock.call('ovs-vsctl', '--timeout=%s' % timeout, *args)]) @mock.patch.object(linux_net, '_ovs_vsctl') def test_set_mtu_request(self, mock_vsctl): dev = 'fake-dev' mtu = 'fake-mtu' timeout = 120 linux_net._set_mtu_request(dev, mtu, timeout=timeout) args = ['--', 'set', 'interface', dev, 'mtu_request=%s' % mtu] mock_vsctl.assert_called_with(args, timeout=timeout) @mock.patch.object(linux_net, '_delete_net_dev') @mock.patch.object(linux_net, '_ovs_vsctl') def test_delete_ovs_vif_port_delete_netdev( self, mock_vsctl, mock_delete_net_dev): bridge = 'fake-bridge' dev = 'fake-dev' timeout = 120 linux_net.delete_ovs_vif_port(bridge, dev, timeout=timeout) args = ['--', '--if-exists', 'del-port', bridge, dev] mock_vsctl.assert_called_with(args, timeout=timeout) mock_delete_net_dev.assert_called() @mock.patch.object(linux_net, '_delete_net_dev') @mock.patch.object(linux_net, '_ovs_vsctl') def test_delete_ovs_vif_port(self, mock_vsctl, mock_delete_net_dev): bridge = 'fake-bridge' dev = 'fake-dev' timeout = 120 linux_net.delete_ovs_vif_port( bridge, dev, timeout=timeout, delete_netdev=False) args = ['--', '--if-exists', 'del-port', bridge, dev] mock_vsctl.assert_called_with(args, timeout=timeout) mock_delete_net_dev.assert_not_called() @mock.patch.object(linux_net, '_ovs_vsctl') def test_ovs_supports_mtu_requests(self, mock_vsctl): args = ['--columns=mtu_request', 'list', 'interface'] timeout = 120 msg = 'ovs-vsctl: Interface does not contain' + \ ' a column whose name matches "mtu_request"' mock_vsctl.return_value = (None, msg) result = linux_net._ovs_supports_mtu_requests(timeout=timeout) mock_vsctl.assert_called_with(args, timeout=timeout) self.assertFalse(result) mock_vsctl.return_value = (None, None) result = linux_net._ovs_supports_mtu_requests(timeout=timeout) mock_vsctl.assert_called_with(args, timeout=timeout) self.assertTrue(result) @mock.patch('six.moves.builtins.open') @mock.patch.object(os.path, 'isfile') def test_is_switchdev_ioerror(self, mock_isfile, mock_open): mock_isfile.side_effect = [True] mock_open.return_value.__enter__ = lambda s: s readline_mock = mock_open.return_value.readline readline_mock.side_effect = ( [IOError()]) test_switchdev = linux_net._is_switchdev('pf_ifname') self.assertEqual(test_switchdev, False) @mock.patch('six.moves.builtins.open') @mock.patch.object(os.path, 'isfile') def test_is_switchdev_empty(self, mock_isfile, mock_open): mock_isfile.side_effect = [True] mock_open.return_value.__enter__ = lambda s: s readline_mock = mock_open.return_value.readline readline_mock.side_effect = ( ['']) open_calls = ( [mock.call('/sys/class/net/pf_ifname/phys_switch_id', 'r'), mock.call().readline(), mock.call().__exit__(None, None, None)]) test_switchdev = linux_net._is_switchdev('pf_ifname') mock_open.assert_has_calls(open_calls) self.assertEqual(test_switchdev, False) @mock.patch('six.moves.builtins.open') @mock.patch.object(os.path, 'isfile') def test_is_switchdev_positive(self, mock_isfile, mock_open): mock_isfile.side_effect = [True] mock_open.return_value.__enter__ = lambda s: s readline_mock = mock_open.return_value.readline readline_mock.side_effect = ( ['pf_sw_id']) open_calls = ( [mock.call('/sys/class/net/pf_ifname/phys_switch_id', 'r'), mock.call().readline(), mock.call().__exit__(None, None, None)]) test_switchdev = linux_net._is_switchdev('pf_ifname') mock_open.assert_has_calls(open_calls) self.assertEqual(test_switchdev, True) def test_parse_vf_number(self): self.assertEqual(linux_net._parse_vf_number("0"), "0") self.assertEqual(linux_net._parse_vf_number("pf13vf42"), "42") self.assertEqual(linux_net._parse_vf_number("VF19@PF13"), "19") self.assertIsNone(linux_net._parse_vf_number("p7")) self.assertIsNone(linux_net._parse_vf_number("pf31")) self.assertIsNone(linux_net._parse_vf_number("g4rbl3d")) def test_parse_pf_number(self): self.assertIsNone(linux_net._parse_pf_number("0")) self.assertEqual(linux_net._parse_pf_number("pf13vf42"), "13") self.assertEqual(linux_net._parse_pf_number("VF19@PF13"), "13") self.assertIsNone(linux_net._parse_pf_number("p7")) self.assertEqual(linux_net._parse_pf_number("pf31"), "31") self.assertIsNone(linux_net._parse_pf_number("g4rbl3d")) @mock.patch('six.moves.builtins.open') @mock.patch.object(os.path, 'isfile') @mock.patch.object(os, 'listdir') @mock.patch.object(linux_net, "get_function_by_ifname") def test_get_representor_port(self, mock_get_function_by_ifname, mock_listdir, mock_isfile, mock_open): mock_listdir.return_value = [ 'pf_ifname', 'rep_vf_1', 'rep_vf_2' ] mock_isfile.side_effect = [True, True] mock_open.return_value.__enter__ = lambda s: s readline_mock = mock_open.return_value.readline readline_mock.side_effect = ( ['pf_sw_id', 'pf_sw_id', '1', 'pf_sw_id', 'pf0vf2']) # PCI IDs mocked: # PF0: 0000:0a:00.0 # PF0VF1: 0000:0a:02.1 PF0VF2: 0000:0a:02.2 mock_get_function_by_ifname.side_effect = ( [("0000:0a:00.0", True), ("0000:0a:02.1", False), ("0000:0a:02.2", False), ("0000:0a:00.0", True)]) open_calls = ( [mock.call('/sys/class/net/pf_ifname/phys_switch_id', 'r'), mock.call().readline(), mock.call().__exit__(None, None, None), mock.call('/sys/class/net/rep_vf_1/phys_switch_id', 'r'), mock.call().readline(), mock.call().__exit__(None, None, None), mock.call('/sys/class/net/rep_vf_1/phys_port_name', 'r'), mock.call().readline(), mock.call().__exit__(None, None, None), mock.call('/sys/class/net/rep_vf_2/phys_switch_id', 'r'), mock.call().readline(), mock.call().__exit__(None, None, None), mock.call('/sys/class/net/rep_vf_2/phys_port_name', 'r'), mock.call().readline(), mock.call().__exit__(None, None, None)]) ifname = linux_net.get_representor_port('pf_ifname', '2') mock_open.assert_has_calls(open_calls) self.assertEqual('rep_vf_2', ifname) @mock.patch('six.moves.builtins.open') @mock.patch.object(os.path, 'isfile') @mock.patch.object(os, 'listdir') @mock.patch.object(linux_net, "get_function_by_ifname") def test_get_representor_port_2_pfs( self, mock_get_function_by_ifname, mock_listdir, mock_isfile, mock_open): mock_listdir.return_value = [ 'pf_ifname1', 'pf_ifname2', 'rep_pf1_vf_1', 'rep_pf1_vf_2', 'rep_pf2_vf_1', 'rep_pf2_vf_2', ] mock_isfile.side_effect = [True, True, True, True] mock_open.return_value.__enter__ = lambda s: s readline_mock = mock_open.return_value.readline readline_mock.side_effect = ( ['pf_sw_id', 'pf_sw_id', 'VF1@PF1', 'pf_sw_id', 'vf2@pf1', 'pf_sw_id', 'pf2vf1', 'pf_sw_id', 'pf2vf2']) # PCI IDs mocked: # PF1: 0000:0a:00.1 PF2: 0000:0a:00.2 # PF1VF1: 0000:0a:02.1 PF1VF2: 0000:0a:02.2 # PF2VF1: 0000:0a:04.1 PF2VF2: 0000:0a:04.2 mock_get_function_by_ifname.side_effect = ( [("0000:0a:00.1", True), ("0000:0a:00.2", True), ("0000:0a:02.1", False), ("0000:0a:00.2", True), ("0000:0a:02.2", False), ("0000:0a:00.2", True), ("0000:0a:04.1", False), ("0000:0a:00.2", True), ("0000:0a:04.2", False), ("0000:0a:00.2", True)]) ifname = linux_net.get_representor_port('pf_ifname2', '2') self.assertEqual('rep_pf2_vf_2', ifname) @mock.patch('six.moves.builtins.open') @mock.patch.object(os.path, 'isfile') @mock.patch.object(os, 'listdir') @mock.patch.object(linux_net, "get_function_by_ifname") def test_get_representor_port_not_found( self, mock_get_function_by_ifname, mock_listdir, mock_isfile, mock_open): mock_listdir.return_value = [ 'pf_ifname', 'rep_vf_1', 'rep_vf_2' ] mock_isfile.side_effect = [True, True] mock_open.return_value.__enter__ = lambda s: s readline_mock = mock_open.return_value.readline readline_mock.side_effect = ( ['pf_sw_id', 'pf_sw_id', '1', 'pf_sw_id', '2']) # PCI IDs mocked: # PF0: 0000:0a:00.0 # PF0VF1: 0000:0a:02.1 PF0VF2: 0000:0a:02.2 mock_get_function_by_ifname.side_effect = ( [("0000:0a:00.0", True), ("0000:0a:02.1", False), ("0000:0a:02.2", False)]) self.assertRaises( exception.RepresentorNotFound, linux_net.get_representor_port, 'pf_ifname', '3'), @mock.patch('six.moves.builtins.open') @mock.patch.object(os.path, 'isfile') @mock.patch.object(os, 'listdir') @mock.patch.object(linux_net, "get_function_by_ifname") def test_get_representor_port_exception_io_error( self, mock_get_function_by_ifname, mock_listdir, mock_isfile, mock_open): mock_listdir.return_value = [ 'pf_ifname', 'rep_vf_1', 'rep_vf_2' ] mock_isfile.side_effect = [True, True] mock_open.return_value.__enter__ = lambda s: s readline_mock = mock_open.return_value.readline readline_mock.side_effect = ( ['pf_sw_id', 'pf_sw_id', IOError(), 'pf_sw_id', '2']) # PCI IDs mocked: # PF0: 0000:0a:00.0 # PF0VF1: 0000:0a:02.1 PF0VF2: 0000:0a:02.2 mock_get_function_by_ifname.side_effect = ( [("0000:0a:00.0", True), ("0000:0a:02.1", False), ("0000:0a:02.2", False), ("0000:0a:00.0", True)]) self.assertRaises( exception.RepresentorNotFound, linux_net.get_representor_port, 'pf_ifname', '3') @mock.patch('six.moves.builtins.open') @mock.patch.object(os.path, 'isfile') @mock.patch.object(os, 'listdir') @mock.patch.object(linux_net, "get_function_by_ifname") def test_get_representor_port_exception_value_error( self, mock_get_function_by_ifname, mock_listdir, mock_isfile, mock_open): mock_listdir.return_value = [ 'pf_ifname', 'rep_vf_1', 'rep_vf_2' ] mock_isfile.side_effect = [True, True] mock_open.return_value.__enter__ = lambda s: s readline_mock = mock_open.return_value.readline readline_mock.side_effect = ( ['pf_sw_id', 'pf_sw_id', '1', 'pf_sw_id', 'a']) # PCI IDs mocked: # PF0: 0000:0a:00.0 # PF0VF1: 0000:0a:02.1 PF0VF2: 0000:0a:02.2 mock_get_function_by_ifname.side_effect = ( [("0000:0a:00.0", True), ("0000:0a:02.1", False), ("0000:0a:02.2", False)]) self.assertRaises( exception.RepresentorNotFound, linux_net.get_representor_port, 'pf_ifname', '3') @mock.patch('six.moves.builtins.open') @mock.patch.object(os.path, 'isfile') @mock.patch.object(os, 'listdir') def test_physical_function_inferface_name( self, mock_listdir, mock_isfile, mock_open): mock_listdir.return_value = ['foo', 'bar'] mock_isfile.side_effect = [True, True] mock_open.return_value.__enter__ = lambda s: s readline_mock = mock_open.return_value.readline readline_mock.side_effect = ( ['', 'valid_switch']) ifname = linux_net.get_ifname_by_pci_address( '0000:00:00.1', pf_interface=True, switchdev=False) self.assertEqual(ifname, 'foo') @mock.patch('six.moves.builtins.open') @mock.patch.object(os.path, 'isfile') @mock.patch.object(os, 'listdir') def test_physical_function_inferface_name_with_switchdev( self, mock_listdir, mock_isfile, mock_open): mock_listdir.return_value = ['foo', 'bar'] mock_isfile.side_effect = [True, True] mock_open.return_value.__enter__ = lambda s: s readline_mock = mock_open.return_value.readline readline_mock.side_effect = ( ['', 'valid_switch']) ifname = linux_net.get_ifname_by_pci_address( '0000:00:00.1', pf_interface=True, switchdev=True) self.assertEqual(ifname, 'bar') @mock.patch.object(os, 'listdir') def test_get_ifname_by_pci_address_exception(self, mock_listdir): mock_listdir.side_effect = OSError('No such file or directory') self.assertRaises( exception.PciDeviceNotFoundById, linux_net.get_ifname_by_pci_address, '0000:00:00.1' ) @mock.patch.object(os, 'readlink') @mock.patch.object(glob, 'iglob') def test_vf_number_found(self, mock_iglob, mock_readlink): mock_iglob.return_value = [ '/sys/bus/pci/devices/0000:00:00.1/physfn/virtfn3', ] mock_readlink.return_value = '../../0000:00:00.1' vf_num = linux_net.get_vf_num_by_pci_address('0000:00:00.1') self.assertEqual(vf_num, '3') @mock.patch.object(os, 'readlink') @mock.patch.object(glob, 'iglob') def test_vf_number_not_found(self, mock_iglob, mock_readlink): mock_iglob.return_value = [ '/sys/bus/pci/devices/0000:00:00.1/physfn/virtfn3', ] mock_readlink.return_value = '../../0000:00:00.2' self.assertRaises( exception.PciDeviceNotFoundById, linux_net.get_vf_num_by_pci_address, '0000:00:00.1' ) @mock.patch.object(os, 'readlink') @mock.patch.object(glob, 'iglob') def test_get_vf_num_by_pci_address_exception( self, mock_iglob, mock_readlink): mock_iglob.return_value = [ '/sys/bus/pci/devices/0000:00:00.1/physfn/virtfn3', ] mock_readlink.side_effect = OSError('No such file or directory') self.assertRaises( exception.PciDeviceNotFoundById, linux_net.get_vf_num_by_pci_address, '0000:00:00.1' ) @mock.patch.object(os.path, 'exists', return_value=True) @mock.patch.object(builtins, 'open') def test__arp_filtering(self, mock_open, *args): mock_open.side_effect = mock.mock_open(read_data=mock.Mock()) linux_net._arp_filtering("br0") mock_open.assert_has_calls([ mock.call('/proc/sys/net/ipv4/conf/br0/arp_ignore', 'w'), mock.call('/proc/sys/net/ipv4/conf/br0/arp_announce', 'w')]) mock_open.side_effect.return_value.write.assert_has_calls([ mock.call('1'), mock.call('2')])