OpenStack Networking (Neutron)
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 

1881 lines
74 KiB

# 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 errno
import socket
import mock
import netaddr
from neutron_lib import constants
from neutron_lib import exceptions
import pyroute2
from pyroute2.netlink.rtnl import ifinfmsg
from pyroute2.netlink.rtnl import ndmsg
from pyroute2 import NetlinkError
import testtools
from neutron.agent.common import utils # noqa
from neutron.agent.linux import ip_lib
from neutron.common import exceptions as n_exc
from neutron import privileged
from neutron.privileged.agent.linux import ip_lib as priv_lib
from neutron.tests import base
NETNS_SAMPLE = [
'12345678-1234-5678-abcd-1234567890ab',
'bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb',
'cccccccc-cccc-cccc-cccc-cccccccccccc']
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 fe80::3023:39ff:febc:22ae/64 scope link tentative
valid_lft forever preferred_lft forever
inet6 fe80::3023:39ff:febc:22af/64 scope link tentative dadfailed
valid_lft forever preferred_lft forever
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 fe80::3023:39ff:febc:22ae/64 scope link tentative
valid_lft forever preferred_lft forever
inet6 fe80::3023:39ff:febc:22af/64 scope link tentative dadfailed
valid_lft forever preferred_lft forever
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_SAMPLE3 = ("""
2: eth0@NONE: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 qdisc mq state UP
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
""")
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
""")
GATEWAY_SAMPLE5 = ("""
default via 192.168.99.1 proto static
""")
GATEWAY_SAMPLE6 = ("""
default via 192.168.99.1 proto static metric 100
""")
GATEWAY_SAMPLE7 = ("""
default dev qg-31cd36 metric 1
""")
IPv6_GATEWAY_SAMPLE1 = ("""
default via 2001:470:9:1224:4508:b885:5fb:740b metric 100
2001:db8::/64 proto kernel scope link src 2001:470:9:1224:dfcc:aaff:feb9:76ce
""")
IPv6_GATEWAY_SAMPLE2 = ("""
default via 2001:470:9:1224:4508:b885:5fb:740b metric 100
""")
IPv6_GATEWAY_SAMPLE3 = ("""
2001:db8::/64 proto kernel scope link src 2001:470:9:1224:dfcc:aaff:feb9:76ce
""")
IPv6_GATEWAY_SAMPLE4 = ("""
default via fe80::dfcc:aaff:feb9:76ce
""")
IPv6_GATEWAY_SAMPLE5 = ("""
default via 2001:470:9:1224:4508:b885:5fb:740b metric 1024
""")
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")
RULE_V4_SAMPLE = ("""
0: from all lookup local
32766: from all lookup main
32767: from all lookup default
101: from 192.168.45.100 lookup 2
""")
RULE_V6_SAMPLE = ("""
0: from all lookup local
32766: from all lookup main
32767: from all lookup default
201: from 2001:db8::1 lookup 3
""")
class TestSubProcessBase(base.BaseTestCase):
def setUp(self):
super(TestSubProcessBase, self).setUp()
self.execute_p = mock.patch('neutron.agent.common.utils.execute')
self.execute = self.execute_p.start()
def test_execute_wrapper(self):
base = ip_lib.SubProcessBase()
base._execute(['o'], 'link', ('list',), run_as_root=True)
self.execute.assert_called_once_with(['ip', '-o', 'link', 'list'],
run_as_root=True,
log_fail_as_error=True)
def test_execute_wrapper_int_options(self):
base = ip_lib.SubProcessBase()
base._execute([4], 'link', ('list',))
self.execute.assert_called_once_with(['ip', '-4', 'link', 'list'],
run_as_root=False,
log_fail_as_error=True)
def test_execute_wrapper_no_options(self):
base = ip_lib.SubProcessBase()
base._execute([], 'link', ('list',))
self.execute.assert_called_once_with(['ip', 'link', 'list'],
run_as_root=False,
log_fail_as_error=True)
def test_run_no_namespace(self):
base = ip_lib.SubProcessBase()
base._run([], 'link', ('list',))
self.execute.assert_called_once_with(['ip', 'link', 'list'],
run_as_root=False,
log_fail_as_error=True)
def test_run_namespace(self):
base = ip_lib.SubProcessBase(namespace='ns')
base._run([], 'link', ('list',))
self.execute.assert_called_once_with(['ip', 'netns', 'exec', 'ns',
'ip', 'link', 'list'],
run_as_root=True,
log_fail_as_error=True)
def test_as_root_namespace(self):
base = ip_lib.SubProcessBase(namespace='ns')
base._as_root([], 'link', ('list',))
self.execute.assert_called_once_with(['ip', 'netns', 'exec', 'ns',
'ip', 'link', 'list'],
run_as_root=True,
log_fail_as_error=True)
class TestIpWrapper(base.BaseTestCase):
def setUp(self):
super(TestIpWrapper, self).setUp()
self.execute_p = mock.patch.object(ip_lib.IPWrapper, '_execute')
self.execute = self.execute_p.start()
@mock.patch('os.path.islink')
@mock.patch('os.listdir', return_value=['lo'])
def test_get_devices(self, mocked_listdir, mocked_islink):
retval = ip_lib.IPWrapper().get_devices()
mocked_islink.assert_called_once_with('/sys/class/net/lo')
self.assertEqual([], retval)
@mock.patch('neutron.agent.common.utils.execute')
def test_get_devices_namespaces(self, mocked_execute):
fake_str = mock.Mock()
fake_str.split.return_value = ['lo']
mocked_execute.return_value = fake_str
retval = ip_lib.IPWrapper(namespace='foo').get_devices()
mocked_execute.assert_called_once_with(
['ip', 'netns', 'exec', 'foo', 'find', '/sys/class/net',
'-maxdepth', '1', '-type', 'l', '-printf', '%f '],
run_as_root=True, log_fail_as_error=True)
self.assertTrue(fake_str.split.called)
self.assertEqual([], retval)
@mock.patch('neutron.agent.common.utils.execute')
def test_get_devices_namespaces_ns_not_exists(self, mocked_execute):
mocked_execute.side_effect = RuntimeError(
"Cannot open network namespace")
with mock.patch.object(ip_lib.IpNetnsCommand, 'exists',
return_value=False):
retval = ip_lib.IPWrapper(namespace='foo').get_devices()
self.assertEqual([], retval)
@mock.patch('neutron.agent.common.utils.execute')
def test_get_devices_namespaces_ns_exists(self, mocked_execute):
mocked_execute.side_effect = RuntimeError(
"Cannot open network namespace")
with mock.patch.object(ip_lib.IpNetnsCommand, 'exists',
return_value=True):
self.assertRaises(RuntimeError,
ip_lib.IPWrapper(namespace='foo').get_devices)
@mock.patch('neutron.agent.common.utils.execute')
def test_get_devices_exclude_loopback_and_gre(self, mocked_execute):
device_name = 'somedevice'
mocked_execute.return_value = 'lo gre0 sit0 ip6gre0 ' + device_name
devices = ip_lib.IPWrapper(namespace='foo').get_devices(
exclude_loopback=True, exclude_fb_tun_devices=True)
somedevice = devices.pop()
self.assertEqual(device_name, somedevice.name)
self.assertFalse(devices)
@mock.patch.object(pyroute2.netns, 'listnetns')
@mock.patch.object(priv_lib, 'list_netns')
def test_get_namespaces_non_root(self, priv_listnetns, listnetns):
self.config(group='AGENT', use_helper_for_ns_read=False)
listnetns.return_value = NETNS_SAMPLE
retval = ip_lib.list_network_namespaces()
self.assertEqual(retval,
['12345678-1234-5678-abcd-1234567890ab',
'bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb',
'cccccccc-cccc-cccc-cccc-cccccccccccc'])
self.assertEqual(1, listnetns.call_count)
self.assertFalse(priv_listnetns.called)
@mock.patch.object(pyroute2.netns, 'listnetns')
@mock.patch.object(priv_lib, 'list_netns')
def test_get_namespaces_root(self, priv_listnetns, listnetns):
self.config(group='AGENT', use_helper_for_ns_read=True)
priv_listnetns.return_value = NETNS_SAMPLE
retval = ip_lib.list_network_namespaces()
self.assertEqual(retval,
['12345678-1234-5678-abcd-1234567890ab',
'bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb',
'cccccccc-cccc-cccc-cccc-cccccccccccc'])
self.assertEqual(1, priv_listnetns.call_count)
self.assertFalse(listnetns.called)
@mock.patch.object(priv_lib, 'create_interface')
def test_add_tuntap(self, create):
ip_lib.IPWrapper().add_tuntap('tap0')
create.assert_called_once_with('tap0', None, 'tuntap', mode='tap')
@mock.patch.object(priv_lib, 'create_interface')
def test_add_veth(self, create):
ip_lib.IPWrapper().add_veth('tap0', 'tap1')
create.assert_called_once_with(
'tap0', None, 'veth', peer={'ifname': 'tap1'})
@mock.patch.object(priv_lib, 'create_interface')
def test_add_macvtap(self, create):
ip_lib.IPWrapper().add_macvtap('macvtap0', 'eth0', 'bridge')
create.assert_called_once_with(
'macvtap0', None, 'macvtap', physical_interface='eth0',
mode='bridge')
@mock.patch.object(priv_lib, 'delete_interface')
def test_del_veth(self, delete):
ip_lib.IPWrapper().del_veth('fpr-1234')
delete.assert_called_once_with('fpr-1234', None)
@mock.patch.object(priv_lib, 'create_interface')
def test_add_veth_with_namespaces(self, create):
ns2 = 'ns2'
with mock.patch.object(ip_lib.IPWrapper, 'ensure_namespace') as en:
ip_lib.IPWrapper().add_veth('tap0', 'tap1', namespace2=ns2)
en.assert_has_calls([mock.call(ns2)])
create.assert_called_once_with(
'tap0', None, 'veth',
peer={'ifname': 'tap1', 'net_ns_fd': 'ns2'})
@mock.patch.object(priv_lib, 'create_interface')
def test_add_dummy(self, create):
ip_lib.IPWrapper().add_dummy('dummy0')
create.assert_called_once_with('dummy0', None, 'dummy')
def test_get_device(self):
dev = ip_lib.IPWrapper(namespace='ns').device('eth0')
self.assertEqual(dev.namespace, 'ns')
self.assertEqual(dev.name, 'eth0')
@mock.patch.object(priv_lib, 'create_netns')
def test_ensure_namespace(self, create):
with mock.patch.object(ip_lib, 'IPDevice') as ip_dev:
ip = ip_lib.IPWrapper()
with mock.patch.object(ip.netns, 'exists') as ns_exists:
with mock.patch('neutron.agent.common.utils.execute'):
ns_exists.return_value = False
ip.ensure_namespace('ns')
create.assert_called_once_with('ns')
ns_exists.assert_called_once_with('ns')
ip_dev.assert_has_calls([mock.call('lo', namespace='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().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(namespace='ns')
with mock.patch.object(ip, 'get_devices') as get_devices:
get_devices.return_value = []
self.assertTrue(ip.namespace_is_empty())
self.assertTrue(get_devices.called)
def test_namespace_is_empty(self):
ip = ip_lib.IPWrapper(namespace='ns')
with mock.patch.object(ip, 'get_devices') as get_devices:
get_devices.return_value = [mock.Mock()]
self.assertFalse(ip.namespace_is_empty())
self.assertTrue(get_devices.called)
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(namespace='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(namespace='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(namespace='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)
@mock.patch.object(priv_lib, 'create_interface')
def test_add_vlan(self, create):
retval = ip_lib.IPWrapper().add_vlan('eth0.1', 'eth0', '1')
self.assertIsInstance(retval, ip_lib.IPDevice)
self.assertEqual(retval.name, 'eth0.1')
create.assert_called_once_with('eth0.1',
None,
'vlan',
physical_interface='eth0',
vlan_id='1')
@mock.patch.object(priv_lib, 'create_interface')
def test_add_vxlan_valid_srcport_length(self, create):
self.call_params = {}
def fake_create_interface(ifname, namespace, kind, **kwargs):
self.call_params = dict(
ifname=ifname,
namespace=namespace,
kind=kind,
**kwargs)
create.side_effect = fake_create_interface
expected_call_params = {
'ifname': 'vxlan0',
'namespace': None,
'kind': 'vxlan',
'vxlan_id': 'vni0',
'vxlan_group': 'group0',
'physical_interface': 'dev0',
'vxlan_ttl': 'ttl0',
'vxlan_tos': 'tos0',
'vxlan_local': 'local0',
'vxlan_proxy': True,
'vxlan_port_range': ('1', '2')}
retval = ip_lib.IPWrapper().add_vxlan('vxlan0', 'vni0',
group='group0',
dev='dev0', ttl='ttl0',
tos='tos0',
local='local0', proxy=True,
srcport=(1, 2))
self.assertIsInstance(retval, ip_lib.IPDevice)
self.assertEqual(retval.name, 'vxlan0')
self.assertDictEqual(expected_call_params, self.call_params)
def test_add_vxlan_invalid_srcport_length(self):
wrapper = ip_lib.IPWrapper()
self.assertRaises(n_exc.NetworkVxlanPortRangeError,
wrapper.add_vxlan, 'vxlan0', 'vni0', group='group0',
dev='dev0', ttl='ttl0', tos='tos0',
local='local0', proxy=True,
srcport=('1', '2', '3'))
def test_add_vxlan_invalid_srcport_range(self):
wrapper = ip_lib.IPWrapper()
self.assertRaises(n_exc.NetworkVxlanPortRangeError,
wrapper.add_vxlan, 'vxlan0', 'vni0', group='group0',
dev='dev0', ttl='ttl0', tos='tos0',
local='local0', proxy=True,
srcport=(2000, 1000))
@mock.patch.object(priv_lib, 'create_interface')
def test_add_vxlan_dstport(self, create):
self.call_params = {}
def fake_create_interface(ifname, namespace, kind, **kwargs):
self.call_params = dict(
ifname=ifname,
namespace=namespace,
kind=kind,
**kwargs)
create.side_effect = fake_create_interface
expected_call_params = {
'ifname': 'vxlan0',
'namespace': None,
'kind': 'vxlan',
'vxlan_id': 'vni0',
'vxlan_group': 'group0',
'physical_interface': 'dev0',
'vxlan_ttl': 'ttl0',
'vxlan_tos': 'tos0',
'vxlan_local': 'local0',
'vxlan_proxy': True,
'vxlan_port_range': ('1', '2'),
'vxlan_port': 4789}
retval = ip_lib.IPWrapper().add_vxlan('vxlan0', 'vni0',
group='group0',
dev='dev0', ttl='ttl0',
tos='tos0',
local='local0', proxy=True,
srcport=(1, 2),
dstport=4789)
self.assertIsInstance(retval, ip_lib.IPDevice)
self.assertEqual(retval.name, 'vxlan0')
self.assertDictEqual(expected_call_params, self.call_params)
def test_add_device_to_namespace(self):
dev = mock.Mock()
ip_lib.IPWrapper(namespace='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().add_device_to_namespace(dev)
self.assertEqual([], dev.mock_calls)
class TestIPDevice(base.BaseTestCase):
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', namespace='ns1')
dev2 = ip_lib.IPDevice('tap0', namespace='ns2')
self.assertNotEqual(dev1, dev2)
def test_eq_other_is_none(self):
dev1 = ip_lib.IPDevice('tap0', namespace='ns1')
self.assertIsNotNone(dev1)
def test_str(self):
self.assertEqual(str(ip_lib.IPDevice('tap0')), 'tap0')
class TestIPCommandBase(base.BaseTestCase):
def setUp(self):
super(TestIPCommandBase, self).setUp()
self.ip = mock.Mock()
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(['o'], ('link'))
self.ip.assert_has_calls([mock.call._run(['o'], 'foo', ('link'))])
def test_as_root_namespace_false(self):
self.ip_cmd._as_root([], ('link'))
self.ip.assert_has_calls(
[mock.call._as_root([],
'foo',
('link'),
use_root_namespace=False)])
def test_as_root_namespace_true(self):
self.ip_cmd._as_root([], ('link'), use_root_namespace=True)
self.ip.assert_has_calls(
[mock.call._as_root([],
'foo',
('link'),
use_root_namespace=True)])
def test_as_root_namespace_true_with_options(self):
self.ip_cmd._as_root('o', 'link', use_root_namespace=True)
self.ip.assert_has_calls(
[mock.call._as_root('o',
'foo',
('link'),
use_root_namespace=True)])
class TestIPDeviceCommandBase(base.BaseTestCase):
def setUp(self):
super(TestIPDeviceCommandBase, self).setUp()
self.ip_dev = mock.Mock()
self.ip_dev.name = 'eth0'
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(base.BaseTestCase):
def setUp(self):
super(TestIPCmdBase, self).setUp()
self.parent = mock.Mock()
self.parent.name = 'eth0'
def _assert_call(self, options, args):
self.parent._run.assert_has_calls([
mock.call(options, self.command, args)])
def _assert_sudo(self, options, args, use_root_namespace=False):
self.parent._as_root.assert_has_calls(
[mock.call(options, self.command, args,
use_root_namespace=use_root_namespace)])
class TestIpRuleCommand(TestIPCmdBase):
def setUp(self):
super(TestIpRuleCommand, self).setUp()
self.parent._as_root.return_value = ''
self.command = 'rule'
self.rule_cmd = ip_lib.IpRuleCommand(self.parent)
def _test_add_rule(self, ip, table, priority):
ip_version = netaddr.IPNetwork(ip).version
self.rule_cmd.add(ip, table=table, priority=priority)
self._assert_sudo([ip_version], (['show']))
self._assert_sudo([ip_version], ('add', 'from', ip,
'priority', str(priority),
'table', str(table),
'type', 'unicast'))
def _test_add_rule_exists(self, ip, table, priority, output):
self.parent._as_root.return_value = output
ip_version = netaddr.IPNetwork(ip).version
self.rule_cmd.add(ip, table=table, priority=priority)
self._assert_sudo([ip_version], (['show']))
def _test_delete_rule(self, ip, table, priority):
ip_version = netaddr.IPNetwork(ip).version
self.rule_cmd.delete(ip, table=table, priority=priority)
self._assert_sudo([ip_version],
('del', 'from', ip, 'priority', str(priority),
'table', str(table), 'type', 'unicast'))
def test__parse_line(self):
def test(ip_version, line, expected):
actual = self.rule_cmd._parse_line(ip_version, line)
self.assertEqual(expected, actual)
test(4, "4030201:\tfrom 1.2.3.4/24 lookup 10203040",
{'from': '1.2.3.4/24',
'table': '10203040',
'type': 'unicast',
'priority': '4030201'})
test(6, "1024: from all iif qg-c43b1928-48 lookup noscope",
{'priority': '1024',
'from': '::/0',
'type': 'unicast',
'iif': 'qg-c43b1928-48',
'table': 'noscope'})
def test__make_canonical_all_v4(self):
actual = self.rule_cmd._make_canonical(4, {'from': 'all'})
self.assertEqual({'from': '0.0.0.0/0', 'type': 'unicast'}, actual)
def test__make_canonical_all_v6(self):
actual = self.rule_cmd._make_canonical(6, {'from': 'all'})
self.assertEqual({'from': '::/0', 'type': 'unicast'}, actual)
def test__make_canonical_lookup(self):
actual = self.rule_cmd._make_canonical(6, {'lookup': 'table'})
self.assertEqual({'table': 'table', 'type': 'unicast'}, actual)
def test__make_canonical_iif(self):
actual = self.rule_cmd._make_canonical(6, {'iif': 'iface_name'})
self.assertEqual({'iif': 'iface_name', 'type': 'unicast'}, actual)
def test__make_canonical_fwmark(self):
actual = self.rule_cmd._make_canonical(6, {'fwmark': '0x400'})
self.assertEqual({'fwmark': '0x400/0xffffffff',
'type': 'unicast'}, actual)
def test__make_canonical_fwmark_with_mask(self):
actual = self.rule_cmd._make_canonical(6, {'fwmark': '0x400/0x00ff'})
self.assertEqual({'fwmark': '0x400/0xff', 'type': 'unicast'}, actual)
def test__make_canonical_fwmark_integer(self):
actual = self.rule_cmd._make_canonical(6, {'fwmark': 0x400})
self.assertEqual({'fwmark': '0x400/0xffffffff',
'type': 'unicast'}, actual)
def test__make_canonical_fwmark_iterable(self):
actual = self.rule_cmd._make_canonical(6, {'fwmark': (0x400, 0xffff)})
self.assertEqual({'fwmark': '0x400/0xffff', 'type': 'unicast'}, actual)
def test_add_rule_v4(self):
self._test_add_rule('192.168.45.100', 2, 100)
def test_add_rule_v4_exists(self):
self._test_add_rule_exists('192.168.45.100', 2, 101, RULE_V4_SAMPLE)
def test_add_rule_v6(self):
self._test_add_rule('2001:db8::1', 3, 200)
def test_add_rule_v6_exists(self):
self._test_add_rule_exists('2001:db8::1', 3, 201, RULE_V6_SAMPLE)
def test_delete_rule_v4(self):
self._test_delete_rule('192.168.45.100', 2, 100)
def test_delete_rule_v6(self):
self._test_delete_rule('2001:db8::1', 3, 200)
class TestIpLinkCommand(TestIPCmdBase):
def setUp(self):
super(TestIpLinkCommand, self).setUp()
self.command = 'link'
self.link_cmd = ip_lib.IpLinkCommand(self.parent)
@mock.patch.object(priv_lib, 'set_link_attribute')
def test_set_address(self, set_link_attribute):
self.link_cmd.set_address('aa:bb:cc:dd:ee:ff')
set_link_attribute.assert_called_once_with(
self.parent.name, self.parent.namespace,
address='aa:bb:cc:dd:ee:ff')
@mock.patch.object(priv_lib, 'set_link_flags')
def test_set_allmulticast_on(self, set_link_flags):
self.link_cmd.set_allmulticast_on()
set_link_flags.assert_called_once_with(
self.parent.name, self.parent.namespace, ifinfmsg.IFF_ALLMULTI)
@mock.patch.object(priv_lib, 'set_link_attribute')
def test_set_mtu(self, set_link_attribute):
self.link_cmd.set_mtu(1500)
set_link_attribute.assert_called_once_with(
self.parent.name, self.parent.namespace, mtu=1500)
@mock.patch.object(priv_lib, 'set_link_attribute')
def test_set_up(self, set_link_attribute):
self.link_cmd.set_up()
set_link_attribute.assert_called_once_with(
self.parent.name, self.parent.namespace, state='up')
@mock.patch.object(priv_lib, 'set_link_attribute')
def test_set_down(self, set_link_attribute):
self.link_cmd.set_down()
set_link_attribute.assert_called_once_with(
self.parent.name, self.parent.namespace, state='down')
@mock.patch.object(priv_lib, 'set_link_attribute')
def test_set_netns(self, set_link_attribute):
original_namespace = self.parent.namespace
self.link_cmd.set_netns('foo')
set_link_attribute.assert_called_once_with(
'eth0', original_namespace, net_ns_fd='foo')
self.assertEqual(self.parent.namespace, 'foo')
@mock.patch.object(priv_lib, 'set_link_attribute')
def test_set_name(self, set_link_attribute):
original_name = self.parent.name
self.link_cmd.set_name('tap1')
set_link_attribute.assert_called_once_with(
original_name, self.parent.namespace, ifname='tap1')
self.assertEqual(self.parent.name, 'tap1')
@mock.patch.object(priv_lib, 'set_link_attribute')
def test_set_alias(self, set_link_attribute):
self.link_cmd.set_alias('openvswitch')
set_link_attribute.assert_called_once_with(
self.parent.name, self.parent.namespace, ifalias='openvswitch')
@mock.patch.object(priv_lib, 'delete_interface')
def test_delete(self, delete):
self.link_cmd.delete()
delete.assert_called_once_with(self.parent.name, self.parent.namespace)
@mock.patch.object(priv_lib, 'get_link_attributes')
def test_settings_property(self, get_link_attributes):
self.link_cmd.attributes
get_link_attributes.assert_called_once_with(
self.parent.name, self.parent.namespace)
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)
@mock.patch.object(priv_lib, 'add_ip_address')
def test_add_address(self, add):
self.addr_cmd.add('192.168.45.100/24')
add.assert_called_once_with(
4,
'192.168.45.100',
24,
self.parent.name,
self.addr_cmd._parent.namespace,
'global',
'192.168.45.255')
@mock.patch.object(priv_lib, 'add_ip_address')
def test_add_address_scoped(self, add):
self.addr_cmd.add('192.168.45.100/24', scope='link')
add.assert_called_once_with(
4,
'192.168.45.100',
24,
self.parent.name,
self.addr_cmd._parent.namespace,
'link',
'192.168.45.255')
@mock.patch.object(priv_lib, 'add_ip_address')
def test_add_address_no_broadcast(self, add):
self.addr_cmd.add('192.168.45.100/24', add_broadcast=False)
add.assert_called_once_with(
4,
'192.168.45.100',
24,
self.parent.name,
self.addr_cmd._parent.namespace,
'global',
None)
@mock.patch.object(priv_lib, 'delete_ip_address')
def test_del_address(self, delete):
self.addr_cmd.delete('192.168.45.100/24')
delete.assert_called_once_with(
4,
'192.168.45.100',
24,
self.parent.name,
self.addr_cmd._parent.namespace)
@mock.patch.object(priv_lib, 'flush_ip_addresses')
def test_flush(self, flush):
self.addr_cmd.flush(6)
flush.assert_called_once_with(
6, self.parent.name, self.addr_cmd._parent.namespace)
def test_list(self):
expected_brd = [
dict(name='eth0', scope='global', tentative=False, dadfailed=False,
dynamic=False, cidr='172.16.77.240/24',
broadcast='172.16.77.255')]
expected_no_brd = [
dict(name='eth0', scope='global', tentative=False, dadfailed=False,
dynamic=False, cidr='172.16.77.240/24', broadcast=None)]
expected_ipv6 = [
dict(name='eth0', scope='global', dadfailed=False, tentative=False,
dynamic=True, cidr='2001:470:9:1224:5595:dd51:6ba2:e788/64',
broadcast=None),
dict(name='eth0', scope='link', dadfailed=False, tentative=True,
dynamic=False, cidr='fe80::3023:39ff:febc:22ae/64',
broadcast=None),
dict(name='eth0', scope='link', dadfailed=True, tentative=True,
dynamic=False, cidr='fe80::3023:39ff:febc:22af/64',
broadcast=None),
dict(name='eth0', scope='global', dadfailed=False, tentative=False,
dynamic=True, cidr='2001:470:9:1224:fd91:272:581e:3a32/64',
broadcast=None),
dict(name='eth0', scope='global', dadfailed=False, tentative=False,
dynamic=True, cidr='2001:470:9:1224:4508:b885:5fb:740b/64',
broadcast=None),
dict(name='eth0', scope='global', dadfailed=False, tentative=False,
dynamic=True, cidr='2001:470:9:1224:dfcc:aaff:feb9:76ce/64',
broadcast=None),
dict(name='eth0', scope='link', dadfailed=False, tentative=False,
dynamic=False, cidr='fe80::dfcc:aaff:feb9:76ce/64',
broadcast=None)]
cases = [
(ADDR_SAMPLE, expected_brd + expected_ipv6),
(ADDR_SAMPLE2, expected_no_brd + expected_ipv6)]
for test_case, expected in cases:
self.parent._run = mock.Mock(return_value=test_case)
self.assertEqual(expected, self.addr_cmd.list())
self._assert_call([], ('show', 'tap0'))
def test_wait_until_address_ready(self):
self.parent._run.return_value = ADDR_SAMPLE
# this address is not tentative or failed so it should return
self.assertIsNone(self.addr_cmd.wait_until_address_ready(
'2001:470:9:1224:fd91:272:581e:3a32'))
def test_wait_until_address_ready_non_existent_address(self):
self.addr_cmd.list = mock.Mock(return_value=[])
with testtools.ExpectedException(ip_lib.AddressNotReady):
self.addr_cmd.wait_until_address_ready('abcd::1234')
def test_wait_until_address_ready_timeout(self):
tentative_address = 'fe80::3023:39ff:febc:22ae'
self.addr_cmd.list = mock.Mock(return_value=[
dict(scope='link', dadfailed=False, tentative=True, dynamic=False,
cidr=tentative_address + '/64')])
with testtools.ExpectedException(ip_lib.AddressNotReady):
self.addr_cmd.wait_until_address_ready(tentative_address,
wait_time=1)
def test_list_filtered(self):
expected_brd = [
dict(name='eth0', scope='global', tentative=False, dadfailed=False,
dynamic=False, cidr='172.16.77.240/24',
broadcast='172.16.77.255')]
expected_no_brd = [
dict(name='eth0', scope='global', tentative=False, dadfailed=False,
dynamic=False, cidr='172.16.77.240/24', broadcast=None)]
cases = [
(ADDR_SAMPLE, expected_brd), (ADDR_SAMPLE2, expected_no_brd)]
for test_case, expected in cases:
output = '\n'.join(test_case.split('\n')[0:4])
self.parent._run.return_value = output
self.assertEqual(
expected,
self.addr_cmd.list(
'global', filters=['permanent']))
self._assert_call([], ('show', 'tap0', 'permanent', 'scope',
'global'))
def test_get_devices_with_ip(self):
# This can only verify that get_devices_with_ip() returns a dict
# with the correct entry, it doesn't actually test that it only
# returns items filtered by the arguments since it isn't calling
# /sbin/ip at all.
self.parent._run.return_value = ADDR_SAMPLE3
devices = self.addr_cmd.get_devices_with_ip(to='172.16.77.240/24')
self.assertEqual(1, len(devices))
expected = {'cidr': '172.16.77.240/24',
'broadcast': '172.16.77.255',
'dadfailed': False,
'dynamic': False,
'name': 'eth0',
'scope': 'global',
'tentative': False}
self.assertEqual(expected, devices[0])
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)
self.ip_version = constants.IP_VERSION_4
self.table = 14
self.metric = 100
self.cidr = '192.168.45.100/24'
self.ip = '10.0.0.1'
self.gateway = '192.168.45.100'
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'}},
{'sample': GATEWAY_SAMPLE5,
'expected': {'gateway': '192.168.99.1'}},
{'sample': GATEWAY_SAMPLE6,
'expected': {'gateway': '192.168.99.1',
'metric': 100}},
{'sample': GATEWAY_SAMPLE7,
'expected': {'metric': 1}}]
def test_add_gateway(self):
self.route_cmd.add_gateway(self.gateway, self.metric, self.table)
self._assert_sudo([self.ip_version],
('replace', 'default',
'via', self.gateway,
'metric', self.metric,
'dev', self.parent.name,
'table', self.table))
def test_add_gateway_subtable(self):
self.route_cmd.table(self.table).add_gateway(self.gateway, self.metric)
self._assert_sudo([self.ip_version],
('replace', 'default',
'via', self.gateway,
'metric', self.metric,
'dev', self.parent.name,
'table', self.table))
def test_del_gateway_success(self):
self.route_cmd.delete_gateway(self.gateway, table=self.table)
self._assert_sudo([self.ip_version],
('del', 'default',
'via', self.gateway,
'dev', self.parent.name,
'table', self.table))
def test_del_gateway_success_subtable(self):
self.route_cmd.table(table=self.table).delete_gateway(self.gateway)
self._assert_sudo([self.ip_version],
('del', 'default',
'via', self.gateway,
'dev', self.parent.name,
'table', self.table))
def test_del_gateway_cannot_find_device(self):
self.parent._as_root.side_effect = RuntimeError("Cannot find device")
exc = self.assertRaises(exceptions.DeviceNotFoundError,
self.route_cmd.delete_gateway,
self.gateway, table=self.table)
self.assertIn(self.parent.name, str(exc))
def test_del_gateway_other_error(self):
self.parent._as_root.side_effect = RuntimeError()
self.assertRaises(RuntimeError, self.route_cmd.delete_gateway,
self.gateway, table=self.table)
def test_get_gateway(self):
for test_case in self.test_cases:
self.parent._run = mock.Mock(return_value=test_case['sample'])
self.assertEqual(self.route_cmd.get_gateway(),
test_case['expected'])
def test_flush_route_table(self):
self.route_cmd.flush(self.ip_version, self.table)
self._assert_sudo([self.ip_version], ('flush', 'table', self.table))
def test_add_route(self):
self.route_cmd.add_route(self.cidr, self.ip, self.table)
self._assert_sudo([self.ip_version],
('replace', self.cidr,
'via', self.ip,
'dev', self.parent.name,
'table', self.table))
def test_add_route_no_via(self):
self.route_cmd.add_route(self.cidr, table=self.table)
self._assert_sudo([self.ip_version],
('replace', self.cidr,
'dev', self.parent.name,
'table', self.table))
def test_add_route_with_scope(self):
self.route_cmd.add_route(self.cidr, scope='link')
self._assert_sudo([self.ip_version],
('replace', self.cidr,
'dev', self.parent.name,
'scope', 'link'))
def test_add_route_no_device(self):
self.parent._as_root.side_effect = RuntimeError("Cannot find device")
self.assertRaises(exceptions.DeviceNotFoundError,
self.route_cmd.add_route,
self.cidr, self.ip, self.table)
def test_delete_route(self):
self.route_cmd.delete_route(self.cidr, self.ip, self.table)
self._assert_sudo([self.ip_version],
('del', self.cidr,
'via', self.ip,
'dev', self.parent.name,
'table', self.table))
def test_delete_route_no_via(self):
self.route_cmd.delete_route(self.cidr, table=self.table)
self._assert_sudo([self.ip_version],
('del', self.cidr,
'dev', self.parent.name,
'table', self.table))
def test_delete_route_with_scope(self):
self.route_cmd.delete_route(self.cidr, scope='link')
self._assert_sudo([self.ip_version],
('del', self.cidr,
'dev', self.parent.name,
'scope', 'link'))
def test_delete_route_no_device(self):
self.parent._as_root.side_effect = RuntimeError("Cannot find device")
self.assertRaises(exceptions.DeviceNotFoundError,
self.route_cmd.delete_route,
self.cidr, self.ip, self.table)
def test_list_routes(self):
self.parent._run.return_value = (
"default via 172.124.4.1 dev eth0 metric 100\n"
"10.0.0.0/22 dev eth0 scope link\n"
"172.24.4.0/24 dev eth0 proto kernel src 172.24.4.2\n")
routes = self.route_cmd.table(self.table).list_routes(self.ip_version)
self.assertEqual([{'cidr': '0.0.0.0/0',
'dev': 'eth0',
'metric': '100',
'table': 14,
'via': '172.124.4.1'},
{'cidr': '10.0.0.0/22',
'dev': 'eth0',
'scope': 'link',
'table': 14},
{'cidr': '172.24.4.0/24',
'dev': 'eth0',
'proto': 'kernel',
'src': '172.24.4.2',
'table': 14}], routes)
def test_list_onlink_routes_subtable(self):
self.parent._run.return_value = (
"10.0.0.0/22\n"
"172.24.4.0/24 proto kernel src 172.24.4.2\n")
routes = self.route_cmd.table(self.table).list_onlink_routes(
self.ip_version)
self.assertEqual(['10.0.0.0/22'], [r['cidr'] for r in routes])
self._assert_call([self.ip_version],
('list', 'dev', self.parent.name,
'table', self.table, 'scope', 'link'))
def test_add_onlink_route_subtable(self):
self.route_cmd.table(self.table).add_onlink_route(self.cidr)
self._assert_sudo([self.ip_version],
('replace', self.cidr,
'dev', self.parent.name,
'table', self.table,
'scope', 'link'))
def test_delete_onlink_route_subtable(self):
self.route_cmd.table(self.table).delete_onlink_route(self.cidr)
self._assert_sudo([self.ip_version],
('del', self.cidr,
'dev', self.parent.name,
'table', self.table,
'scope', 'link'))
class TestIPv6IpRouteCommand(TestIpRouteCommand):
def setUp(self):
super(TestIPv6IpRouteCommand, self).setUp()
self.ip_version = constants.IP_VERSION_6
self.cidr = '2001:db8::/64'
self.ip = '2001:db8::100'
self.gateway = '2001:db8::1'
self.test_cases = [{'sample': IPv6_GATEWAY_SAMPLE1,
'expected':
{'gateway': '2001:470:9:1224:4508:b885:5fb:740b',
'metric': 100}},
{'sample': IPv6_GATEWAY_SAMPLE2,
'expected':
{'gateway': '2001:470:9:1224:4508:b885:5fb:740b',
'metric': 100}},
{'sample': IPv6_GATEWAY_SAMPLE3,
'expected': None},
{'sample': IPv6_GATEWAY_SAMPLE4,
'expected':
{'gateway': 'fe80::dfcc:aaff:feb9:76ce'}},
{'sample': IPv6_GATEWAY_SAMPLE5,
'expected':
{'gateway': '2001:470:9:1224:4508:b885:5fb:740b',
'metric': 1024}}]
def test_list_routes(self):
self.parent._run.return_value = (
"default via 2001:db8::1 dev eth0 metric 100\n"
"2001:db8::/64 dev eth0 proto kernel src 2001:db8::2\n")
routes = self.route_cmd.table(self.table).list_routes(self.ip_version)
self.assertEqual([{'cidr': '::/0',
'dev': 'eth0',
'metric': '100',
'table': 14,
'via': '2001:db8::1'},
{'cidr': '2001:db8::/64',
'dev': 'eth0',
'proto': 'kernel',
'src': '2001:db8::2',
'table': 14}], routes)
class TestIPRoute(TestIpRouteCommand):
"""Leverage existing tests for IpRouteCommand for IPRoute
This test leverages the tests written for IpRouteCommand. The difference
is that the 'dev' argument should not be passed for each of the commands.
So, this test removes the dev argument from the expected arguments in each
assert.
"""
def setUp(self):
super(TestIPRoute, self).setUp()
self.parent = ip_lib.IPRoute()
self.parent._run = mock.Mock()
self.parent._as_root = mock.Mock()
self.route_cmd = self.parent.route
self.check_dev_args = False
def _remove_dev_args(self, args):
def args_without_dev():
previous = None
for arg in args:
if 'dev' not in (arg, previous):
yield arg
previous = arg
return tuple(arg for arg in args_without_dev())
def _assert_call(self, options, args):
if not self.check_dev_args:
args = self._remove_dev_args(args)
super(TestIPRoute, self)._assert_call(options, args)
def _assert_sudo(self, options, args, use_root_namespace=False):
if not self.check_dev_args:
args = self._remove_dev_args(args)
super(TestIPRoute, self)._assert_sudo(options, args)
def test_del_gateway_cannot_find_device(self):
# This test doesn't make sense for this case since dev won't be passed
pass
class TestIpNetnsCommand(TestIPCmdBase):
def setUp(self):
super(TestIpNetnsCommand, self).setUp()
self.command = 'netns'
self.netns_cmd = ip_lib.IpNetnsCommand(self.parent)
@mock.patch.object(priv_lib, 'create_netns')
def test_add_namespace(self, create):
with mock.patch('neutron.agent.common.utils.execute') as execute:
ns = self.netns_cmd.add('ns')
create.assert_called_once_with('ns')
self.assertEqual(ns.namespace, 'ns')
execute.assert_called_once_with(
['ip', 'netns', 'exec', 'ns',
'sysctl', '-w', 'net.ipv4.conf.all.promote_secondaries=1'],
run_as_root=True, check_exit_code=True, extra_ok_codes=None,
log_fail_as_error=True)
@mock.patch.object(priv_lib, 'remove_netns')
def test_delete_namespace(self, remove):
self.netns_cmd.delete('ns')
remove.assert_called_once_with('ns')
@mock.patch.object(pyroute2.netns, 'listnetns')
@mock.patch.object(priv_lib, 'list_netns')
def test_namespace_exists_use_helper(self, priv_listnetns, listnetns):
self.config(group='AGENT', use_helper_for_ns_read=True)
priv_listnetns.return_value = NETNS_SAMPLE
# need another instance to avoid mocking
netns_cmd = ip_lib.IpNetnsCommand(ip_lib.SubProcessBase())
self.assertTrue(
netns_cmd.exists('bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb'))
self.assertEqual(1, priv_listnetns.call_count)
self.assertFalse(listnetns.called)
@mock.patch.object(pyroute2.netns, 'listnetns')
@mock.patch.object(priv_lib, 'list_netns')
def test_namespace_does_not_exist_no_helper(self, priv_listnetns,
listnetns):
self.config(group='AGENT', use_helper_for_ns_read=False)
listnetns.return_value = NETNS_SAMPLE
# need another instance to avoid mocking
netns_cmd = ip_lib.IpNetnsCommand(ip_lib.SubProcessBase())
self.assertFalse(
netns_cmd.exists('bbbbbbbb-1111-2222-3333-bbbbbbbbbbbb'))
self.assertEqual(1, listnetns.call_count)
self.assertFalse(priv_listnetns.called)
def test_execute(self):
self.parent.namespace = 'ns'
with mock.patch('neutron.agent.common.utils.execute') as execute:
self.netns_cmd.execute(['ip', 'link', 'list'])
execute.assert_called_once_with(['ip', 'netns', 'exec', 'ns', 'ip',
'link', 'list'],
run_as_root=True,
check_exit_code=True,
extra_ok_codes=None,
log_fail_as_error=True)
def test_execute_env_var_prepend(self):
self.parent.namespace = 'ns'
with mock.patch('neutron.agent.common.utils.execute') as execute:
env = dict(FOO=1, BAR=2)
self.netns_cmd.execute(['ip', 'link', 'list'], env)
execute.assert_called_once_with(
['ip', 'netns', 'exec', 'ns', 'env'] +
['%s=%s' % (k, v) for k, v in env.items()] +
['ip', 'link', 'list'],
run_as_root=True, check_exit_code=True, extra_ok_codes=None,
log_fail_as_error=True)
def test_execute_nosudo_with_no_namespace(self):
with mock.patch('neutron.agent.common.utils.execute') as execute:
self.parent.namespace = None
self.netns_cmd.execute(['test'])
execute.assert_called_once_with(['test'],
check_exit_code=True,
extra_ok_codes=None,
run_as_root=False,
log_fail_as_error=True)
class TestDeviceExists(base.BaseTestCase):
def test_ensure_device_is_ready(self):
ip_lib_mock = mock.Mock()
with mock.patch.object(ip_lib, 'IPDevice', return_value=ip_lib_mock):
self.assertTrue(ip_lib.ensure_device_is_ready("eth0"))
self.assertTrue(ip_lib_mock.link.set_up.called)
ip_lib_mock.reset_mock()
# device doesn't exists
ip_lib_mock.link.set_up.side_effect = RuntimeError
self.assertFalse(ip_lib.ensure_device_is_ready("eth0"))
def test_ensure_device_is_ready_no_link_address(self):
with mock.patch.object(
priv_lib, 'get_link_attributes'
) as get_link_attributes, mock.patch.object(
priv_lib, 'set_link_attribute'
) as set_link_attribute:
get_link_attributes.return_value = {}
self.assertFalse(ip_lib.ensure_device_is_ready("lo"))
get_link_attributes.assert_called_once_with("lo", None)
set_link_attribute.assert_not_called()
class TestGetRoutingTable(base.BaseTestCase):
ip_db_interfaces = {
1: {
'family': 0,
'txqlen': 0,
'ipdb_scope': 'system',
'index': 1,
'operstate': 'DOWN',
'num_tx_queues': 1,
'group': 0,
'carrier_changes': 0,
'ipaddr': [],
'neighbours': [],
'ifname': 'lo',
'promiscuity': 0,
'linkmode': 0,
'broadcast': '00:00:00:00:00:00',
'address': '00:00:00:00:00:00',
'vlans': [],
'ipdb_priority': 0,
'qdisc': 'noop',
'mtu': 65536,
'num_rx_queues': 1,
'carrier': 1,
'flags': 8,
'ifi_type': 772,
'ports': []
},
2: {
'family': 0,
'txqlen': 500,
'ipdb_scope': 'system',
'index': 2,
'operstate': 'DOWN',
'num_tx_queues': 1,
'group': 0,
'carrier_changes': 1,
'ipaddr': ['1111:1111:1111:1111::3/64', '10.0.0.3/24'],
'neighbours': [],
'ifname': 'tap-1',
'promiscuity': 0,
'linkmode': 0,
'broadcast': 'ff:ff:ff:ff:ff:ff',
'address': 'b6:d5:f6:a8:2e:62',
'vlans': [],
'ipdb_priority': 0,
'kind': 'tun',
'qdisc': 'fq_codel',
'mtu': 1500,
'num_rx_queues': 1,
'carrier': 0,
'flags': 4099,
'ifi_type': 1,
'ports': []
},
'tap-1': {
'family': 0,
'txqlen': 500,
'ipdb_scope': 'system',
'index': 2,
'operstate': 'DOWN',
'num_tx_queues': 1,
'group': 0,
'carrier_changes': 1,
'ipaddr': ['1111:1111:1111:1111::3/64', '10.0.0.3/24'],
'neighbours': [],
'ifname': 'tap-1',
'promiscuity': 0,
'linkmode': 0,
'broadcast': 'ff:ff:ff:ff:ff:ff',
'address': 'b6:d5:f6:a8:2e:62',
'vlans': [],
'ipdb_priority': 0,
'kind': 'tun',
'qdisc': 'fq_codel',
'mtu': 1500,
'num_rx_queues': 1,
'carrier': 0,
'flags': 4099,
'ifi_type': 1,
'ports': []
},
'lo': {
'family': 0,
'txqlen': 0,
'ipdb_scope': 'system',
'index': 1,
'operstate': 'DOWN',
'num_tx_queues': 1,
'group': 0,
'carrier_changes': 0,
'ipaddr': [],
'neighbours': [],
'ifname': 'lo',
'promiscuity': 0,
'linkmode': 0,
'broadcast': '00:00:00:00:00:00',
'address': '00:00:00:00:00:00',
'vlans': [],
'ipdb_priority': 0,
'qdisc': 'noop',
'mtu': 65536,
'num_rx_queues': 1,
'carrier': 1,
'flags': 8,
'ifi_type': 772,
'ports': []
}
}
ip_db_routes = [
{
'oif': 2,
'dst_len': 24,
'family': 2,
'proto': 3,
'tos': 0,
'dst': '10.0.1.0/24',
'flags': 16,
'ipdb_priority': 0,
'metrics': {},
'scope': 0,
'encap': {},
'src_len': 0,
'table': 254,
'multipath': [],
'type': 1,
'gateway': '10.0.0.1',
'ipdb_scope': 'system'
}, {
'oif': 2,
'type': 1,
'dst_len': 24,
'family': 2,
'proto': 2,
'tos': 0,
'dst': '10.0.0.0/24',
'ipdb_priority': 0,
'metrics': {},
'flags': 16,
'encap': {},
'src_len': 0,
'table': 254,
'multipath': [],
'prefsrc': '10.0.0.3',
'scope': 253,
'ipdb_scope': 'system'
}, {
'oif': 2,
'dst_len': 0,
'family': 2,
'proto': 3,
'tos': 0,
'dst': 'default',
'flags': 16,
'ipdb_priority': 0,
'metrics': {},
'scope': 0,
'encap': {},
'src_len': 0,
'table': 254,
'multipath': [],
'type': 1,
'gateway': '10.0.0.2',
'ipdb_scope': 'system'
}, {
'metrics': {},
'oif': 2,
'dst_len': 64,
'family': socket.AF_INET6,
'proto': 2,
'tos': 0,
'dst': '1111:1111:1111:1111::/64',
'pref': '00',
'ipdb_priority': 0,
'priority': 256,
'flags': 0,
'encap': {},
'src_len': 0,
'table': 254,
'multipath': [],
'type': 1,
'scope': 0,
'ipdb_scope': 'system'
}, {
'metrics': {},
'oif': 2,
'dst_len': 64,
'family': socket.AF_INET6,
'proto': 3,
'tos': 0,
'dst': '1111:1111:1111:1112::/64',
'pref': '00',
'flags': 0,
'ipdb_priority': 0,
'priority': 1024,
'scope': 0,
'encap': {},
'src_len': 0,
'table': 254,
'multipath': [],
'type': 1,
'gateway': '1111:1111:1111:1111::1',
'ipdb_scope': 'system'
}
]
ip_db_multipath_routes = [
{
'dst_len': 24,
'family': socket.AF_INET,
'proto': 3,
'tos': 0,
'dst': '10.0.1.0/24',
'flags': 16,
'ipdb_priority': 0,
'metrics': {},
'scope': 0,
'encap': {},
'src_len': 0,
'table': 254,
'multipath': ({'oif': 1, 'family': socket.AF_INET},
{'oif': 2, 'dst_len': 24, 'family': socket.AF_INET,
'proto': 2, 'tos': 0, 'pref': '00',
'priority': 256, 'flags': 0, 'encap': {},
'src_len': 0, 'table': 254, 'type': 1,
'scope': 0}),
'type': 1,
'gateway': '10.0.0.1',
'ipdb_scope': 'system'
}, {
'metrics': {},
'dst_len': 64,
'family': socket.AF_INET6,
'proto': 2,
'tos': 0,
'dst': '1111:1111:1111:1111::/64',
'pref': '00',
'ipdb_priority': 0,
'priority': 256,
'flags': 0,
'encap': {},
'src_len': 0,
'table': 254,
'multipath': ({'oif': 1, 'family': socket.AF_INET6},
{'oif': 2, 'dst_len': 64, 'family': socket.AF_INET6,
'proto': 2, 'tos': 0, 'pref': '00',
'priority': 256, 'flags': 0, 'encap': {},
'src_len': 0, 'table': 254, 'type': 1,
'scope': 0}),
'type': 1,
'scope': 0,
'ipdb_scope': 'system'
}
]
def setUp(self):
super(TestGetRoutingTable, self).setUp()
self.addCleanup(privileged.default.set_client_mode, True)
privileged.default.set_client_mode(False)
@mock.patch.object(pyroute2, 'IPDB')
@mock.patch.object(pyroute2, 'NetNS')
def test_get_routing_table_nonexistent_namespace(self,
mock_netns, mock_ip_db):
mock_netns.side_effect = OSError(errno.ENOENT, None)
with testtools.ExpectedException(ip_lib.NetworkNamespaceNotFound):
ip_lib.get_routing_table(4, 'ns')
@mock.patch.object(pyroute2, 'IPDB')
@mock.patch.object(pyroute2, 'NetNS')
def test_get_routing_table_other_error(self, mock_netns, mock_ip_db):
expected_exception = OSError(errno.EACCES, None)
mock_netns.side_effect = expected_exception
with testtools.ExpectedException(expected_exception.__class__):
ip_lib.get_routing_table(4, 'ns')
@mock.patch.object(pyroute2, 'IPDB')
@mock.patch.object(pyroute2, 'NetNS')
def _test_get_routing_table(self, version, ip_db_routes, expected,
mock_netns, mock_ip_db):
mock_ip_db_instance = mock_ip_db.return_value
mock_ip_db_enter = mock_ip_db_instance.__enter__.return_value
mock_ip_db_enter.interfaces = self.ip_db_interfaces
mock_ip_db_enter.routes = ip_db_routes
self.assertEqual(expected, ip_lib.get_routing_table(version))
def test_get_routing_table_4(self):
expected = [{'destination': '10.0.1.0/24',
'nexthop': '10.0.0.1',
'device': 'tap-1',
'scope': 'universe'},
{'destination': '10.0.0.0/24',
'nexthop': None,
'device': 'tap-1',
'scope': 'link'},
{'destination': 'default',
'nexthop': '10.0.0.2',
'device': 'tap-1',
'scope': 'universe'}]
self._test_get_routing_table(4, self.ip_db_routes, expected)
def test_get_routing_table_6(self):
expected = [{'destination': '1111:1111:1111:1111::/64',
'nexthop': None,
'device': 'tap-1',
'scope': 'universe'},
{'destination': '1111:1111:1111:1112::/64',
'nexthop': '1111:1111:1111:1111::1',
'device': 'tap-1',
'scope': 'universe'}]
self._test_get_routing_table(6, self.ip_db_routes, expected)
def test_get_routing_table_multipath_4(self):
expected = [{'destination': '10.0.1.0/24',
'nexthop': '10.0.0.1',
'device': 'lo',
'scope': 'universe'},
{'destination': '10.0.1.0/24',
'nexthop': '10.0.0.1',
'device': 'tap-1',
'scope': 'universe'}]
self._test_get_routing_table(4, self.ip_db_multipath_routes, expected)
def test_get_routing_table_multipath_6(self):
expected = [{'destination': '1111:1111:1111:1111::/64',
'nexthop': None,
'device': 'lo',
'scope': 'universe'},
{'destination': '1111:1111:1111:1111::/64',
'nexthop': None,
'device': 'tap-1',
'scope': 'universe'}]
self._test_get_routing_table(6, self.ip_db_multipath_routes, expected)
class TestIpNeighCommand(TestIPCmdBase):
def setUp(self):
super(TestIpNeighCommand, self).setUp()
self.parent.name = 'tap0'
self.command = 'neigh'
self.neigh_cmd = ip_lib.IpNeighCommand(self.parent)
self.addCleanup(privileged.default.set_client_mode, True)
privileged.default.set_client_mode(False)
@mock.patch.object(pyroute2, 'NetNS')
def test_add_entry(self, mock_netns):
mock_netns_instance = mock_netns.return_value
mock_netns_enter = mock_netns_instance.__enter__.return_value
mock_netns_enter.link_lookup.return_value = [1]
self.neigh_cmd.add('192.168.45.100', 'cc:dd:ee:ff:ab:cd')
mock_netns_enter.link_lookup.assert_called_once_with(ifname='tap0')
mock_netns_enter.neigh.assert_called_once_with(
'replace',
dst='192.168.45.100',
lladdr='cc:dd:ee:ff:ab:cd',
family=2,
ifindex=1,
state=ndmsg.states['permanent'])
@mock.patch.object(pyroute2, 'NetNS')
def test_add_entry_nonexistent_namespace(self, mock_netns):
mock_netns.side_effect = OSError(errno.ENOENT, None)
with testtools.ExpectedException(ip_lib.NetworkNamespaceNotFound):
self.neigh_cmd.add('192.168.45.100', 'cc:dd:ee:ff:ab:cd')
@mock.patch.object(pyroute2, 'NetNS')
def test_add_entry_other_error(self, mock_netns):
expected_exception = OSError(errno.EACCES, None)
mock_netns.side_effect = expected_exception
with testtools.ExpectedException(expected_exception.__class__):
self.neigh_cmd.add('192.168.45.100', 'cc:dd:ee:ff:ab:cd')
@mock.patch.object(pyroute2, 'NetNS')
def test_delete_entry(self, mock_netns):
mock_netns_instance = mock_netns.return_value
mock_netns_enter = mock_netns_instance.__enter__.return_value
mock_netns_enter.link_lookup.return_value = [1]
self.neigh_cmd.delete('192.168.45.100', 'cc:dd:ee:ff:ab:cd')
mock_netns_enter.link_lookup.assert_called_once_with(ifname='tap0')
mock_netns_enter.neigh.assert_called_once_with(
'delete',
dst='192.168.45.100',
lladdr='cc:dd:ee:ff:ab:cd',
family=2,
ifindex=1)
@mock.patch.object(priv_lib, '_run_iproute_neigh')
def test_delete_entry_not_exist(self, mock_run_iproute):
# trying to delete a non-existent entry shouldn't raise an error
mock_run_iproute.side_effect = NetlinkError(errno.ENOENT, None)
self.neigh_cmd.delete('192.168.45.100', 'cc:dd:ee:ff:ab:cd')
@mock.patch.object(pyroute2, 'NetNS')
def test_dump_entries(self, mock_netns):
mock_netns_instance = mock_netns.return_value
mock_netns_enter = mock_netns_instance.__enter__.return_value
mock_netns_enter.link_lookup.return_value = [1]
self.neigh_cmd.dump(4)
mock_netns_enter.link_lookup.assert_called_once_with(ifname='tap0')
mock_netns_enter.neigh.assert_called_once_with(
'dump',
family=2,
ifindex=1)
def test_flush(self):
self.neigh_cmd.flush(4, '192.168.0.1')
self._assert_sudo([4], ('flush', 'to', '192.168.0.1'))
class TestArpPing(TestIPCmdBase):
@mock.patch.object(ip_lib, 'IPWrapper')
@mock.patch('eventlet.spawn_n')
def test_send_ipv4_addr_adv_notif(self, spawn_n, mIPWrapper):
spawn_n.side_effect = lambda f: f()
ARPING_COUNT = 3
address = '20.0.0.1'
ip_lib.send_ip_addr_adv_notif(mock.sentinel.ns_name,
mock.sentinel.iface_name,
address,
ARPING_COUNT)
self.assertTrue(spawn_n.called)
mIPWrapper.assert_has_calls([
mock.call(namespace=mock.sentinel.ns_name),
mock.call().netns.execute(mock.ANY, extra_ok_codes=[1]),
mock.call().netns.execute(mock.ANY, extra_ok_codes=[1]),
mock.call().netns.execute(mock.ANY, extra_ok_codes=[1, 2]),
mock.call().netns.execute(mock.ANY, extra_ok_codes=[1, 2]),
mock.call().netns.execute(mock.ANY, extra_ok_codes=[1, 2]),
mock.call().netns.execute(mock.ANY, extra_ok_codes=[1, 2])])
ip_wrapper = mIPWrapper(namespace=mock.sentinel.ns_name)
# Just test that arping is called with the right arguments
for arg in ('-A', '-U'):
arping_cmd = ['arping', arg,
'-I', mock.sentinel.iface_name,
'-c', 1,
'-w', mock.ANY,
address]
ip_wrapper.netns.execute.assert_any_call(arping_cmd,
extra_ok_codes=mock.ANY)
@mock.patch.object(ip_lib, 'IPWrapper')
@mock.patch('eventlet.spawn_n')
def test_send_ipv4_addr_adv_notif_nodev(self, spawn_n, mIPWrapper):
spawn_n.side_effect = lambda f: f()
ip_wrapper = mIPWrapper(namespace=mock.sentinel.ns_name)
ip_wrapper.netns.execute.side_effect = RuntimeError
ARPING_COUNT = 3
address = '20.0.0.1'
with mock.patch.object(ip_lib, 'device_exists_with_ips_and_mac',
return_value=False):
ip_lib.send_ip_addr_adv_notif(mock.sentinel.ns_name,
mock.sentinel.iface_name,
address,
ARPING_COUNT)
# should return early with a single call when ENODEV
mIPWrapper.assert_has_calls([
mock.call(namespace=mock.sentinel.ns_name),
mock.call().netns.execute(mock.ANY, extra_ok_codes=[1])
] * 1)
@mock.patch('eventlet.spawn_n')
def test_no_ipv6_addr_notif(self, spawn_n):
ipv6_addr = 'fd00::1'
ip_lib.send_ip_addr_adv_notif(mock.sentinel.ns_name,
mock.sentinel.iface_name,
ipv6_addr,
3)
self.assertFalse(spawn_n.called)
class TestAddNamespaceToCmd(base.BaseTestCase):
def test_add_namespace_to_cmd_with_namespace(self):
cmd = ['ping', '8.8.8.8']
self.assertEqual(['ip', 'netns', 'exec', 'tmp'] + cmd,
ip_lib.add_namespace_to_cmd(cmd, 'tmp'))
def test_add_namespace_to_cmd_without_namespace(self):
cmd = ['ping', '8.8.8.8']
self.assertEqual(cmd, ip_lib.add_namespace_to_cmd(cmd, None))
class TestSetIpNonlocalBindForHaNamespace(base.BaseTestCase):
def test_setting_failure(self):
"""Make sure message is formatted correctly."""
with mock.patch.object(ip_lib, 'set_ip_nonlocal_bind', return_value=1):
ip_lib.set_ip_nonlocal_bind_for_namespace('foo')
class TestSysctl(base.BaseTestCase):
def setUp(self):
super(TestSysctl, self).setUp()
self.execute_p = mock.patch.object(ip_lib.IpNetnsCommand, 'execute')
self.execute = self.execute_p.start()
def test_disable_ipv6_when_ipv6_globally_enabled(self):
dev = ip_lib.IPDevice('tap0', 'ns1')
with mock.patch.object(ip_lib.ipv6_utils,
'is_enabled_and_bind_by_default',
return_value=True):
dev.disable_ipv6()
self.execute.assert_called_once_with(
['sysctl', '-w', 'net.ipv6.conf.tap0.disable_ipv6=1'],
log_fail_as_error=True, run_as_root=True)
def test_disable_ipv6_when_ipv6_globally_disabled(self):
dev = ip_lib.IPDevice('tap0', 'ns1')
with mock.patch.object(ip_lib.ipv6_utils,
'is_enabled_and_bind_by_default',
return_value=False):
dev.disable_ipv6()
self.assertFalse(self.execute.called)
class TestConntrack(base.BaseTestCase):
def setUp(self):
super(TestConntrack, self).setUp()
self.execute_p = mock.patch.object(ip_lib.IpNetnsCommand, 'execute')
self.execute = self.execute_p.start()
def test_delete_socket_conntrack_state(self):
device = ip_lib.IPDevice('tap0', 'ns1')
ip_str = '1.1.1.1'
dport = '3378'
protocol = 'tcp'
expect_cmd = ["conntrack", "-D", "-d", ip_str, '-p', protocol,
'--dport', dport]
device.delete_socket_conntrack_state(ip_str, dport, protocol)
self.execute.assert_called_once_with(expect_cmd, check_exit_code=True,
extra_ok_codes=[1])