From 0892382a3c7100a5dc23b3d347760de3142be5ca Mon Sep 17 00:00:00 2001 From: Rosario Di Somma Date: Mon, 29 Apr 2013 22:22:32 +0200 Subject: [PATCH] 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 Change-Id: I7136c9bd367532e12e8c58fb4469f100d10f4ed9 --- test/unit/common/test_agent_linux_utils.py | 92 +++ test/unit/common/test_linux_interface.py | 251 ++++++++ test/unit/common/test_linux_ip_lib.py | 680 +++++++++++++++++++++ test/unit/openvswitch/__init__.py | 0 test/unit/openvswitch/test_osv_lib.py | 351 +++++++++++ test_requirements.txt | 3 + 6 files changed, 1377 insertions(+) create mode 100644 test/unit/common/test_agent_linux_utils.py create mode 100644 test/unit/common/test_linux_interface.py create mode 100644 test/unit/common/test_linux_ip_lib.py create mode 100644 test/unit/openvswitch/__init__.py create mode 100644 test/unit/openvswitch/test_osv_lib.py diff --git a/test/unit/common/test_agent_linux_utils.py b/test/unit/common/test_agent_linux_utils.py new file mode 100644 index 00000000..dd485ee2 --- /dev/null +++ b/test/unit/common/test_agent_linux_utils.py @@ -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') diff --git a/test/unit/common/test_linux_interface.py b/test/unit/common/test_linux_interface.py new file mode 100644 index 00000000..111a908d --- /dev/null +++ b/test/unit/common/test_linux_interface.py @@ -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()]) diff --git a/test/unit/common/test_linux_ip_lib.py b/test/unit/common/test_linux_ip_lib.py new file mode 100644 index 00000000..b8d9b191 --- /dev/null +++ b/test/unit/common/test_linux_ip_lib.py @@ -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: mtu 16436 qdisc noqueue state UNKNOWN \\' + 'link/loopback 00:00:00:00:00:00 brd 00:00:00:00:00:00', + '2: eth0: 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: 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: 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: 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: 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')) diff --git a/test/unit/openvswitch/__init__.py b/test/unit/openvswitch/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/test/unit/openvswitch/test_osv_lib.py b/test/unit/openvswitch/test_osv_lib.py new file mode 100644 index 00000000..b8c09b9b --- /dev/null +++ b/test/unit/openvswitch/test_osv_lib.py @@ -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() diff --git a/test_requirements.txt b/test_requirements.txt index c03c3378..d0b557cf 100644 --- a/test_requirements.txt +++ b/test_requirements.txt @@ -8,3 +8,6 @@ eventlet==0.12.1 iso8601==0.1.4 python-novaclient WebOb==1.2.3 +mox==0.5.3 +testtools +fixtures