neutron/neutron/tests/unit/test_linux_interface.py
Ihar Hrachyshka 548ab45a50 hacking: enable H238 (old style class declaration, use new style)
The rule was added in hacking 0.10 and is useful for migration to Python 3.

Change-Id: I8b45413cf34e5a9db8074f9029410e3b22a92640
2015-01-12 23:46:11 +01:00

612 lines
24 KiB
Python

# 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 mock
from neutron.agent.common import config
from neutron.agent.linux import dhcp
from neutron.agent.linux import interface
from neutron.agent.linux import ip_lib
from neutron.agent.linux import ovs_lib
from neutron.agent.linux import utils
from neutron.extensions import flavor
from neutron.openstack.common import uuidutils
from neutron.tests import base
class BaseChild(interface.LinuxInterfaceDriver):
def plug(*args):
pass
def unplug(*args):
pass
class FakeNetwork(object):
id = '12345678-1234-5678-90ab-ba0987654321'
class FakeSubnet(object):
cidr = '192.168.1.1/24'
class FakeAllocation(object):
subnet = FakeSubnet()
ip_address = '192.168.1.2'
ip_version = 4
class FakePort(object):
id = 'abcdef01-1234-5678-90ab-ba0987654321'
fixed_ips = [FakeAllocation]
device_id = 'cccccccc-cccc-cccc-cccc-cccccccccccc'
network = FakeNetwork()
network_id = network.id
class TestBase(base.BaseTestCase):
def setUp(self):
super(TestBase, self).setUp()
self.conf = config.setup_conf()
self.conf.register_opts(interface.OPTS)
config.register_root_helper(self.conf)
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()
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)
self.ip_dev().route.list_onlink_routes.return_value = []
bc = BaseChild(self.conf)
ns = '12345678-1234-5678-90ab-ba0987654321'
bc.init_l3('tap0', ['192.168.1.2/24'], namespace=ns,
extra_subnets=[{'cidr': '172.20.0.0/24'}])
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'),
mock.call().route.list_onlink_routes(),
mock.call().route.add_onlink_route('172.20.0.0/24')])
def test_l3_init_delete_onlink_routes(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)
self.ip_dev().route.list_onlink_routes.return_value = ['172.20.0.0/24']
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().route.list_onlink_routes(),
mock.call().route.delete_onlink_route('172.20.0.0/24')])
def test_l3_init_with_preserve(self):
addresses = [dict(ip_version=4, scope='global',
dynamic=False, cidr='192.168.1.3/32')]
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,
preserve_ips=['192.168.1.3/32'])
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')])
self.assertFalse(self.ip_dev().addr.delete.called)
def test_l3_init_with_ipv6(self):
addresses = [dict(ip_version=6,
scope='global',
dynamic=False,
cidr='2001:db8:a::123/64')]
self.ip_dev().addr.list = mock.Mock(return_value=addresses)
bc = BaseChild(self.conf)
ns = '12345678-1234-5678-90ab-ba0987654321'
bc.init_l3('tap0', ['2001:db8:a::124/64'], 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(6, '2001:db8:a::124/64',
'2001:db8:a:0:ffff:ffff:ffff:ffff'),
mock.call().addr.delete(6, '2001:db8:a::123/64')])
def test_l3_init_with_duplicated_ipv6(self):
addresses = [dict(ip_version=6,
scope='global',
dynamic=False,
cidr='2001:db8:a::123/64')]
self.ip_dev().addr.list = mock.Mock(return_value=addresses)
bc = BaseChild(self.conf)
ns = '12345678-1234-5678-90ab-ba0987654321'
bc.init_l3('tap0', ['2001:db8:a::123/64'], namespace=ns)
self.assertFalse(self.ip_dev().addr.add.called)
def test_l3_init_with_duplicated_ipv6_uncompact(self):
addresses = [dict(ip_version=6,
scope='global',
dynamic=False,
cidr='2001:db8:a::123/64')]
self.ip_dev().addr.list = mock.Mock(return_value=addresses)
bc = BaseChild(self.conf)
ns = '12345678-1234-5678-90ab-ba0987654321'
bc.init_l3('tap0',
['2001:db8:a:0000:0000:0000:0000:0123/64'],
namespace=ns)
self.assertFalse(self.ip_dev().addr.add.called)
class TestOVSInterfaceDriver(TestBase):
def test_get_device_name(self):
br = interface.OVSInterfaceDriver(self.conf)
device_name = br.get_device_name(FakePort())
self.assertEqual('tapabcdef01-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_alt_bridge(self):
self._test_plug(bridge='br-foo')
def test_plug_configured_bridge(self):
br = 'br-v'
self.conf.set_override('ovs_use_veth', False)
self.conf.set_override('ovs_integration_bridge', br)
self.assertEqual(self.conf.ovs_integration_bridge, br)
def device_exists(dev, root_helper=None, namespace=None):
return dev == br
ovs = interface.OVSInterfaceDriver(self.conf)
with mock.patch.object(ovs, '_ovs_add_port') as add_port:
self.device_exists.side_effect = device_exists
ovs.plug('01234567-1234-1234-99',
'port-1234',
'tap0',
'aa:bb:cc:dd:ee:ff',
bridge=None,
namespace=None)
add_port.assert_called_once_with('br-v',
'tap0',
'port-1234',
'aa:bb:cc:dd:ee:ff',
internal=True)
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
with mock.patch.object(ovs_lib.OVSBridge, 'replace_port') as replace:
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)
replace.assert_called_once_with(
'tap0',
('type', 'internal'),
('external-ids:iface-id', 'port-1234'),
('external-ids:iface-status', 'active'),
('external-ids:attached-mac', 'aa:bb:cc:dd:ee:ff'))
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_mtu_int(self):
self.assertIsNone(self.conf.network_device_mtu)
self.conf.set_override('network_device_mtu', 9000)
self.assertEqual(self.conf.network_device_mtu, 9000)
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('neutron.agent.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 TestOVSInterfaceDriverWithVeth(TestOVSInterfaceDriver):
def setUp(self):
super(TestOVSInterfaceDriverWithVeth, self).setUp()
self.conf.set_override('ovs_use_veth', True)
def test_get_device_name(self):
br = interface.OVSInterfaceDriver(self.conf)
device_name = br.get_device_name(FakePort())
self.assertEqual('ns-abcdef01-12', device_name)
def test_plug_with_prefix(self):
self._test_plug(devname='qr-0', prefix='qr-')
def _test_plug(self, devname=None, bridge=None, namespace=None,
prefix=None, mtu=None):
if not devname:
devname = 'ns-0'
if not bridge:
bridge = 'br-int'
def device_exists(dev, root_helper=None, namespace=None):
return dev == bridge
ovs = interface.OVSInterfaceDriver(self.conf)
self.device_exists.side_effect = device_exists
root_dev = mock.Mock()
ns_dev = mock.Mock()
self.ip().add_veth = mock.Mock(return_value=(root_dev, ns_dev))
expected = [mock.call('sudo'),
mock.call().add_veth('tap0', devname,
namespace2=namespace)]
with mock.patch.object(ovs_lib.OVSBridge, 'replace_port') as replace:
ovs.plug('01234567-1234-1234-99',
'port-1234',
devname,
'aa:bb:cc:dd:ee:ff',
bridge=bridge,
namespace=namespace,
prefix=prefix)
replace.assert_called_once_with(
'tap0',
('external-ids:iface-id', 'port-1234'),
('external-ids:iface-status', 'active'),
('external-ids:attached-mac', 'aa:bb:cc:dd:ee:ff'))
ns_dev.assert_has_calls(
[mock.call.link.set_address('aa:bb:cc:dd:ee:ff')])
if mtu:
ns_dev.assert_has_calls([mock.call.link.set_mtu(mtu)])
root_dev.assert_has_calls([mock.call.link.set_mtu(mtu)])
self.ip.assert_has_calls(expected)
root_dev.assert_has_calls([mock.call.link.set_up()])
ns_dev.assert_has_calls([mock.call.link.set_up()])
def test_plug_mtu(self):
self.conf.set_override('network_device_mtu', 9000)
self._test_plug(mtu=9000)
def test_unplug(self, bridge=None):
if not bridge:
bridge = 'br-int'
with mock.patch('neutron.agent.linux.ovs_lib.OVSBridge') as ovs_br:
ovs = interface.OVSInterfaceDriver(self.conf)
ovs.unplug('ns-0', bridge=bridge)
ovs_br.assert_has_calls([mock.call(bridge, 'sudo'),
mock.call().delete_port('tap0')])
self.ip_dev.assert_has_calls([mock.call('ns-0', 'sudo', None),
mock.call().link.delete()])
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, mtu=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)
mac_address = 'aa:bb:cc:dd:ee:ff'
br.plug('01234567-1234-1234-99',
'port-1234',
'ns-0',
mac_address,
namespace=namespace)
ip_calls = [mock.call('sudo'),
mock.call().add_veth('tap0', 'ns-0', namespace2=namespace)]
ns_veth.assert_has_calls([mock.call.link.set_address(mac_address)])
if mtu:
ns_veth.assert_has_calls([mock.call.link.set_mtu(mtu)])
root_veth.assert_has_calls([mock.call.link.set_mtu(mtu)])
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('neutron.agent.linux.interface.LOG.info') as log:
br = interface.BridgeInterfaceDriver(self.conf)
br.plug('01234567-1234-1234-99',
'port-1234',
'tap0',
'aa:bb:cc:dd:ee:ff')
self.assertFalse(self.ip_dev.called)
self.assertEqual(log.call_count, 1)
def test_plug_mtu(self):
self.device_exists.return_value = False
self.conf.set_override('network_device_mtu', 9000)
self._test_plug(mtu=9000)
def test_unplug_no_device(self):
self.device_exists.return_value = False
self.ip_dev().link.delete.side_effect = RuntimeError
with mock.patch('neutron.agent.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('neutron.agent.linux.interface.LOG.debug') as log:
br = interface.BridgeInterfaceDriver(self.conf)
br.unplug('tap0')
self.assertEqual(log.call_count, 1)
self.ip_dev.assert_has_calls([mock.call('tap0', 'sudo', None),
mock.call().link.delete()])
class TestMetaInterfaceDriver(TestBase):
def setUp(self):
super(TestMetaInterfaceDriver, self).setUp()
config.register_interface_driver_opts_helper(self.conf)
self.conf.register_opts(dhcp.OPTS)
self.client_cls_p = mock.patch('neutronclient.v2_0.client.Client')
client_cls = self.client_cls_p.start()
self.client_inst = mock.Mock()
client_cls.return_value = self.client_inst
fake_network = {'network': {flavor.FLAVOR_NETWORK: 'fake1'}}
fake_port = {'ports':
[{'mac_address':
'aa:bb:cc:dd:ee:ffa', 'network_id': 'test'}]}
self.client_inst.list_ports.return_value = fake_port
self.client_inst.show_network.return_value = fake_network
self.conf.set_override('auth_url', 'http://localhost:35357/v2.0')
self.conf.set_override('auth_region', 'RegionOne')
self.conf.set_override('admin_user', 'neutron')
self.conf.set_override('admin_password', 'password')
self.conf.set_override('admin_tenant_name', 'service')
self.conf.set_override(
'meta_flavor_driver_mappings',
'fake1:neutron.agent.linux.interface.OVSInterfaceDriver,'
'fake2:neutron.agent.linux.interface.BridgeInterfaceDriver')
self.conf.set_override('endpoint_type', 'internalURL')
def test_get_driver_by_network_id(self):
meta_interface = interface.MetaInterfaceDriver(self.conf)
driver = meta_interface._get_driver_by_network_id('test')
self.assertIsInstance(driver, interface.OVSInterfaceDriver)
def test_set_device_plugin_tag(self):
meta_interface = interface.MetaInterfaceDriver(self.conf)
driver = meta_interface._get_driver_by_network_id('test')
meta_interface._set_device_plugin_tag(driver,
'tap0',
namespace=None)
expected = [mock.call('tap0', 'sudo', None),
mock.call().link.set_alias('fake1')]
self.ip_dev.assert_has_calls(expected)
namespace = '01234567-1234-1234-99'
meta_interface._set_device_plugin_tag(driver,
'tap1',
namespace=namespace)
expected = [mock.call('tap1', 'sudo', '01234567-1234-1234-99'),
mock.call().link.set_alias('fake1')]
self.ip_dev.assert_has_calls(expected)
def test_get_device_plugin_tag(self):
meta_interface = interface.MetaInterfaceDriver(self.conf)
self.ip_dev().link.alias = 'fake1'
plugin_tag0 = meta_interface._get_device_plugin_tag('tap0',
namespace=None)
expected = [mock.call('tap0', 'sudo', None)]
self.ip_dev.assert_has_calls(expected)
self.assertEqual('fake1', plugin_tag0)
namespace = '01234567-1234-1234-99'
expected = [mock.call('tap1', 'sudo', '01234567-1234-1234-99')]
plugin_tag1 = meta_interface._get_device_plugin_tag(
'tap1',
namespace=namespace)
self.ip_dev.assert_has_calls(expected)
self.assertEqual('fake1', plugin_tag1)
class TestIVSInterfaceDriver(TestBase):
def setUp(self):
super(TestIVSInterfaceDriver, self).setUp()
def test_get_device_name(self):
br = interface.IVSInterfaceDriver(self.conf)
device_name = br.get_device_name(FakePort())
self.assertEqual('ns-abcdef01-12', device_name)
def test_plug_with_prefix(self):
self._test_plug(devname='qr-0', prefix='qr-')
def _test_plug(self, devname=None, namespace=None,
prefix=None, mtu=None):
if not devname:
devname = 'ns-0'
def device_exists(dev, root_helper=None, namespace=None):
return dev == 'indigo'
ivs = interface.IVSInterfaceDriver(self.conf)
self.device_exists.side_effect = device_exists
root_dev = mock.Mock()
_ns_dev = mock.Mock()
ns_dev = mock.Mock()
self.ip().add_veth = mock.Mock(return_value=(root_dev, _ns_dev))
self.ip().device = mock.Mock(return_value=(ns_dev))
expected = [mock.call('sudo'), mock.call().add_veth('tap0', devname),
mock.call().device(devname)]
ivsctl_cmd = ['ivs-ctl', 'add-port', 'tap0']
with mock.patch.object(utils, 'execute') as execute:
ivs.plug('01234567-1234-1234-99',
'port-1234',
devname,
'aa:bb:cc:dd:ee:ff',
namespace=namespace,
prefix=prefix)
execute.assert_called_once_with(ivsctl_cmd, 'sudo')
ns_dev.assert_has_calls(
[mock.call.link.set_address('aa:bb:cc:dd:ee:ff')])
if mtu:
ns_dev.assert_has_calls([mock.call.link.set_mtu(mtu)])
root_dev.assert_has_calls([mock.call.link.set_mtu(mtu)])
if namespace:
expected.extend(
[mock.call().ensure_namespace(namespace),
mock.call().ensure_namespace().add_device_to_namespace(
mock.ANY)])
self.ip.assert_has_calls(expected)
root_dev.assert_has_calls([mock.call.link.set_up()])
ns_dev.assert_has_calls([mock.call.link.set_up()])
def test_plug_mtu(self):
self.conf.set_override('network_device_mtu', 9000)
self._test_plug(mtu=9000)
def test_plug_namespace(self):
self._test_plug(namespace='mynamespace')
def test_unplug(self):
ivs = interface.IVSInterfaceDriver(self.conf)
ivsctl_cmd = ['ivs-ctl', 'del-port', 'tap0']
with mock.patch.object(utils, 'execute') as execute:
ivs.unplug('ns-0')
execute.assert_called_once_with(ivsctl_cmd, 'sudo')
self.ip_dev.assert_has_calls([mock.call('ns-0', 'sudo', None),
mock.call().link.delete()])
class TestMidonetInterfaceDriver(TestBase):
def setUp(self):
self.conf = config.setup_conf()
self.conf.register_opts(interface.OPTS)
config.register_root_helper(self.conf)
self.driver = interface.MidonetInterfaceDriver(self.conf)
self.network_id = uuidutils.generate_uuid()
self.port_id = uuidutils.generate_uuid()
self.device_name = "tap0"
self.mac_address = "aa:bb:cc:dd:ee:ff"
self.bridge = "br-test"
self.namespace = "ns-test"
super(TestMidonetInterfaceDriver, self).setUp()
def test_plug(self):
cmd = ['mm-ctl', '--bind-port', self.port_id, 'tap0']
self.device_exists.return_value = False
root_dev = mock.Mock()
ns_dev = mock.Mock()
self.ip().add_veth = mock.Mock(return_value=(root_dev, ns_dev))
with mock.patch.object(utils, 'execute') as execute:
self.driver.plug(
self.network_id, self.port_id,
self.device_name, self.mac_address,
self.bridge, self.namespace)
execute.assert_called_once_with(cmd, 'sudo')
expected = [mock.call(), mock.call('sudo'),
mock.call().add_veth(self.device_name,
self.device_name,
namespace2=self.namespace),
mock.call().ensure_namespace(self.namespace),
mock.call().ensure_namespace().add_device_to_namespace(
mock.ANY)]
ns_dev.assert_has_calls(
[mock.call.link.set_address(self.mac_address)])
root_dev.assert_has_calls([mock.call.link.set_up()])
ns_dev.assert_has_calls([mock.call.link.set_up()])
self.ip.assert_has_calls(expected, True)
def test_unplug(self):
self.driver.unplug(self.device_name, self.bridge, self.namespace)
self.ip_dev.assert_has_calls([
mock.call(self.device_name, self.driver.root_helper,
self.namespace),
mock.call().link.delete()])
self.ip.assert_has_calls(mock.call().garbage_collect_namespace())