154 lines
6.6 KiB
Python
154 lines
6.6 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 mock
|
|
import testtools
|
|
|
|
import fixtures
|
|
from os_vif.internal.ip.api import ip as ip_lib
|
|
from oslo_concurrency import lockutils
|
|
from oslo_concurrency import processutils
|
|
from oslo_config import cfg
|
|
from oslo_config import fixture as config_fixture
|
|
from oslo_log.fixture import logging_error as log_fixture
|
|
|
|
from vif_plug_linux_bridge import linux_net
|
|
from vif_plug_linux_bridge import privsep
|
|
|
|
CONF = cfg.CONF
|
|
|
|
|
|
class LinuxNetTest(testtools.TestCase):
|
|
|
|
def setUp(self):
|
|
super(LinuxNetTest, self).setUp()
|
|
|
|
privsep.vif_plug.set_client_mode(False)
|
|
lock_path = self.useFixture(fixtures.TempDir()).path
|
|
self.fixture = self.useFixture(
|
|
config_fixture.Config(lockutils.CONF))
|
|
self.fixture.config(lock_path=lock_path,
|
|
group='oslo_concurrency')
|
|
self.useFixture(log_fixture.get_logging_handle_error_fixture())
|
|
|
|
@mock.patch.object(ip_lib, "set")
|
|
def test_set_device_mtu(self, mock_ip_set):
|
|
linux_net._set_device_mtu(dev='fakedev', mtu=1500)
|
|
mock_ip_set.assert_called_once_with('fakedev', mtu=1500,
|
|
check_exit_code=[0, 2, 254])
|
|
|
|
@mock.patch.object(processutils, "execute")
|
|
def test_set_device_invalid_mtu(self, mock_exec):
|
|
linux_net._set_device_mtu(dev='fakedev', mtu=None)
|
|
mock_exec.assert_not_called()
|
|
|
|
@mock.patch.object(ip_lib, "add")
|
|
@mock.patch.object(ip_lib, "set")
|
|
@mock.patch.object(ip_lib, "exists", return_value=False)
|
|
@mock.patch.object(linux_net, "_set_device_mtu")
|
|
def test_ensure_vlan(self, mock_set_mtu, mock_dev_exists, mock_ip_set,
|
|
mock_ip_add):
|
|
linux_net._ensure_vlan_privileged(123, 'fake-bridge',
|
|
mac_address='fake-mac',
|
|
mtu=1500)
|
|
self.assertTrue(mock_dev_exists.called)
|
|
set_calls = [mock.call('vlan123', address='fake-mac',
|
|
check_exit_code=[0, 2, 254]),
|
|
mock.call('vlan123', state='up',
|
|
check_exit_code=[0, 2, 254])]
|
|
mock_ip_add.assert_called_once_with(
|
|
'vlan123', 'vlan', link='fake-bridge', vlan_id=123,
|
|
check_exit_code=[0, 2, 254])
|
|
mock_ip_set.assert_has_calls(set_calls)
|
|
mock_set_mtu.assert_called_once_with('vlan123', 1500)
|
|
|
|
@mock.patch.object(linux_net, "_ensure_bridge_privileged")
|
|
@mock.patch.object(linux_net, "_ensure_bridge_filtering")
|
|
def test_ensure_bridge(self, mock_filtering, mock_priv):
|
|
linux_net.ensure_bridge("br0", None, filtering=False)
|
|
|
|
mock_priv.assert_called_once_with("br0", None, None, True,
|
|
filtering=False, mtu=None)
|
|
mock_filtering.assert_not_called()
|
|
|
|
linux_net.ensure_bridge("br0", None, filtering=True)
|
|
mock_filtering.assert_called_once_with("br0", True)
|
|
|
|
@mock.patch.object(ip_lib, "exists", return_value=False)
|
|
@mock.patch.object(ip_lib, "add")
|
|
def test_ensure_bridge_addbr_exception(self, mock_add, mock_dev_exists):
|
|
mock_add.side_effect = ValueError()
|
|
with testtools.ExpectedException(ValueError):
|
|
linux_net.ensure_bridge("br0", None, filtering=False)
|
|
|
|
@mock.patch.object(ip_lib, "add")
|
|
@mock.patch.object(ip_lib, "set")
|
|
@mock.patch.object(linux_net, "_arp_filtering")
|
|
@mock.patch.object(linux_net, "_set_device_mtu")
|
|
@mock.patch.object(linux_net, "_disable_ipv6")
|
|
@mock.patch.object(linux_net, "_update_bridge_routes")
|
|
@mock.patch.object(ip_lib, "exists")
|
|
def test_ensure_bridge_priv_mtu_not_called(self, mock_dev_exists,
|
|
mock_routes, mock_disable_ipv6, mock_set_mtu, mock_arp_filtering,
|
|
mock_ip_set, mock_add):
|
|
"""This test validates that mtus are updated only if an interface
|
|
is added to the bridge
|
|
"""
|
|
mock_dev_exists.return_value = False
|
|
linux_net._ensure_bridge_privileged("fake-bridge", None,
|
|
None, False, mtu=1500)
|
|
mock_set_mtu.assert_not_called()
|
|
mock_ip_set.assert_called_once_with('fake-bridge', state='up')
|
|
|
|
@mock.patch.object(ip_lib, "add")
|
|
@mock.patch.object(ip_lib, "set")
|
|
@mock.patch.object(linux_net, "_arp_filtering")
|
|
@mock.patch.object(linux_net, "_set_device_mtu")
|
|
@mock.patch.object(linux_net, "_disable_ipv6")
|
|
@mock.patch.object(linux_net, "_update_bridge_routes")
|
|
@mock.patch.object(ip_lib, "exists")
|
|
def test_ensure_bridge_priv_mtu_order(self, mock_dev_exists, mock_routes,
|
|
mock_disable_ipv6, mock_set_mtu, mock_arp_filtering, mock_ip_set,
|
|
mock_add):
|
|
"""This test validates that when adding an interface
|
|
to a bridge, the interface mtu is updated first
|
|
followed by the bridge. This is required to work around
|
|
https://bugs.launchpad.net/ubuntu/+source/linux/+bug/1399064
|
|
"""
|
|
mock_dev_exists.side_effect = [False, True]
|
|
linux_net._ensure_bridge_privileged("fake-bridge", "fake-interface",
|
|
None, False, mtu=1500)
|
|
calls = [mock.call('fake-interface', 1500),
|
|
mock.call('fake-bridge', 1500)]
|
|
mock_set_mtu.assert_has_calls(calls)
|
|
calls = [mock.call('fake-bridge', state = 'up'),
|
|
mock.call('fake-interface', master='fake-bridge', state='up',
|
|
check_exit_code=[0, 2, 254])]
|
|
mock_ip_set.assert_has_calls(calls)
|
|
|
|
@mock.patch('six.moves.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')
|