b37de19c58
Till kernel 5.7 PF and VF representors are exposed as virtual device. They are not linked to its parent PCI device like how uplink representor is linked. Starting from kernel 5.8 due to new change [1] the PF and VF representors are linked to their parent PCI device, and so "get_ifname_by_pci_address" fails to get the correct UpLink Representor. This patch modifys the behviour of "get_ifname_by_pci_address" to check the physical port name of the netdev in vf_pci_addr_path/physfn/net to match the formart for the uplink "p\d+". [1] https://git.kernel.org/pub/scm/linux/kernel/git/netdev/net.git/commit/?id=123f0f53dd64b67e34142485fe866a8a581f12f1 Closes-Bug: #1892132 Change-Id: I49f6ae3f0e6bfbf555c8284bfd70371ce90da0c7
399 lines
17 KiB
Python
399 lines
17 KiB
Python
# 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 os.path
|
|
|
|
from unittest import mock
|
|
|
|
import testtools
|
|
|
|
from os_vif.internal.ip.api import ip as ip_lib
|
|
|
|
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(linux_net, "_arp_filtering")
|
|
@mock.patch.object(linux_net, "set_interface_state")
|
|
@mock.patch.object(linux_net, "_disable_ipv6")
|
|
@mock.patch.object(ip_lib, "add")
|
|
@mock.patch.object(ip_lib, "exists", return_value=False)
|
|
def test_ensure_bridge(self, mock_dev_exists, mock_add,
|
|
mock_disable_ipv6, mock_set_state,
|
|
mock_arp_filtering):
|
|
linux_net.ensure_bridge("br0")
|
|
|
|
mock_dev_exists.assert_called_once_with("br0")
|
|
mock_add.assert_called_once_with("br0", "bridge", ageing=0)
|
|
mock_disable_ipv6.assert_called_once_with("br0")
|
|
mock_set_state.assert_called_once_with("br0", "up")
|
|
mock_arp_filtering.assert_called_once_with("br0")
|
|
|
|
@mock.patch.object(linux_net, "_arp_filtering")
|
|
@mock.patch.object(linux_net, "set_interface_state")
|
|
@mock.patch.object(linux_net, "_disable_ipv6")
|
|
@mock.patch.object(ip_lib, "add")
|
|
@mock.patch.object(ip_lib, "exists", return_value=True)
|
|
def test_ensure_bridge_exists(self, mock_dev_exists, mock_add,
|
|
mock_disable_ipv6, mock_set_state,
|
|
mock_arp_filtering):
|
|
linux_net.ensure_bridge("br0")
|
|
|
|
mock_dev_exists.assert_called_once_with("br0")
|
|
mock_add.assert_not_called()
|
|
mock_disable_ipv6.assert_called_once_with("br0")
|
|
mock_set_state.assert_called_once_with("br0", "up")
|
|
mock_arp_filtering.assert_called_once_with("br0")
|
|
|
|
@mock.patch('builtins.open')
|
|
@mock.patch("os.path.exists")
|
|
def test__disable_ipv6(self, mock_exists, mock_open):
|
|
|
|
exists_path = "/proc/sys/net/ipv6/conf/br0/disable_ipv6"
|
|
mock_exists.return_value = False
|
|
|
|
linux_net._disable_ipv6("br0")
|
|
mock_exists.assert_called_once_with(exists_path)
|
|
mock_open.assert_not_called()
|
|
|
|
mock_exists.reset_mock()
|
|
mock_exists.return_value = True
|
|
linux_net._disable_ipv6("br0")
|
|
mock_exists.assert_called_once_with(exists_path)
|
|
mock_open.assert_called_once_with(exists_path, 'w')
|
|
|
|
@mock.patch.object(os.path, 'exists', return_value=True)
|
|
@mock.patch('builtins.open')
|
|
def test__arp_filtering(self, mock_open, *args):
|
|
mock_open.side_effect = mock.mock_open()
|
|
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')])
|
|
|
|
@mock.patch.object(ip_lib, "delete")
|
|
@mock.patch.object(ip_lib, "exists", return_value=False)
|
|
def test_delete_bridge_none(self, mock_dev_exists, mock_delete):
|
|
linux_net.delete_bridge("br0", "vnet1")
|
|
|
|
mock_delete.assert_not_called()
|
|
mock_dev_exists.assert_called_once_with("br0")
|
|
|
|
@mock.patch.object(linux_net, "set_interface_state")
|
|
@mock.patch.object(ip_lib, "delete")
|
|
@mock.patch.object(ip_lib, "exists", return_value=True)
|
|
def test_delete_bridge_exists(self, mock_dev_exists, mock_delete,
|
|
mock_set_state):
|
|
linux_net.delete_bridge("br0", "vnet1")
|
|
|
|
mock_dev_exists.assert_has_calls([mock.call("br0"),
|
|
mock.call("vnet1")])
|
|
mock_delete.assert_called_once_with("br0", check_exit_code=[0, 2, 254])
|
|
mock_set_state.assert_called_once_with("vnet1", "down")
|
|
|
|
@mock.patch.object(linux_net, "set_interface_state")
|
|
@mock.patch.object(ip_lib, "delete")
|
|
@mock.patch.object(ip_lib, "exists")
|
|
def test_delete_interface_not_present(self, mock_dev_exists, mock_delete,
|
|
mock_set_state):
|
|
mock_dev_exists.return_value = next(lambda: (yield True),
|
|
(yield False))
|
|
|
|
linux_net.delete_bridge("br0", "vnet1")
|
|
|
|
mock_dev_exists.assert_has_calls([mock.call("br0"),
|
|
mock.call("vnet1")])
|
|
mock_delete.assert_called_once_with("br0", check_exit_code=[0, 2, 254])
|
|
mock_set_state.assert_not_called()
|
|
|
|
@mock.patch.object(ip_lib, "set")
|
|
def test_add_bridge_port(self, mock_set):
|
|
linux_net.add_bridge_port("br0", "vnet1")
|
|
mock_set.assert_called_once_with("vnet1", master="br0")
|
|
|
|
@mock.patch.object(linux_net, '_get_phys_switch_id')
|
|
def test_is_switchdev_ioerror(self, mock__get_phys_switch_id):
|
|
mock__get_phys_switch_id.side_effect = ([IOError()])
|
|
test_switchdev = linux_net._is_switchdev('pf_ifname')
|
|
self.assertEqual(test_switchdev, False)
|
|
|
|
@mock.patch.object(linux_net, '_get_phys_switch_id')
|
|
def test_is_switchdev_empty(self, mock__get_phys_switch_id):
|
|
mock__get_phys_switch_id.return_value = ''
|
|
test_switchdev = linux_net._is_switchdev('pf_ifname')
|
|
self.assertEqual(test_switchdev, False)
|
|
|
|
@mock.patch.object(linux_net, '_get_phys_switch_id')
|
|
def test_is_switchdev_positive(self, mock__get_phys_switch_id):
|
|
mock__get_phys_switch_id.return_value = 'pf_sw_id'
|
|
test_switchdev = linux_net._is_switchdev('pf_ifname')
|
|
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.object(os, 'listdir')
|
|
@mock.patch.object(linux_net, "_get_pf_func")
|
|
@mock.patch.object(linux_net, "_get_phys_port_name")
|
|
@mock.patch.object(linux_net, '_get_phys_switch_id')
|
|
def test_get_representor_port(self, mock__get_phys_switch_id,
|
|
mock__get_phys_port_name,
|
|
mock__get_pf_func,
|
|
mock_listdir):
|
|
mock_listdir.return_value = [
|
|
'pf_ifname', 'rep_vf_1', 'rep_vf_2'
|
|
]
|
|
mock__get_phys_switch_id.return_value = 'pf_sw_id'
|
|
mock__get_pf_func.return_value = "0"
|
|
mock__get_phys_port_name.side_effect = (['1', "pf0vf1", "pf0vf2"])
|
|
ifname = linux_net.get_representor_port('pf_ifname', '2')
|
|
self.assertEqual('rep_vf_2', ifname)
|
|
|
|
@mock.patch.object(os, 'listdir')
|
|
@mock.patch.object(linux_net, "_get_pf_func")
|
|
@mock.patch.object(linux_net, "_get_phys_port_name")
|
|
@mock.patch.object(linux_net, "_get_phys_switch_id")
|
|
def test_get_representor_port_2_pfs(
|
|
self, mock__get_phys_switch_id, mock__get_phys_port_name,
|
|
mock__get_pf_func, mock_listdir):
|
|
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__get_phys_switch_id.return_value = 'pf_sw_id'
|
|
mock__get_pf_func.return_value = "2"
|
|
mock__get_phys_port_name.side_effect = (
|
|
["p1", "p2", "VF1@PF1", "pf2vf1", "vf2@pf1", "pf2vf2"])
|
|
ifname = linux_net.get_representor_port('pf_ifname2', '2')
|
|
self.assertEqual('rep_pf2_vf_2', ifname)
|
|
|
|
@mock.patch.object(os, 'listdir')
|
|
@mock.patch.object(linux_net, "_get_pf_func")
|
|
@mock.patch.object(linux_net, "_get_phys_switch_id")
|
|
@mock.patch.object(linux_net, "_get_phys_port_name")
|
|
def test_get_representor_port_not_found(
|
|
self, mock__get_phys_port_name, mock__get_phys_switch_id,
|
|
mock__get_pf_func, mock_listdir):
|
|
mock_listdir.return_value = [
|
|
'pf_ifname', 'rep_vf_1', 'rep_vf_2'
|
|
]
|
|
mock__get_phys_switch_id.return_value = 'pf_sw_id'
|
|
mock__get_pf_func.return_value = "0"
|
|
mock__get_phys_port_name.side_effect = (
|
|
["p0", "1", "2"])
|
|
self.assertRaises(
|
|
exception.RepresentorNotFound,
|
|
linux_net.get_representor_port,
|
|
'pf_ifname', '3'),
|
|
|
|
@mock.patch.object(os, 'listdir')
|
|
@mock.patch.object(linux_net, "_get_pf_func")
|
|
@mock.patch.object(linux_net, "_get_phys_port_name")
|
|
@mock.patch.object(linux_net, "_get_phys_switch_id")
|
|
def test_get_representor_port_exception_io_error(
|
|
self, mock__get_phys_switch_id, mock__get_phys_port_name,
|
|
mock__get_pf_func, mock_listdir):
|
|
mock_listdir.return_value = [
|
|
'pf_ifname', 'rep_vf_1', 'rep_vf_2'
|
|
]
|
|
mock__get_phys_switch_id.side_effect = (
|
|
['pf_sw_id', 'pf_sw_id', IOError(), 'pf_sw_id', '2'])
|
|
mock__get_pf_func.return_value = "0"
|
|
mock__get_phys_port_name.side_effect = (
|
|
["p0", "pf0vf0", "pf0vf1"])
|
|
self.assertRaises(
|
|
exception.RepresentorNotFound,
|
|
linux_net.get_representor_port,
|
|
'pf_ifname', '3')
|
|
|
|
@mock.patch.object(os, 'listdir')
|
|
@mock.patch.object(linux_net, "_get_pf_func")
|
|
@mock.patch.object(linux_net, "_get_phys_port_name")
|
|
@mock.patch.object(linux_net, "_get_phys_switch_id")
|
|
def test_get_representor_port_exception_value_error(
|
|
self, mock__get_phys_switch_id, mock__get_phys_port_name,
|
|
mock__get_pf_func, mock_listdir):
|
|
mock_listdir.return_value = [
|
|
'pf_ifname', 'rep_vf_1', 'rep_vf_2'
|
|
]
|
|
mock__get_phys_switch_id.return_value = 'pf_sw_id'
|
|
mock__get_phys_port_name.side_effect = (['p0', '1', 'a'])
|
|
mock__get_pf_func.return_value = "0"
|
|
self.assertRaises(
|
|
exception.RepresentorNotFound,
|
|
linux_net.get_representor_port,
|
|
'pf_ifname', '3')
|
|
|
|
@mock.patch.object(os, 'listdir')
|
|
@mock.patch.object(linux_net, '_get_phys_switch_id')
|
|
@mock.patch.object(linux_net, "_get_phys_port_name")
|
|
def test_physical_function_interface_name(
|
|
self, mock__get_phys_port_name, mock__get_phys_switch_id,
|
|
mock_listdir):
|
|
mock_listdir.return_value = ['foo', 'bar']
|
|
mock__get_phys_switch_id.side_effect = (
|
|
['', 'valid_switch'])
|
|
mock__get_phys_port_name.side_effect = (["p1"])
|
|
ifname = linux_net.get_ifname_by_pci_address(
|
|
'0000:00:00.1', pf_interface=True, switchdev=False)
|
|
self.assertEqual(ifname, 'foo')
|
|
|
|
@mock.patch.object(os, 'listdir')
|
|
@mock.patch.object(linux_net, "_get_phys_switch_id")
|
|
@mock.patch.object(linux_net, "_get_phys_port_name")
|
|
def test_physical_function_interface_name_with_switchdev(
|
|
self, mock__get_phys_port_name, mock__get_phys_switch_id,
|
|
mock_listdir):
|
|
mock_listdir.return_value = ['foo', 'bar']
|
|
mock__get_phys_switch_id.side_effect = (
|
|
['', 'valid_switch'])
|
|
mock__get_phys_port_name.side_effect = (["p1s0"])
|
|
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')
|
|
@mock.patch.object(linux_net, "_get_phys_switch_id")
|
|
@mock.patch.object(linux_net, "_get_phys_port_name")
|
|
def test_physical_function_interface_name_with_representors(
|
|
self, mock__get_phys_port_name, mock__get_phys_switch_id,
|
|
mock_listdir):
|
|
# Get the PF that matches the phys_port_name regex
|
|
mock_listdir.return_value = ['enp2s0f0_0', 'enp2s0f0_1', 'enp2s0f0']
|
|
mock__get_phys_switch_id.side_effect = (
|
|
['valid_switch', 'valid_switch', 'valid_switch'])
|
|
mock__get_phys_port_name.side_effect = (["pf0vf0", "pf0vf1", "p0"])
|
|
ifname = linux_net.get_ifname_by_pci_address(
|
|
'0000:00:00.1', pf_interface=True, switchdev=True)
|
|
self.assertEqual(ifname, 'enp2s0f0')
|
|
|
|
@mock.patch.object(os, 'listdir')
|
|
@mock.patch.object(linux_net, "_get_phys_switch_id")
|
|
@mock.patch.object(linux_net, "_get_phys_port_name")
|
|
def test_physical_function_interface_name_with_fallback_To_first_netdev(
|
|
self, mock__get_phys_port_name, mock__get_phys_switch_id,
|
|
mock_listdir):
|
|
# Try with switchdev mode to get PF but fail because there is no match
|
|
# for the phys_port_name then fallback to first interface found
|
|
mock_listdir.return_value = ['enp2s0f0_0', 'enp2s0f0_1', 'enp2s0f0']
|
|
mock__get_phys_switch_id.side_effect = (['valid_switch',
|
|
'valid_switch',
|
|
'valid_switch'])
|
|
mock__get_phys_port_name.side_effect = (["pf0vf0", "pf0vf1", "pf0vf2"])
|
|
ifname = linux_net.get_ifname_by_pci_address(
|
|
'0000:00:00.1', pf_interface=True, switchdev=True)
|
|
self.assertEqual(ifname, 'enp2s0f0_0')
|
|
|
|
@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('builtins.open')
|
|
@mock.patch.object(os.path, 'isfile')
|
|
def test__get_phys_port_name(self, mock_isfile, mock_open):
|
|
mock_open.return_value.__enter__ = lambda s: s
|
|
readline_mock = mock_open.return_value.readline
|
|
readline_mock.return_value = 'pf0vf0'
|
|
mock_isfile.return_value = True
|
|
phys_port_name = linux_net._get_phys_port_name("vf_ifname")
|
|
self.assertEqual(phys_port_name, 'pf0vf0')
|
|
|
|
@mock.patch.object(os.path, 'isfile')
|
|
def test__get_phys_port_name_not_found(self, mock_isfile):
|
|
mock_isfile.return_value = False
|
|
phys_port_name = linux_net._get_phys_port_name("vf_ifname")
|
|
self.assertIsNone(phys_port_name)
|
|
|
|
@mock.patch('builtins.open')
|
|
@mock.patch.object(os.path, 'isfile')
|
|
def test__get_phys_switch_id(self, mock_isfile, mock_open):
|
|
mock_open.return_value.__enter__ = lambda s: s
|
|
readline_mock = mock_open.return_value.readline
|
|
readline_mock.return_value = '66e40000039b0398'
|
|
mock_isfile.return_value = True
|
|
phys_port_name = linux_net._get_phys_switch_id("ifname")
|
|
self.assertEqual(phys_port_name, '66e40000039b0398')
|
|
|
|
@mock.patch.object(os.path, 'isfile')
|
|
def test__get_phys_switch_id_not_found(self, mock_isfile):
|
|
mock_isfile.return_value = False
|
|
phys_port_name = linux_net._get_phys_switch_id("ifname")
|
|
self.assertIsNone(phys_port_name)
|