Bring the tests for the interface module and the modules it
depends on. DHC-1048 We copied the interface module from quantum into the akanda-rug without its test. This fix brings in the tests for the module, the modules it depends on and adapt them to run without quantun as dependency. Signed-off-by: Rosario Di Somma <rosario.disomma@dreamhost.com> Change-Id: I7136c9bd367532e12e8c58fb4469f100d10f4ed9
This commit is contained in:
Rosario Di Somma
parent
caab04c21f
commit
0892382a3c
|
|
@ -0,0 +1,92 @@
|
|||
# vim: tabstop=4 shiftwidth=4 softtabstop=4
|
||||
|
||||
# Copyright 2012, Nicira, Inc.
|
||||
#
|
||||
# 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.
|
||||
# @author: Dan Wendlandt, Nicira, Inc.
|
||||
|
||||
import fixtures
|
||||
import mock
|
||||
import testtools
|
||||
|
||||
from akanda.rug.common.linux import utils
|
||||
|
||||
|
||||
class AgentUtilsExecuteTest(testtools.TestCase):
|
||||
def setUp(self):
|
||||
super(AgentUtilsExecuteTest, self).setUp()
|
||||
self.root_helper = "echo"
|
||||
self.test_file = self.useFixture(
|
||||
fixtures.TempDir()).join("test_execute.tmp")
|
||||
open(self.test_file, 'w').close()
|
||||
|
||||
def test_without_helper(self):
|
||||
result = utils.execute(["ls", self.test_file])
|
||||
self.assertEqual(result, "%s\n" % self.test_file)
|
||||
|
||||
def test_with_helper(self):
|
||||
result = utils.execute(["ls", self.test_file],
|
||||
self.root_helper)
|
||||
self.assertEqual(result, "ls %s\n" % self.test_file)
|
||||
|
||||
def test_stderr(self):
|
||||
stdout, stderr = utils.execute(["ls", self.test_file],
|
||||
return_stderr=True)
|
||||
self.assertEqual(stdout, "%s\n" % self.test_file)
|
||||
self.assertEqual(stderr, "")
|
||||
|
||||
def test_check_exit_code(self):
|
||||
stdout = utils.execute(["ls", self.test_file[:-1]],
|
||||
check_exit_code=False)
|
||||
self.assertEqual(stdout, "")
|
||||
self.assertRaises(RuntimeError, utils.execute,
|
||||
["ls", self.test_file[:-1]])
|
||||
|
||||
def test_process_input(self):
|
||||
result = utils.execute(["cat"], process_input="%s\n" %
|
||||
self.test_file[:-1])
|
||||
self.assertEqual(result, "%s\n" % self.test_file[:-1])
|
||||
|
||||
def test_with_addl_env(self):
|
||||
result = utils.execute(["ls", self.test_file],
|
||||
addl_env={'foo': 'bar'})
|
||||
self.assertEqual(result, "%s\n" % self.test_file)
|
||||
|
||||
|
||||
class AgentUtilsGetInterfaceMAC(testtools.TestCase):
|
||||
def test_get_interface_mac(self):
|
||||
expect_val = '01:02:03:04:05:06'
|
||||
with mock.patch('fcntl.ioctl') as ioctl:
|
||||
ioctl.return_value = ''.join(['\x00' * 18,
|
||||
'\x01\x02\x03\x04\x05\x06',
|
||||
'\x00' * 232])
|
||||
actual_val = utils.get_interface_mac('eth0')
|
||||
self.assertEqual(actual_val, expect_val)
|
||||
|
||||
|
||||
class AgentUtilsReplaceFile(testtools.TestCase):
|
||||
def test_replace_file(self):
|
||||
# make file to replace
|
||||
with mock.patch('tempfile.NamedTemporaryFile') as ntf:
|
||||
ntf.return_value.name = '/baz'
|
||||
with mock.patch('os.chmod') as chmod:
|
||||
with mock.patch('os.rename') as rename:
|
||||
utils.replace_file('/foo', 'bar')
|
||||
|
||||
expected = [mock.call('w+', dir='/', delete=False),
|
||||
mock.call().write('bar'),
|
||||
mock.call().close()]
|
||||
|
||||
ntf.assert_has_calls(expected)
|
||||
chmod.assert_called_once_with('/baz', 0644)
|
||||
rename.assert_called_once_with('/baz', '/foo')
|
||||
|
|
@ -0,0 +1,251 @@
|
|||
# vim: tabstop=4 shiftwidth=4 softtabstop=4
|
||||
|
||||
# Copyright 2012 OpenStack LLC
|
||||
# All Rights Reserved.
|
||||
#
|
||||
# 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 unittest
|
||||
|
||||
import mock
|
||||
from oslo.config import cfg
|
||||
|
||||
from akanda.rug.common.linux import interface
|
||||
from akanda.rug.common.linux import ip_lib
|
||||
from akanda.rug.common.linux import utils
|
||||
|
||||
|
||||
class BaseChild(interface.LinuxInterfaceDriver):
|
||||
def plug(*args):
|
||||
pass
|
||||
|
||||
def unplug(*args):
|
||||
pass
|
||||
|
||||
|
||||
class FakeNetwork:
|
||||
id = '12345678-1234-5678-90ab-ba0987654321'
|
||||
|
||||
|
||||
class FakeSubnet:
|
||||
cidr = '192.168.1.1/24'
|
||||
|
||||
|
||||
class FakeAllocation:
|
||||
subnet = FakeSubnet()
|
||||
ip_address = '192.168.1.2'
|
||||
ip_version = 4
|
||||
|
||||
|
||||
class FakePort:
|
||||
id = 'abcdef01-1234-5678-90ab-ba0987654321'
|
||||
fixed_ips = [FakeAllocation]
|
||||
device_id = 'cccccccc-cccc-cccc-cccc-cccccccccccc'
|
||||
network = FakeNetwork()
|
||||
network_id = network.id
|
||||
|
||||
|
||||
class TestBase(unittest.TestCase):
|
||||
def setUp(self):
|
||||
root_helper_opt = [
|
||||
cfg.StrOpt('root_helper', default='sudo'),
|
||||
]
|
||||
self.conf = cfg.CONF
|
||||
self.conf.register_opts(root_helper_opt)
|
||||
self.ip_dev_p = mock.patch.object(ip_lib, 'IPDevice')
|
||||
self.ip_dev = self.ip_dev_p.start()
|
||||
self.ip_p = mock.patch.object(ip_lib, 'IPWrapper')
|
||||
self.ip = self.ip_p.start()
|
||||
self.device_exists_p = mock.patch.object(ip_lib, 'device_exists')
|
||||
self.device_exists = self.device_exists_p.start()
|
||||
|
||||
def tearDown(self):
|
||||
# sometimes a test may turn this off
|
||||
try:
|
||||
self.device_exists_p.stop()
|
||||
except RuntimeError:
|
||||
pass
|
||||
self.ip_dev_p.stop()
|
||||
self.ip_p.stop()
|
||||
# Note(rods): some tests override the default options so let's reset
|
||||
# the config object
|
||||
self.conf.reset()
|
||||
|
||||
|
||||
class TestABCDriver(TestBase):
|
||||
def test_get_device_name(self):
|
||||
bc = BaseChild(self.conf)
|
||||
device_name = bc.get_device_name(FakePort())
|
||||
self.assertEqual('tapabcdef01-12', device_name)
|
||||
|
||||
def test_l3_init(self):
|
||||
addresses = [dict(ip_version=4, scope='global',
|
||||
dynamic=False, cidr='172.16.77.240/24')]
|
||||
self.ip_dev().addr.list = mock.Mock(return_value=addresses)
|
||||
|
||||
bc = BaseChild(self.conf)
|
||||
ns = '12345678-1234-5678-90ab-ba0987654321'
|
||||
bc.init_l3('tap0', ['192.168.1.2/24'], namespace=ns)
|
||||
self.ip_dev.assert_has_calls(
|
||||
[mock.call('tap0', 'sudo', namespace=ns),
|
||||
mock.call().addr.list(scope='global', filters=['permanent']),
|
||||
mock.call().addr.add(4, '192.168.1.2/24', '192.168.1.255'),
|
||||
mock.call().addr.delete(4, '172.16.77.240/24')])
|
||||
|
||||
|
||||
class TestOVSInterfaceDriver(TestBase):
|
||||
|
||||
def test_plug_no_ns(self):
|
||||
self._test_plug()
|
||||
|
||||
def test_plug_with_ns(self):
|
||||
self._test_plug(namespace='01234567-1234-1234-99')
|
||||
|
||||
def test_plug_alt_bridge(self):
|
||||
self._test_plug(bridge='br-foo')
|
||||
|
||||
def _test_plug(self, additional_expectation=[], bridge=None,
|
||||
namespace=None):
|
||||
|
||||
if not bridge:
|
||||
bridge = 'br-int'
|
||||
|
||||
def device_exists(dev, root_helper=None, namespace=None):
|
||||
return dev == bridge
|
||||
|
||||
vsctl_cmd = ['ovs-vsctl', '--', '--may-exist', 'add-port',
|
||||
bridge, 'tap0', '--', 'set', 'Interface', 'tap0',
|
||||
'type=internal', '--', 'set', 'Interface', 'tap0',
|
||||
'external-ids:iface-id=port-1234', '--', 'set',
|
||||
'Interface', 'tap0',
|
||||
'external-ids:iface-status=active', '--', 'set',
|
||||
'Interface', 'tap0',
|
||||
'external-ids:attached-mac=aa:bb:cc:dd:ee:ff']
|
||||
|
||||
with mock.patch.object(utils, 'execute') as execute:
|
||||
ovs = interface.OVSInterfaceDriver(self.conf)
|
||||
self.device_exists.side_effect = device_exists
|
||||
ovs.plug('01234567-1234-1234-99',
|
||||
'port-1234',
|
||||
'tap0',
|
||||
'aa:bb:cc:dd:ee:ff',
|
||||
bridge=bridge,
|
||||
namespace=namespace)
|
||||
execute.assert_called_once_with(vsctl_cmd, 'sudo')
|
||||
|
||||
expected = [mock.call('sudo'),
|
||||
mock.call().device('tap0'),
|
||||
mock.call().device().link.set_address('aa:bb:cc:dd:ee:ff')]
|
||||
expected.extend(additional_expectation)
|
||||
if namespace:
|
||||
expected.extend(
|
||||
[mock.call().ensure_namespace(namespace),
|
||||
mock.call().ensure_namespace().add_device_to_namespace(
|
||||
mock.ANY)])
|
||||
expected.extend([mock.call().device().link.set_up()])
|
||||
|
||||
self.ip.assert_has_calls(expected)
|
||||
|
||||
def test_plug_mtu(self):
|
||||
self.conf.set_override('network_device_mtu', 9000)
|
||||
self._test_plug([mock.call().device().link.set_mtu(9000)])
|
||||
|
||||
def test_unplug(self, bridge=None):
|
||||
if not bridge:
|
||||
bridge = 'br-int'
|
||||
with mock.patch('akanda.rug.common.linux.ovs_lib.OVSBridge') as ovs_br:
|
||||
ovs = interface.OVSInterfaceDriver(self.conf)
|
||||
ovs.unplug('tap0')
|
||||
ovs_br.assert_has_calls([mock.call(bridge, 'sudo'),
|
||||
mock.call().delete_port('tap0')])
|
||||
|
||||
|
||||
class TestBridgeInterfaceDriver(TestBase):
|
||||
def test_get_device_name(self):
|
||||
br = interface.BridgeInterfaceDriver(self.conf)
|
||||
device_name = br.get_device_name(FakePort())
|
||||
self.assertEqual('ns-abcdef01-12', device_name)
|
||||
|
||||
def test_plug_no_ns(self):
|
||||
self._test_plug()
|
||||
|
||||
def test_plug_with_ns(self):
|
||||
self._test_plug(namespace='01234567-1234-1234-99')
|
||||
|
||||
def _test_plug(self, namespace=None):
|
||||
def device_exists(device, root_helper=None, namespace=None):
|
||||
return device.startswith('brq')
|
||||
|
||||
root_veth = mock.Mock()
|
||||
ns_veth = mock.Mock()
|
||||
|
||||
self.ip().add_veth = mock.Mock(return_value=(root_veth, ns_veth))
|
||||
|
||||
self.device_exists.side_effect = device_exists
|
||||
br = interface.BridgeInterfaceDriver(self.conf)
|
||||
br.plug('01234567-1234-1234-99',
|
||||
'port-1234',
|
||||
'ns-0',
|
||||
'aa:bb:cc:dd:ee:ff',
|
||||
namespace=namespace)
|
||||
|
||||
ip_calls = [mock.call('sudo'), mock.call().add_veth('tap0', 'ns-0')]
|
||||
if namespace:
|
||||
ip_calls.extend([
|
||||
mock.call().ensure_namespace('01234567-1234-1234-99'),
|
||||
mock.call().ensure_namespace().add_device_to_namespace(
|
||||
ns_veth)])
|
||||
|
||||
self.ip.assert_has_calls(ip_calls)
|
||||
|
||||
root_veth.assert_has_calls([mock.call.link.set_up()])
|
||||
ns_veth.assert_has_calls([mock.call.link.set_up()])
|
||||
|
||||
def test_plug_dev_exists(self):
|
||||
self.device_exists.return_value = True
|
||||
with mock.patch('akanda.rug.common.linux.interface.LOG.warn') as log:
|
||||
br = interface.BridgeInterfaceDriver(self.conf)
|
||||
br.plug('01234567-1234-1234-99',
|
||||
'port-1234',
|
||||
'tap0',
|
||||
'aa:bb:cc:dd:ee:ff')
|
||||
self.ip_dev.assert_has_calls([])
|
||||
self.assertEquals(log.call_count, 1)
|
||||
|
||||
def test_unplug(self):
|
||||
self.device_exists.return_value = True
|
||||
with mock.patch('akanda.rug.common.linux.interface.LOG.debug') as log:
|
||||
br = interface.BridgeInterfaceDriver(self.conf)
|
||||
br.unplug('tap0')
|
||||
log.assert_called_once()
|
||||
self.execute.assert_has_calls(
|
||||
[mock.call(['ip', 'link', 'delete', 'tap0'], 'sudo')])
|
||||
|
||||
def test_unplug_no_device(self):
|
||||
self.device_exists.return_value = False
|
||||
self.ip_dev().link.delete.side_effect = RuntimeError
|
||||
with mock.patch('akanda.rug.common.linux.interface.LOG') as log:
|
||||
br = interface.BridgeInterfaceDriver(self.conf)
|
||||
br.unplug('tap0')
|
||||
[mock.call(), mock.call('tap0', 'sudo'), mock.call().link.delete()]
|
||||
self.assertEqual(log.error.call_count, 1)
|
||||
|
||||
def test_unplug(self):
|
||||
self.device_exists.return_value = True
|
||||
with mock.patch('akanda.rug.common.linux.interface.LOG.debug') as log:
|
||||
br = interface.BridgeInterfaceDriver(self.conf)
|
||||
br.unplug('tap0')
|
||||
log.assert_called_once()
|
||||
|
||||
self.ip_dev.assert_has_calls([mock.call('tap0', 'sudo', None),
|
||||
mock.call().link.delete()])
|
||||
|
|
@ -0,0 +1,680 @@
|
|||
# vim: tabstop=4 shiftwidth=4 softtabstop=4
|
||||
|
||||
# Copyright 2012 OpenStack Foundation
|
||||
# All Rights Reserved.
|
||||
#
|
||||
# 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 unittest
|
||||
|
||||
import mock
|
||||
|
||||
from akanda.rug.common.linux import ip_lib
|
||||
|
||||
|
||||
NETNS_SAMPLE = [
|
||||
'12345678-1234-5678-abcd-1234567890ab',
|
||||
'bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb',
|
||||
'cccccccc-cccc-cccc-cccc-cccccccccccc']
|
||||
|
||||
LINK_SAMPLE = [
|
||||
'1: lo: <LOOPBACK,UP,LOWER_UP> mtu 16436 qdisc noqueue state UNKNOWN \\'
|
||||
'link/loopback 00:00:00:00:00:00 brd 00:00:00:00:00:00',
|
||||
'2: eth0: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 qdisc mq state UP '
|
||||
'qlen 1000\ link/ether cc:dd:ee:ff:ab:cd brd ff:ff:ff:ff:ff:ff'
|
||||
'\ alias openvswitch',
|
||||
'3: br-int: <BROADCAST,MULTICAST> mtu 1500 qdisc noop state DOWN '
|
||||
'\ link/ether aa:bb:cc:dd:ee:ff brd ff:ff:ff:ff:ff:ff',
|
||||
'4: gw-ddc717df-49: <BROADCAST,MULTICAST> mtu 1500 qdisc noop '
|
||||
'state DOWN \ link/ether fe:dc:ba:fe:dc:ba brd ff:ff:ff:ff:ff:ff']
|
||||
|
||||
ADDR_SAMPLE = ("""
|
||||
2: eth0: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 qdisc mq state UP qlen 1000
|
||||
link/ether dd:cc:aa:b9:76:ce brd ff:ff:ff:ff:ff:ff
|
||||
inet 172.16.77.240/24 brd 172.16.77.255 scope global eth0
|
||||
inet6 2001:470:9:1224:5595:dd51:6ba2:e788/64 scope global temporary dynamic
|
||||
valid_lft 14187sec preferred_lft 3387sec
|
||||
inet6 2001:470:9:1224:fd91:272:581e:3a32/64 scope global temporary """
|
||||
"""deprecated dynamic
|
||||
valid_lft 14187sec preferred_lft 0sec
|
||||
inet6 2001:470:9:1224:4508:b885:5fb:740b/64 scope global temporary """
|
||||
"""deprecated dynamic
|
||||
valid_lft 14187sec preferred_lft 0sec
|
||||
inet6 2001:470:9:1224:dfcc:aaff:feb9:76ce/64 scope global dynamic
|
||||
valid_lft 14187sec preferred_lft 3387sec
|
||||
inet6 fe80::dfcc:aaff:feb9:76ce/64 scope link
|
||||
valid_lft forever preferred_lft forever
|
||||
""")
|
||||
|
||||
ADDR_SAMPLE2 = ("""
|
||||
2: eth0: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 qdisc mq state UP qlen 1000
|
||||
link/ether dd:cc:aa:b9:76:ce brd ff:ff:ff:ff:ff:ff
|
||||
inet 172.16.77.240/24 scope global eth0
|
||||
inet6 2001:470:9:1224:5595:dd51:6ba2:e788/64 scope global temporary dynamic
|
||||
valid_lft 14187sec preferred_lft 3387sec
|
||||
inet6 2001:470:9:1224:fd91:272:581e:3a32/64 scope global temporary """
|
||||
"""deprecated dynamic
|
||||
valid_lft 14187sec preferred_lft 0sec
|
||||
inet6 2001:470:9:1224:4508:b885:5fb:740b/64 scope global temporary """
|
||||
"""deprecated dynamic
|
||||
valid_lft 14187sec preferred_lft 0sec
|
||||
inet6 2001:470:9:1224:dfcc:aaff:feb9:76ce/64 scope global dynamic
|
||||
valid_lft 14187sec preferred_lft 3387sec
|
||||
inet6 fe80::dfcc:aaff:feb9:76ce/64 scope link
|
||||
valid_lft forever preferred_lft forever
|
||||
""")
|
||||
|
||||
GATEWAY_SAMPLE1 = ("""
|
||||
default via 10.35.19.254 metric 100
|
||||
10.35.16.0/22 proto kernel scope link src 10.35.17.97
|
||||
""")
|
||||
|
||||
GATEWAY_SAMPLE2 = ("""
|
||||
default via 10.35.19.254 metric 100
|
||||
""")
|
||||
|
||||
GATEWAY_SAMPLE3 = ("""
|
||||
10.35.16.0/22 proto kernel scope link src 10.35.17.97
|
||||
""")
|
||||
|
||||
GATEWAY_SAMPLE4 = ("""
|
||||
default via 10.35.19.254
|
||||
""")
|
||||
|
||||
DEVICE_ROUTE_SAMPLE = ("10.0.0.0/24 scope link src 10.0.0.2")
|
||||
|
||||
SUBNET_SAMPLE1 = ("10.0.0.0/24 dev qr-23380d11-d2 scope link src 10.0.0.1\n"
|
||||
"10.0.0.0/24 dev tap1d7888a7-10 scope link src 10.0.0.2")
|
||||
SUBNET_SAMPLE2 = ("10.0.0.0/24 dev tap1d7888a7-10 scope link src 10.0.0.2\n"
|
||||
"10.0.0.0/24 dev qr-23380d11-d2 scope link src 10.0.0.1")
|
||||
|
||||
|
||||
class TestSubProcessBase(unittest.TestCase):
|
||||
def setUp(self):
|
||||
super(TestSubProcessBase, self).setUp()
|
||||
self.execute_p = mock.patch('akanda.rug.common.linux.utils.execute')
|
||||
self.execute = self.execute_p.start()
|
||||
self.addCleanup(self.execute_p.stop)
|
||||
|
||||
def test_execute_wrapper(self):
|
||||
ip_lib.SubProcessBase._execute('o', 'link', ('list',), 'sudo')
|
||||
|
||||
self.execute.assert_called_once_with(['ip', '-o', 'link', 'list'],
|
||||
root_helper='sudo')
|
||||
|
||||
def test_execute_wrapper_int_options(self):
|
||||
ip_lib.SubProcessBase._execute([4], 'link', ('list',))
|
||||
|
||||
self.execute.assert_called_once_with(['ip', '-4', 'link', 'list'],
|
||||
root_helper=None)
|
||||
|
||||
def test_execute_wrapper_no_options(self):
|
||||
ip_lib.SubProcessBase._execute([], 'link', ('list',))
|
||||
|
||||
self.execute.assert_called_once_with(['ip', 'link', 'list'],
|
||||
root_helper=None)
|
||||
|
||||
def test_run_no_namespace(self):
|
||||
base = ip_lib.SubProcessBase('sudo')
|
||||
base._run([], 'link', ('list',))
|
||||
self.execute.assert_called_once_with(['ip', 'link', 'list'],
|
||||
root_helper=None)
|
||||
|
||||
def test_run_namespace(self):
|
||||
base = ip_lib.SubProcessBase('sudo', 'ns')
|
||||
base._run([], 'link', ('list',))
|
||||
self.execute.assert_called_once_with(['ip', 'netns', 'exec', 'ns',
|
||||
'ip', 'link', 'list'],
|
||||
root_helper='sudo')
|
||||
|
||||
def test_as_root_namespace(self):
|
||||
base = ip_lib.SubProcessBase('sudo', 'ns')
|
||||
base._as_root([], 'link', ('list',))
|
||||
self.execute.assert_called_once_with(['ip', 'netns', 'exec', 'ns',
|
||||
'ip', 'link', 'list'],
|
||||
root_helper='sudo')
|
||||
|
||||
def test_as_root_no_root_helper(self):
|
||||
base = ip_lib.SubProcessBase()
|
||||
self.assertRaisesRegexp(Exception,
|
||||
'Sudo is required to run this command',
|
||||
base._as_root,
|
||||
[], 'link', ('list',))
|
||||
|
||||
|
||||
class TestIpWrapper(unittest.TestCase):
|
||||
def setUp(self):
|
||||
super(TestIpWrapper, self).setUp()
|
||||
self.execute_p = mock.patch.object(ip_lib.IPWrapper, '_execute')
|
||||
self.execute = self.execute_p.start()
|
||||
self.addCleanup(self.execute_p.stop)
|
||||
|
||||
def test_get_devices(self):
|
||||
self.execute.return_value = '\n'.join(LINK_SAMPLE)
|
||||
retval = ip_lib.IPWrapper('sudo').get_devices()
|
||||
self.assertEqual(retval,
|
||||
[ip_lib.IPDevice('lo'),
|
||||
ip_lib.IPDevice('eth0'),
|
||||
ip_lib.IPDevice('br-int'),
|
||||
ip_lib.IPDevice('gw-ddc717df-49')])
|
||||
|
||||
self.execute.assert_called_once_with('o', 'link', ('list',),
|
||||
'sudo', None)
|
||||
|
||||
def test_get_devices_malformed_line(self):
|
||||
self.execute.return_value = '\n'.join(LINK_SAMPLE + ['gibberish'])
|
||||
retval = ip_lib.IPWrapper('sudo').get_devices()
|
||||
self.assertEqual(retval,
|
||||
[ip_lib.IPDevice('lo'),
|
||||
ip_lib.IPDevice('eth0'),
|
||||
ip_lib.IPDevice('br-int'),
|
||||
ip_lib.IPDevice('gw-ddc717df-49')])
|
||||
|
||||
self.execute.assert_called_once_with('o', 'link', ('list',),
|
||||
'sudo', None)
|
||||
|
||||
def test_get_namespaces(self):
|
||||
self.execute.return_value = '\n'.join(NETNS_SAMPLE)
|
||||
retval = ip_lib.IPWrapper.get_namespaces('sudo')
|
||||
self.assertEqual(retval,
|
||||
['12345678-1234-5678-abcd-1234567890ab',
|
||||
'bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb',
|
||||
'cccccccc-cccc-cccc-cccc-cccccccccccc'])
|
||||
|
||||
self.execute.assert_called_once_with('', 'netns', ('list',),
|
||||
root_helper='sudo')
|
||||
|
||||
def test_add_tuntap(self):
|
||||
ip_lib.IPWrapper('sudo').add_tuntap('tap0')
|
||||
self.execute.assert_called_once_with('', 'tuntap',
|
||||
('add', 'tap0', 'mode', 'tap'),
|
||||
'sudo', None)
|
||||
|
||||
def test_add_veth(self):
|
||||
ip_lib.IPWrapper('sudo').add_veth('tap0', 'tap1')
|
||||
self.execute.assert_called_once_with('', 'link',
|
||||
('add', 'tap0', 'type', 'veth',
|
||||
'peer', 'name', 'tap1'),
|
||||
'sudo', None)
|
||||
|
||||
def test_get_device(self):
|
||||
dev = ip_lib.IPWrapper('sudo', 'ns').device('eth0')
|
||||
self.assertEqual(dev.root_helper, 'sudo')
|
||||
self.assertEqual(dev.namespace, 'ns')
|
||||
self.assertEqual(dev.name, 'eth0')
|
||||
|
||||
def test_ensure_namespace(self):
|
||||
with mock.patch.object(ip_lib, 'IPDevice') as ip_dev:
|
||||
ip = ip_lib.IPWrapper('sudo')
|
||||
with mock.patch.object(ip.netns, 'exists') as ns_exists:
|
||||
ns_exists.return_value = False
|
||||
ns = ip.ensure_namespace('ns')
|
||||
self.execute.assert_has_calls(
|
||||
[mock.call([], 'netns', ('add', 'ns'), 'sudo', None)])
|
||||
ip_dev.assert_has_calls([mock.call('lo', 'sudo', 'ns'),
|
||||
mock.call().link.set_up()])
|
||||
|
||||
def test_ensure_namespace_existing(self):
|
||||
with mock.patch.object(ip_lib, 'IpNetnsCommand') as ip_ns_cmd:
|
||||
ip_ns_cmd.exists.return_value = True
|
||||
ns = ip_lib.IPWrapper('sudo').ensure_namespace('ns')
|
||||
self.assertFalse(self.execute.called)
|
||||
self.assertEqual(ns.namespace, 'ns')
|
||||
|
||||
def test_namespace_is_empty_no_devices(self):
|
||||
ip = ip_lib.IPWrapper('sudo', 'ns')
|
||||
with mock.patch.object(ip, 'get_devices') as get_devices:
|
||||
get_devices.return_value = []
|
||||
|
||||
self.assertTrue(ip.namespace_is_empty())
|
||||
get_devices.assert_called_once_with(exclude_loopback=True)
|
||||
|
||||
def test_namespace_is_empty(self):
|
||||
ip = ip_lib.IPWrapper('sudo', 'ns')
|
||||
with mock.patch.object(ip, 'get_devices') as get_devices:
|
||||
get_devices.return_value = [mock.Mock()]
|
||||
|
||||
self.assertFalse(ip.namespace_is_empty())
|
||||
get_devices.assert_called_once_with(exclude_loopback=True)
|
||||
|
||||
def test_garbage_collect_namespace_does_not_exist(self):
|
||||
with mock.patch.object(ip_lib, 'IpNetnsCommand') as ip_ns_cmd_cls:
|
||||
ip_ns_cmd_cls.return_value.exists.return_value = False
|
||||
ip = ip_lib.IPWrapper('sudo', 'ns')
|
||||
with mock.patch.object(ip, 'namespace_is_empty') as mock_is_empty:
|
||||
|
||||
self.assertFalse(ip.garbage_collect_namespace())
|
||||
ip_ns_cmd_cls.assert_has_calls([mock.call().exists('ns')])
|
||||
self.assertNotIn(mock.call().delete('ns'),
|
||||
ip_ns_cmd_cls.return_value.mock_calls)
|
||||
self.assertEqual(mock_is_empty.mock_calls, [])
|
||||
|
||||
def test_garbage_collect_namespace_existing_empty_ns(self):
|
||||
with mock.patch.object(ip_lib, 'IpNetnsCommand') as ip_ns_cmd_cls:
|
||||
ip_ns_cmd_cls.return_value.exists.return_value = True
|
||||
|
||||
ip = ip_lib.IPWrapper('sudo', 'ns')
|
||||
|
||||
with mock.patch.object(ip, 'namespace_is_empty') as mock_is_empty:
|
||||
mock_is_empty.return_value = True
|
||||
self.assertTrue(ip.garbage_collect_namespace())
|
||||
|
||||
mock_is_empty.assert_called_once_with()
|
||||
expected = [mock.call().exists('ns'),
|
||||
mock.call().delete('ns')]
|
||||
ip_ns_cmd_cls.assert_has_calls(expected)
|
||||
|
||||
def test_garbage_collect_namespace_existing_not_empty(self):
|
||||
lo_device = mock.Mock()
|
||||
lo_device.name = 'lo'
|
||||
tap_device = mock.Mock()
|
||||
tap_device.name = 'tap1'
|
||||
|
||||
with mock.patch.object(ip_lib, 'IpNetnsCommand') as ip_ns_cmd_cls:
|
||||
ip_ns_cmd_cls.return_value.exists.return_value = True
|
||||
|
||||
ip = ip_lib.IPWrapper('sudo', 'ns')
|
||||
|
||||
with mock.patch.object(ip, 'namespace_is_empty') as mock_is_empty:
|
||||
mock_is_empty.return_value = False
|
||||
|
||||
self.assertFalse(ip.garbage_collect_namespace())
|
||||
|
||||
mock_is_empty.assert_called_once_with()
|
||||
expected = [mock.call(ip),
|
||||
mock.call().exists('ns')]
|
||||
self.assertEqual(ip_ns_cmd_cls.mock_calls, expected)
|
||||
self.assertNotIn(mock.call().delete('ns'),
|
||||
ip_ns_cmd_cls.mock_calls)
|
||||
|
||||
def test_add_device_to_namespace(self):
|
||||
dev = mock.Mock()
|
||||
ip_lib.IPWrapper('sudo', 'ns').add_device_to_namespace(dev)
|
||||
dev.assert_has_calls([mock.call.link.set_netns('ns')])
|
||||
|
||||
def test_add_device_to_namespace_is_none(self):
|
||||
dev = mock.Mock()
|
||||
ip_lib.IPWrapper('sudo').add_device_to_namespace(dev)
|
||||
self.assertEqual(dev.mock_calls, [])
|
||||
|
||||
|
||||
class TestIPDevice(unittest.TestCase):
|
||||
def test_eq_same_name(self):
|
||||
dev1 = ip_lib.IPDevice('tap0')
|
||||
dev2 = ip_lib.IPDevice('tap0')
|
||||
self.assertEqual(dev1, dev2)
|
||||
|
||||
def test_eq_diff_name(self):
|
||||
dev1 = ip_lib.IPDevice('tap0')
|
||||
dev2 = ip_lib.IPDevice('tap1')
|
||||
self.assertNotEqual(dev1, dev2)
|
||||
|
||||
def test_eq_same_namespace(self):
|
||||
dev1 = ip_lib.IPDevice('tap0', 'ns1')
|
||||
dev2 = ip_lib.IPDevice('tap0', 'ns1')
|
||||
self.assertEqual(dev1, dev2)
|
||||
|
||||
def test_eq_diff_namespace(self):
|
||||
dev1 = ip_lib.IPDevice('tap0', 'sudo', 'ns1')
|
||||
dev2 = ip_lib.IPDevice('tap0', 'sudo', 'ns2')
|
||||
self.assertNotEqual(dev1, dev2)
|
||||
|
||||
def test_eq_other_is_none(self):
|
||||
dev1 = ip_lib.IPDevice('tap0', 'sudo', 'ns1')
|
||||
self.assertNotEqual(dev1, None)
|
||||
|
||||
def test_str(self):
|
||||
self.assertEqual(str(ip_lib.IPDevice('tap0')), 'tap0')
|
||||
|
||||
|
||||
class TestIPCommandBase(unittest.TestCase):
|
||||
def setUp(self):
|
||||
super(TestIPCommandBase, self).setUp()
|
||||
self.ip = mock.Mock()
|
||||
self.ip.root_helper = 'sudo'
|
||||
self.ip.namespace = 'namespace'
|
||||
self.ip_cmd = ip_lib.IpCommandBase(self.ip)
|
||||
self.ip_cmd.COMMAND = 'foo'
|
||||
|
||||
def test_run(self):
|
||||
self.ip_cmd._run('link', 'show')
|
||||
self.ip.assert_has_calls([mock.call._run([], 'foo', ('link', 'show'))])
|
||||
|
||||
def test_run_with_options(self):
|
||||
self.ip_cmd._run('link', options='o')
|
||||
self.ip.assert_has_calls([mock.call._run('o', 'foo', ('link', ))])
|
||||
|
||||
def test_as_root(self):
|
||||
self.ip_cmd._as_root('link')
|
||||
self.ip.assert_has_calls(
|
||||
[mock.call._as_root([], 'foo', ('link', ), False)])
|
||||
|
||||
def test_as_root_with_options(self):
|
||||
self.ip_cmd._as_root('link', options='o')
|
||||
self.ip.assert_has_calls(
|
||||
[mock.call._as_root('o', 'foo', ('link', ), False)])
|
||||
|
||||
|
||||
class TestIPDeviceCommandBase(unittest.TestCase):
|
||||
def setUp(self):
|
||||
super(TestIPDeviceCommandBase, self).setUp()
|
||||
self.ip_dev = mock.Mock()
|
||||
self.ip_dev.name = 'eth0'
|
||||
self.ip_dev.root_helper = 'sudo'
|
||||
self.ip_dev._execute = mock.Mock(return_value='executed')
|
||||
self.ip_cmd = ip_lib.IpDeviceCommandBase(self.ip_dev)
|
||||
self.ip_cmd.COMMAND = 'foo'
|
||||
|
||||
def test_name_property(self):
|
||||
self.assertEqual(self.ip_cmd.name, 'eth0')
|
||||
|
||||
|
||||
class TestIPCmdBase(unittest.TestCase):
|
||||
def setUp(self):
|
||||
super(TestIPCmdBase, self).setUp()
|
||||
self.parent = mock.Mock()
|
||||
self.parent.name = 'eth0'
|
||||
self.parent.root_helper = 'sudo'
|
||||
|
||||
def _assert_call(self, options, args):
|
||||
self.parent.assert_has_calls([
|
||||
mock.call._run(options, self.command, args)])
|
||||
|
||||
def _assert_sudo(self, options, args, force_root_namespace=False):
|
||||
self.parent.assert_has_calls(
|
||||
[mock.call._as_root(options, self.command, args,
|
||||
force_root_namespace)])
|
||||
|
||||
|
||||
class TestIpLinkCommand(TestIPCmdBase):
|
||||
def setUp(self):
|
||||
super(TestIpLinkCommand, self).setUp()
|
||||
self.parent._run.return_value = LINK_SAMPLE[1]
|
||||
self.command = 'link'
|
||||
self.link_cmd = ip_lib.IpLinkCommand(self.parent)
|
||||
|
||||
def test_set_address(self):
|
||||
self.link_cmd.set_address('aa:bb:cc:dd:ee:ff')
|
||||
self._assert_sudo([], ('set', 'eth0', 'address', 'aa:bb:cc:dd:ee:ff'))
|
||||
|
||||
def test_set_mtu(self):
|
||||
self.link_cmd.set_mtu(1500)
|
||||
self._assert_sudo([], ('set', 'eth0', 'mtu', 1500))
|
||||
|
||||
def test_set_up(self):
|
||||
self.link_cmd.set_up()
|
||||
self._assert_sudo([], ('set', 'eth0', 'up'))
|
||||
|
||||
def test_set_down(self):
|
||||
self.link_cmd.set_down()
|
||||
self._assert_sudo([], ('set', 'eth0', 'down'))
|
||||
|
||||
def test_set_netns(self):
|
||||
self.link_cmd.set_netns('foo')
|
||||
self._assert_sudo([], ('set', 'eth0', 'netns', 'foo'))
|
||||
self.assertEqual(self.parent.namespace, 'foo')
|
||||
|
||||
def test_set_name(self):
|
||||
self.link_cmd.set_name('tap1')
|
||||
self._assert_sudo([], ('set', 'eth0', 'name', 'tap1'))
|
||||
self.assertEqual(self.parent.name, 'tap1')
|
||||
|
||||
def test_set_alias(self):
|
||||
self.link_cmd.set_alias('openvswitch')
|
||||
self._assert_sudo([], ('set', 'eth0', 'alias', 'openvswitch'))
|
||||
|
||||
def test_delete(self):
|
||||
self.link_cmd.delete()
|
||||
self._assert_sudo([], ('delete', 'eth0'))
|
||||
|
||||
def test_address_property(self):
|
||||
self.parent._execute = mock.Mock(return_value=LINK_SAMPLE[1])
|
||||
self.assertEqual(self.link_cmd.address, 'cc:dd:ee:ff:ab:cd')
|
||||
|
||||
def test_mtu_property(self):
|
||||
self.parent._execute = mock.Mock(return_value=LINK_SAMPLE[1])
|
||||
self.assertEqual(self.link_cmd.mtu, 1500)
|
||||
|
||||
def test_qdisc_property(self):
|
||||
self.parent._execute = mock.Mock(return_value=LINK_SAMPLE[1])
|
||||
self.assertEqual(self.link_cmd.qdisc, 'mq')
|
||||
|
||||
def test_qlen_property(self):
|
||||
self.parent._execute = mock.Mock(return_value=LINK_SAMPLE[1])
|
||||
self.assertEqual(self.link_cmd.qlen, 1000)
|
||||
|
||||
def test_alias_property(self):
|
||||
self.parent._execute = mock.Mock(return_value=LINK_SAMPLE[1])
|
||||
self.assertEqual(self.link_cmd.alias, 'openvswitch')
|
||||
|
||||
def test_state_property(self):
|
||||
self.parent._execute = mock.Mock(return_value=LINK_SAMPLE[1])
|
||||
self.assertEqual(self.link_cmd.state, 'UP')
|
||||
|
||||
def test_settings_property(self):
|
||||
expected = {'mtu': 1500,
|
||||
'qlen': 1000,
|
||||
'state': 'UP',
|
||||
'qdisc': 'mq',
|
||||
'brd': 'ff:ff:ff:ff:ff:ff',
|
||||
'link/ether': 'cc:dd:ee:ff:ab:cd',
|
||||
'alias': 'openvswitch'}
|
||||
self.parent._execute = mock.Mock(return_value=LINK_SAMPLE[1])
|
||||
self.assertEqual(self.link_cmd.attributes, expected)
|
||||
self._assert_call('o', ('show', 'eth0'))
|
||||
|
||||
|
||||
class TestIpAddrCommand(TestIPCmdBase):
|
||||
def setUp(self):
|
||||
super(TestIpAddrCommand, self).setUp()
|
||||
self.parent.name = 'tap0'
|
||||
self.command = 'addr'
|
||||
self.addr_cmd = ip_lib.IpAddrCommand(self.parent)
|
||||
|
||||
def test_add_address(self):
|
||||
self.addr_cmd.add(4, '192.168.45.100/24', '192.168.45.255')
|
||||
self._assert_sudo([4],
|
||||
('add', '192.168.45.100/24', 'brd', '192.168.45.255',
|
||||
'scope', 'global', 'dev', 'tap0'))
|
||||
|
||||
def test_add_address_scoped(self):
|
||||
self.addr_cmd.add(4, '192.168.45.100/24', '192.168.45.255',
|
||||
scope='link')
|
||||
self._assert_sudo([4],
|
||||
('add', '192.168.45.100/24', 'brd', '192.168.45.255',
|
||||
'scope', 'link', 'dev', 'tap0'))
|
||||
|
||||
def test_del_address(self):
|
||||
self.addr_cmd.delete(4, '192.168.45.100/24')
|
||||
self._assert_sudo([4],
|
||||
('del', '192.168.45.100/24', 'dev', 'tap0'))
|
||||
|
||||
def test_flush(self):
|
||||
self.addr_cmd.flush()
|
||||
self._assert_sudo([], ('flush', 'tap0'))
|
||||
|
||||
def test_list(self):
|
||||
expected = [
|
||||
dict(ip_version=4, scope='global',
|
||||
dynamic=False, cidr='172.16.77.240/24',
|
||||
broadcast='172.16.77.255'),
|
||||
dict(ip_version=6, scope='global',
|
||||
dynamic=True, cidr='2001:470:9:1224:5595:dd51:6ba2:e788/64',
|
||||
broadcast='::'),
|
||||
dict(ip_version=6, scope='global',
|
||||
dynamic=True, cidr='2001:470:9:1224:fd91:272:581e:3a32/64',
|
||||
broadcast='::'),
|
||||
dict(ip_version=6, scope='global',
|
||||
dynamic=True, cidr='2001:470:9:1224:4508:b885:5fb:740b/64',
|
||||
broadcast='::'),
|
||||
dict(ip_version=6, scope='global',
|
||||
dynamic=True, cidr='2001:470:9:1224:dfcc:aaff:feb9:76ce/64',
|
||||
broadcast='::'),
|
||||
dict(ip_version=6, scope='link',
|
||||
dynamic=False, cidr='fe80::dfcc:aaff:feb9:76ce/64',
|
||||
broadcast='::')]
|
||||
|
||||
test_cases = [ADDR_SAMPLE, ADDR_SAMPLE2]
|
||||
|
||||
for test_case in test_cases:
|
||||
self.parent._run = mock.Mock(return_value=test_case)
|
||||
self.assertEqual(self.addr_cmd.list(), expected)
|
||||
self._assert_call([], ('show', 'tap0'))
|
||||
|
||||
def test_list_filtered(self):
|
||||
expected = [
|
||||
dict(ip_version=4, scope='global',
|
||||
dynamic=False, cidr='172.16.77.240/24',
|
||||
broadcast='172.16.77.255')]
|
||||
|
||||
test_cases = [ADDR_SAMPLE, ADDR_SAMPLE2]
|
||||
|
||||
for test_case in test_cases:
|
||||
output = '\n'.join(test_case.split('\n')[0:4])
|
||||
self.parent._run.return_value = output
|
||||
self.assertEqual(self.addr_cmd.list('global',
|
||||
filters=['permanent']), expected)
|
||||
self._assert_call([], ('show', 'tap0', 'permanent', 'scope',
|
||||
'global'))
|
||||
|
||||
|
||||
class TestIpRouteCommand(TestIPCmdBase):
|
||||
def setUp(self):
|
||||
super(TestIpRouteCommand, self).setUp()
|
||||
self.parent.name = 'eth0'
|
||||
self.command = 'route'
|
||||
self.route_cmd = ip_lib.IpRouteCommand(self.parent)
|
||||
|
||||
def test_add_gateway(self):
|
||||
gateway = '192.168.45.100'
|
||||
metric = 100
|
||||
self.route_cmd.add_gateway(gateway, metric)
|
||||
self._assert_sudo([],
|
||||
('replace', 'default', 'via', gateway,
|
||||
'metric', metric,
|
||||
'dev', self.parent.name))
|
||||
|
||||
def test_del_gateway(self):
|
||||
gateway = '192.168.45.100'
|
||||
self.route_cmd.delete_gateway(gateway)
|
||||
self._assert_sudo([],
|
||||
('del', 'default', 'via', gateway,
|
||||
'dev', self.parent.name))
|
||||
|
||||
def test_get_gateway(self):
|
||||
test_cases = [{'sample': GATEWAY_SAMPLE1,
|
||||
'expected': {'gateway': '10.35.19.254',
|
||||
'metric': 100}},
|
||||
{'sample': GATEWAY_SAMPLE2,
|
||||
'expected': {'gateway': '10.35.19.254',
|
||||
'metric': 100}},
|
||||
{'sample': GATEWAY_SAMPLE3,
|
||||
'expected': None},
|
||||
{'sample': GATEWAY_SAMPLE4,
|
||||
'expected': {'gateway': '10.35.19.254'}}]
|
||||
for test_case in test_cases:
|
||||
self.parent._run = mock.Mock(return_value=test_case['sample'])
|
||||
self.assertEqual(self.route_cmd.get_gateway(),
|
||||
test_case['expected'])
|
||||
|
||||
def test_pullup_route(self):
|
||||
# interface is not the first in the list - requires
|
||||
# deleting and creating existing entries
|
||||
output = [DEVICE_ROUTE_SAMPLE, SUBNET_SAMPLE1]
|
||||
|
||||
def pullup_side_effect(self, *args):
|
||||
result = output.pop(0)
|
||||
return result
|
||||
|
||||
self.parent._run = mock.Mock(side_effect=pullup_side_effect)
|
||||
self.route_cmd.pullup_route('tap1d7888a7-10')
|
||||
self._assert_sudo([], ('del', '10.0.0.0/24', 'dev', 'qr-23380d11-d2'))
|
||||
self._assert_sudo([], ('append', '10.0.0.0/24', 'proto', 'kernel',
|
||||
'src', '10.0.0.1', 'dev', 'qr-23380d11-d2'))
|
||||
|
||||
def test_pullup_route_first(self):
|
||||
# interface is first in the list - no changes
|
||||
output = [DEVICE_ROUTE_SAMPLE, SUBNET_SAMPLE2]
|
||||
|
||||
def pullup_side_effect(self, *args):
|
||||
result = output.pop(0)
|
||||
return result
|
||||
|
||||
self.parent._run = mock.Mock(side_effect=pullup_side_effect)
|
||||
self.route_cmd.pullup_route('tap1d7888a7-10')
|
||||
# Check two calls - device get and subnet get
|
||||
self.assertEqual(len(self.parent._run.mock_calls), 2)
|
||||
|
||||
|
||||
class TestIpNetnsCommand(TestIPCmdBase):
|
||||
def setUp(self):
|
||||
super(TestIpNetnsCommand, self).setUp()
|
||||
self.command = 'netns'
|
||||
self.netns_cmd = ip_lib.IpNetnsCommand(self.parent)
|
||||
|
||||
def test_add_namespace(self):
|
||||
ns = self.netns_cmd.add('ns')
|
||||
self._assert_sudo([], ('add', 'ns'), force_root_namespace=True)
|
||||
self.assertEqual(ns.namespace, 'ns')
|
||||
|
||||
def test_delete_namespace(self):
|
||||
with mock.patch('akanda.rug.common.linux.utils.execute') as execute:
|
||||
self.netns_cmd.delete('ns')
|
||||
self._assert_sudo([], ('delete', 'ns'), force_root_namespace=True)
|
||||
|
||||
def test_namespace_exists(self):
|
||||
retval = '\n'.join(NETNS_SAMPLE)
|
||||
self.parent._as_root.return_value = retval
|
||||
self.assertTrue(
|
||||
self.netns_cmd.exists('bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb'))
|
||||
self._assert_sudo('o', ('list',), force_root_namespace=True)
|
||||
|
||||
def test_namespace_doest_not_exist(self):
|
||||
retval = '\n'.join(NETNS_SAMPLE)
|
||||
self.parent._as_root.return_value = retval
|
||||
self.assertFalse(
|
||||
self.netns_cmd.exists('bbbbbbbb-1111-2222-3333-bbbbbbbbbbbb'))
|
||||
self._assert_sudo('o', ('list',), force_root_namespace=True)
|
||||
|
||||
def test_execute(self):
|
||||
self.parent.namespace = 'ns'
|
||||
with mock.patch('akanda.rug.common.linux.utils.execute') as execute:
|
||||
self.netns_cmd.execute(['ip', 'link', 'list'])
|
||||
execute.assert_called_once_with(['ip', 'netns', 'exec', 'ns', 'ip',
|
||||
'link', 'list'],
|
||||
root_helper='sudo',
|
||||
check_exit_code=True)
|
||||
|
||||
def test_execute_env_var_prepend(self):
|
||||
self.parent.namespace = 'ns'
|
||||
with mock.patch('akanda.rug.common.linux.utils.execute') as execute:
|
||||
env = dict(FOO=1, BAR=2)
|
||||
self.netns_cmd.execute(['ip', 'link', 'list'], env)
|
||||
execute.assert_called_once_with(
|
||||
['FOO=1', 'BAR=2', 'ip', 'netns', 'exec', 'ns', 'ip', 'link',
|
||||
'list'],
|
||||
root_helper='sudo', check_exit_code=True)
|
||||
|
||||
|
||||
class TestDeviceExists(unittest.TestCase):
|
||||
def test_device_exists(self):
|
||||
with mock.patch.object(ip_lib.IPDevice, '_execute') as _execute:
|
||||
_execute.return_value = LINK_SAMPLE[1]
|
||||
self.assertTrue(ip_lib.device_exists('eth0'))
|
||||
_execute.assert_called_once_with('o', 'link', ('show', 'eth0'))
|
||||
|
||||
def test_device_does_not_exist(self):
|
||||
with mock.patch.object(ip_lib.IPDevice, '_execute') as _execute:
|
||||
_execute.return_value = ''
|
||||
_execute.side_effect = RuntimeError
|
||||
self.assertFalse(ip_lib.device_exists('eth0'))
|
||||
|
|
@ -0,0 +1,351 @@
|
|||
# vim: tabstop=4 shiftwidth=4 softtabstop=4
|
||||
|
||||
# Copyright 2012, Nicira, Inc.
|
||||
#
|
||||
# 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.
|
||||
# @author: Dan Wendlandt, Nicira, Inc.
|
||||
|
||||
import unittest
|
||||
import uuid
|
||||
import mox
|
||||
|
||||
from akanda.rug.common.linux import ovs_lib, utils
|
||||
|
||||
|
||||
def generate_uuid():
|
||||
return str(uuid.uuid4())
|
||||
|
||||
|
||||
class OVS_Lib_Test(unittest.TestCase):
|
||||
"""
|
||||
A test suite to excercise the OVS libraries shared by Quantum agents.
|
||||
Note: these tests do not actually execute ovs-* utilities, and thus
|
||||
can run on any system. That does, however, limit their scope.
|
||||
"""
|
||||
|
||||
def setUp(self):
|
||||
super(OVS_Lib_Test, self).setUp()
|
||||
self.BR_NAME = "br-int"
|
||||
self.TO = "--timeout=2"
|
||||
|
||||
self.mox = mox.Mox()
|
||||
self.root_helper = 'sudo'
|
||||
self.br = ovs_lib.OVSBridge(self.BR_NAME, self.root_helper)
|
||||
self.mox.StubOutWithMock(utils, "execute")
|
||||
self.addCleanup(self.mox.UnsetStubs)
|
||||
|
||||
def test_vifport(self):
|
||||
"""create and stringify vif port, confirm no exceptions"""
|
||||
self.mox.ReplayAll()
|
||||
|
||||
pname = "vif1.0"
|
||||
ofport = 5
|
||||
vif_id = generate_uuid()
|
||||
mac = "ca:fe:de:ad:be:ef"
|
||||
|
||||
# test __init__
|
||||
port = ovs_lib.VifPort(pname, ofport, vif_id, mac, self.br)
|
||||
self.assertEqual(port.port_name, pname)
|
||||
self.assertEqual(port.ofport, ofport)
|
||||
self.assertEqual(port.vif_id, vif_id)
|
||||
self.assertEqual(port.vif_mac, mac)
|
||||
self.assertEqual(port.switch.br_name, self.BR_NAME)
|
||||
|
||||
# test __str__
|
||||
foo = str(port)
|
||||
|
||||
self.mox.VerifyAll()
|
||||
|
||||
def test_reset_bridge(self):
|
||||
utils.execute(["ovs-vsctl", self.TO, "--",
|
||||
"--if-exists", "del-br", self.BR_NAME],
|
||||
root_helper=self.root_helper)
|
||||
utils.execute(["ovs-vsctl", self.TO, "add-br", self.BR_NAME],
|
||||
root_helper=self.root_helper)
|
||||
self.mox.ReplayAll()
|
||||
|
||||
self.br.reset_bridge()
|
||||
self.mox.VerifyAll()
|
||||
|
||||
def test_delete_port(self):
|
||||
pname = "tap5"
|
||||
utils.execute(["ovs-vsctl", self.TO, "--", "--if-exists",
|
||||
"del-port", self.BR_NAME, pname],
|
||||
root_helper=self.root_helper)
|
||||
|
||||
self.mox.ReplayAll()
|
||||
self.br.delete_port(pname)
|
||||
self.mox.VerifyAll()
|
||||
|
||||
def test_add_flow(self):
|
||||
ofport = "99"
|
||||
vid = 4000
|
||||
lsw_id = 18
|
||||
utils.execute(["ovs-ofctl", "add-flow", self.BR_NAME,
|
||||
"hard_timeout=0,idle_timeout=0,"
|
||||
"priority=2,dl_src=ca:fe:de:ad:be:ef"
|
||||
",actions=strip_vlan,output:0"],
|
||||
root_helper=self.root_helper)
|
||||
utils.execute(["ovs-ofctl", "add-flow", self.BR_NAME,
|
||||
"hard_timeout=0,idle_timeout=0,"
|
||||
"priority=1,actions=normal"],
|
||||
root_helper=self.root_helper)
|
||||
utils.execute(["ovs-ofctl", "add-flow", self.BR_NAME,
|
||||
"hard_timeout=0,idle_timeout=0,"
|
||||
"priority=2,actions=drop"],
|
||||
root_helper=self.root_helper)
|
||||
utils.execute(["ovs-ofctl", "add-flow", self.BR_NAME,
|
||||
"hard_timeout=0,idle_timeout=0,"
|
||||
"priority=2,in_port=%s,actions=drop" % ofport],
|
||||
root_helper=self.root_helper)
|
||||
utils.execute(["ovs-ofctl", "add-flow", self.BR_NAME,
|
||||
"hard_timeout=0,idle_timeout=0,"
|
||||
"priority=4,in_port=%s,dl_vlan=%s,"
|
||||
"actions=strip_vlan,set_tunnel:%s,normal"
|
||||
% (ofport, vid, lsw_id)],
|
||||
root_helper=self.root_helper)
|
||||
utils.execute(["ovs-ofctl", "add-flow", self.BR_NAME,
|
||||
"hard_timeout=0,idle_timeout=0,"
|
||||
"priority=3,tun_id=%s,actions="
|
||||
"mod_vlan_vid:%s,output:%s"
|
||||
% (lsw_id, vid, ofport)], root_helper=self.root_helper)
|
||||
self.mox.ReplayAll()
|
||||
|
||||
self.br.add_flow(priority=2, dl_src="ca:fe:de:ad:be:ef",
|
||||
actions="strip_vlan,output:0")
|
||||
self.br.add_flow(priority=1, actions="normal")
|
||||
self.br.add_flow(priority=2, actions="drop")
|
||||
self.br.add_flow(priority=2, in_port=ofport, actions="drop")
|
||||
|
||||
self.br.add_flow(priority=4, in_port=ofport, dl_vlan=vid,
|
||||
actions="strip_vlan,set_tunnel:%s,normal" %
|
||||
(lsw_id))
|
||||
self.br.add_flow(priority=3, tun_id=lsw_id,
|
||||
actions="mod_vlan_vid:%s,output:%s" %
|
||||
(vid, ofport))
|
||||
self.mox.VerifyAll()
|
||||
|
||||
def test_get_port_ofport(self):
|
||||
pname = "tap99"
|
||||
ofport = "6"
|
||||
utils.execute(["ovs-vsctl", self.TO, "get",
|
||||
"Interface", pname, "ofport"],
|
||||
root_helper=self.root_helper).AndReturn(ofport)
|
||||
self.mox.ReplayAll()
|
||||
|
||||
self.assertEqual(self.br.get_port_ofport(pname), ofport)
|
||||
self.mox.VerifyAll()
|
||||
|
||||
def test_get_datapath_id(self):
|
||||
datapath_id = '"0000b67f4fbcc149"'
|
||||
utils.execute(["ovs-vsctl", self.TO, "get",
|
||||
"Bridge", self.BR_NAME, "datapath_id"],
|
||||
root_helper=self.root_helper).AndReturn(datapath_id)
|
||||
self.mox.ReplayAll()
|
||||
|
||||
self.assertEqual(self.br.get_datapath_id(), datapath_id.strip('"'))
|
||||
self.mox.VerifyAll()
|
||||
|
||||
def test_count_flows(self):
|
||||
utils.execute(["ovs-ofctl", "dump-flows", self.BR_NAME],
|
||||
root_helper=self.root_helper).AndReturn('ignore'
|
||||
'\nflow-1\n')
|
||||
self.mox.ReplayAll()
|
||||
|
||||
# counts the number of flows as total lines of output - 2
|
||||
self.assertEqual(self.br.count_flows(), 1)
|
||||
self.mox.VerifyAll()
|
||||
|
||||
def test_delete_flow(self):
|
||||
ofport = "5"
|
||||
lsw_id = 40
|
||||
vid = 39
|
||||
utils.execute(["ovs-ofctl", "del-flows", self.BR_NAME,
|
||||
"in_port=" + ofport], root_helper=self.root_helper)
|
||||
utils.execute(["ovs-ofctl", "del-flows", self.BR_NAME,
|
||||
"tun_id=%s" % lsw_id], root_helper=self.root_helper)
|
||||
utils.execute(["ovs-ofctl", "del-flows", self.BR_NAME,
|
||||
"dl_vlan=%s" % vid], root_helper=self.root_helper)
|
||||
self.mox.ReplayAll()
|
||||
|
||||
self.br.delete_flows(in_port=ofport)
|
||||
self.br.delete_flows(tun_id=lsw_id)
|
||||
self.br.delete_flows(dl_vlan=vid)
|
||||
self.mox.VerifyAll()
|
||||
|
||||
def test_add_tunnel_port(self):
|
||||
pname = "tap99"
|
||||
ip = "9.9.9.9"
|
||||
ofport = "6"
|
||||
|
||||
utils.execute(["ovs-vsctl", self.TO, "add-port",
|
||||
self.BR_NAME, pname], root_helper=self.root_helper)
|
||||
utils.execute(["ovs-vsctl", self.TO, "set", "Interface",
|
||||
pname, "type=gre"], root_helper=self.root_helper)
|
||||
utils.execute(["ovs-vsctl", self.TO, "set", "Interface",
|
||||
pname, "options:remote_ip=" + ip],
|
||||
root_helper=self.root_helper)
|
||||
utils.execute(["ovs-vsctl", self.TO, "set", "Interface",
|
||||
pname, "options:in_key=flow"],
|
||||
root_helper=self.root_helper)
|
||||
utils.execute(["ovs-vsctl", self.TO, "set", "Interface",
|
||||
pname, "options:out_key=flow"],
|
||||
root_helper=self.root_helper)
|
||||
utils.execute(["ovs-vsctl", self.TO, "get",
|
||||
"Interface", pname, "ofport"],
|
||||
root_helper=self.root_helper).AndReturn(ofport)
|
||||
self.mox.ReplayAll()
|
||||
|
||||
self.assertEqual(self.br.add_tunnel_port(pname, ip), ofport)
|
||||
self.mox.VerifyAll()
|
||||
|
||||
def test_add_patch_port(self):
|
||||
pname = "tap99"
|
||||
peer = "bar10"
|
||||
ofport = "6"
|
||||
|
||||
utils.execute(["ovs-vsctl", self.TO, "add-port",
|
||||
self.BR_NAME, pname], root_helper=self.root_helper)
|
||||
utils.execute(["ovs-vsctl", self.TO, "set", "Interface",
|
||||
pname, "type=patch"], root_helper=self.root_helper)
|
||||
utils.execute(["ovs-vsctl", self.TO, "set",
|
||||
"Interface", pname, "options:peer=" + peer],
|
||||
root_helper=self.root_helper)
|
||||
utils.execute(["ovs-vsctl", self.TO, "get",
|
||||
"Interface", pname, "ofport"],
|
||||
root_helper=self.root_helper).AndReturn(ofport)
|
||||
self.mox.ReplayAll()
|
||||
|
||||
self.assertEqual(self.br.add_patch_port(pname, peer), ofport)
|
||||
self.mox.VerifyAll()
|
||||
|
||||
def _test_get_vif_ports(self, is_xen=False):
|
||||
pname = "tap99"
|
||||
ofport = "6"
|
||||
vif_id = generate_uuid()
|
||||
mac = "ca:fe:de:ad:be:ef"
|
||||
|
||||
utils.execute(["ovs-vsctl", self.TO, "list-ports", self.BR_NAME],
|
||||
root_helper=self.root_helper).AndReturn("%s\n" % pname)
|
||||
|
||||
if is_xen:
|
||||
external_ids = ('{xs-vif-uuid="%s", attached-mac="%s"}'
|
||||
% (vif_id, mac))
|
||||
else:
|
||||
external_ids = ('{iface-id="%s", attached-mac="%s"}'
|
||||
% (vif_id, mac))
|
||||
|
||||
utils.execute(["ovs-vsctl", self.TO, "get",
|
||||
"Interface", pname, "external_ids"],
|
||||
root_helper=self.root_helper).AndReturn(external_ids)
|
||||
utils.execute(["ovs-vsctl", self.TO, "get",
|
||||
"Interface", pname, "ofport"],
|
||||
root_helper=self.root_helper).AndReturn(ofport)
|
||||
if is_xen:
|
||||
utils.execute(["xe", "vif-param-get", "param-name=other-config",
|
||||
"param-key=nicira-iface-id", "uuid=" + vif_id],
|
||||
root_helper=self.root_helper).AndReturn(vif_id)
|
||||
self.mox.ReplayAll()
|
||||
|
||||
ports = self.br.get_vif_ports()
|
||||
self.assertEqual(1, len(ports))
|
||||
self.assertEqual(ports[0].port_name, pname)
|
||||
self.assertEqual(ports[0].ofport, ofport)
|
||||
self.assertEqual(ports[0].vif_id, vif_id)
|
||||
self.assertEqual(ports[0].vif_mac, mac)
|
||||
self.assertEqual(ports[0].switch.br_name, self.BR_NAME)
|
||||
self.mox.VerifyAll()
|
||||
|
||||
def test_get_vif_ports_nonxen(self):
|
||||
self._test_get_vif_ports(False)
|
||||
|
||||
def test_get_vif_ports_xen(self):
|
||||
self._test_get_vif_ports(True)
|
||||
|
||||
def test_clear_db_attribute(self):
|
||||
pname = "tap77"
|
||||
utils.execute(["ovs-vsctl", self.TO, "clear", "Port",
|
||||
pname, "tag"], root_helper=self.root_helper)
|
||||
self.mox.ReplayAll()
|
||||
self.br.clear_db_attribute("Port", pname, "tag")
|
||||
self.mox.VerifyAll()
|
||||
|
||||
def test_port_id_regex(self):
|
||||
result = ('external_ids : {attached-mac="fa:16:3e:23:5b:f2",'
|
||||
' iface-id="5c1321a7-c73f-4a77-95e6-9f86402e5c8f",'
|
||||
' iface-status=active}\nname :'
|
||||
' "dhc5c1321a7-c7"\nofport : 2\n')
|
||||
match = self.br.re_id.search(result)
|
||||
vif_mac = match.group('vif_mac')
|
||||
vif_id = match.group('vif_id')
|
||||
port_name = match.group('port_name')
|
||||
ofport = int(match.group('ofport'))
|
||||
self.assertEqual(vif_mac, 'fa:16:3e:23:5b:f2')
|
||||
self.assertEqual(vif_id, '5c1321a7-c73f-4a77-95e6-9f86402e5c8f')
|
||||
self.assertEqual(port_name, 'dhc5c1321a7-c7')
|
||||
self.assertEqual(ofport, 2)
|
||||
|
||||
def test_iface_to_br(self):
|
||||
iface = 'tap0'
|
||||
br = 'br-int'
|
||||
root_helper = 'sudo'
|
||||
utils.execute(["ovs-vsctl", self.TO, "iface-to-br", iface],
|
||||
root_helper=root_helper).AndReturn('br-int')
|
||||
|
||||
self.mox.ReplayAll()
|
||||
self.assertEqual(ovs_lib.get_bridge_for_iface(root_helper, iface), br)
|
||||
self.mox.VerifyAll()
|
||||
|
||||
def test_iface_to_br_handles_ovs_vsctl_exception(self):
|
||||
iface = 'tap0'
|
||||
root_helper = 'sudo'
|
||||
utils.execute(["ovs-vsctl", self.TO, "iface-to-br", iface],
|
||||
root_helper=root_helper).AndRaise(Exception)
|
||||
|
||||
self.mox.ReplayAll()
|
||||
self.assertIsNone(ovs_lib.get_bridge_for_iface(root_helper, iface))
|
||||
self.mox.VerifyAll()
|
||||
|
||||
def test_delete_all_ports(self):
|
||||
self.mox.StubOutWithMock(self.br, 'get_port_name_list')
|
||||
self.br.get_port_name_list().AndReturn(['port1'])
|
||||
self.mox.StubOutWithMock(self.br, 'delete_port')
|
||||
self.br.delete_port('port1')
|
||||
self.mox.ReplayAll()
|
||||
self.br.delete_ports(all_ports=True)
|
||||
self.mox.VerifyAll()
|
||||
|
||||
def test_delete_quantum_ports(self):
|
||||
port1 = ovs_lib.VifPort('tap1234', 1, generate_uuid(),
|
||||
'ca:fe:de:ad:be:ef', 'br')
|
||||
port2 = ovs_lib.VifPort('tap5678', 2, generate_uuid(),
|
||||
'ca:ee:de:ad:be:ef', 'br')
|
||||
ports = [port1, port2]
|
||||
self.mox.StubOutWithMock(self.br, 'get_vif_ports')
|
||||
self.br.get_vif_ports().AndReturn(ports)
|
||||
self.mox.StubOutWithMock(self.br, 'delete_port')
|
||||
self.br.delete_port('tap1234')
|
||||
self.br.delete_port('tap5678')
|
||||
self.mox.ReplayAll()
|
||||
self.br.delete_ports(all_ports=False)
|
||||
self.mox.VerifyAll()
|
||||
|
||||
def test_get_bridges(self):
|
||||
bridges = ['br-int', 'br-ex']
|
||||
root_helper = 'sudo'
|
||||
utils.execute(["ovs-vsctl", self.TO, "list-br"],
|
||||
root_helper=root_helper).AndReturn('br-int\nbr-ex\n')
|
||||
|
||||
self.mox.ReplayAll()
|
||||
self.assertEqual(ovs_lib.get_bridges(root_helper), bridges)
|
||||
self.mox.VerifyAll()
|
||||
|
|
@ -8,3 +8,6 @@ eventlet==0.12.1
|
|||
iso8601==0.1.4
|
||||
python-novaclient
|
||||
WebOb==1.2.3
|
||||
mox==0.5.3
|
||||
testtools
|
||||
fixtures
|
||||
|
|
|
|||
Loading…
Reference in New Issue