126 lines
5.3 KiB
Python
126 lines
5.3 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 contextlib
|
|
import mock
|
|
import os.path
|
|
import six
|
|
import testtools
|
|
|
|
import fixtures
|
|
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
|
|
|
|
if six.PY2:
|
|
nested = contextlib.nested
|
|
else:
|
|
@contextlib.contextmanager
|
|
def nested(*contexts):
|
|
with contextlib.ExitStack() as stack:
|
|
yield [stack.enter_context(c) for c in contexts]
|
|
|
|
|
|
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(processutils, "execute")
|
|
def test_set_device_mtu(self, execute):
|
|
linux_net._set_device_mtu(dev='fakedev', mtu=1500)
|
|
expected = ['ip', 'link', 'set', 'fakedev', 'mtu', 1500]
|
|
execute.assert_called_with(*expected, check_exit_code=mock.ANY)
|
|
|
|
@mock.patch.object(processutils, "execute")
|
|
@mock.patch.object(linux_net, "device_exists", return_value=False)
|
|
@mock.patch.object(linux_net, "_set_device_mtu")
|
|
def test_ensure_vlan(self, mock_set_mtu,
|
|
mock_dev_exists, mock_exec):
|
|
linux_net._ensure_vlan_privileged(123, 'fake-bridge',
|
|
mac_address='fake-mac',
|
|
mtu=1500)
|
|
self.assertTrue(mock_dev_exists.called)
|
|
calls = [mock.call('ip', 'link', 'add', 'link',
|
|
'fake-bridge', 'name', 'vlan123', 'type',
|
|
'vlan', 'id', 123,
|
|
check_exit_code=[0, 2, 254]),
|
|
mock.call('ip', 'link', 'set', 'vlan123',
|
|
'address', 'fake-mac',
|
|
check_exit_code=[0, 2, 254]),
|
|
mock.call('ip', 'link', 'set', 'vlan123', 'up',
|
|
check_exit_code=[0, 2, 254])]
|
|
self.assertEqual(calls, mock_exec.mock_calls)
|
|
mock_set_mtu.assert_called_once_with('vlan123', 1500)
|
|
|
|
@mock.patch.object(processutils, "execute")
|
|
@mock.patch.object(linux_net, "device_exists", return_value=False)
|
|
@mock.patch.object(linux_net, "_set_device_mtu")
|
|
def test_ensure_vlan_invalid_mtu(self, mock_set_mtu,
|
|
mock_dev_exists, mock_exec):
|
|
linux_net._ensure_vlan_privileged(123, 'fake-bridge',
|
|
mac_address='fake-mac', mtu=None)
|
|
self.assertFalse(mock_set_mtu.called)
|
|
|
|
@mock.patch.object(processutils, "execute")
|
|
@mock.patch.object(linux_net, "device_exists", return_value=True)
|
|
def test_ensure_bridge_exists(self, mock_dev_exists, mock_exec):
|
|
linux_net.ensure_bridge("br0", None, filtering=False)
|
|
|
|
self.assertEqual([], mock_exec.mock_calls)
|
|
mock_dev_exists.assert_called_once_with("br0")
|
|
|
|
@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_exec,
|
|
mock_path_exists):
|
|
linux_net.ensure_bridge("br0", None, filtering=False)
|
|
|
|
calls = [mock.call('brctl', 'addbr', 'br0'),
|
|
mock.call('brctl', 'setfd', 'br0', 0),
|
|
mock.call('brctl', 'stp', 'br0', "off"),
|
|
mock.call('ip', 'link', 'set', 'br0', "up")]
|
|
self.assertEqual(calls, mock_exec.mock_calls)
|
|
mock_dev_exists.assert_called_once_with("br0")
|
|
|
|
@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_exec,
|
|
mock_path_exists):
|
|
linux_net.ensure_bridge("br0", None, filtering=False)
|
|
|
|
calls = [mock.call('brctl', 'addbr', 'br0'),
|
|
mock.call('brctl', 'setfd', 'br0', 0),
|
|
mock.call('brctl', 'stp', 'br0', "off"),
|
|
mock.call('tee', '/proc/sys/net/ipv6/conf/br0/disable_ipv6',
|
|
check_exit_code=[0, 1], process_input='1'),
|
|
mock.call('ip', 'link', 'set', 'br0', "up")]
|
|
self.assertEqual(calls, mock_exec.mock_calls)
|
|
mock_dev_exists.assert_called_once_with("br0")
|