def get_interface_name(self, network, port=None):
if not port:
port = self._get_or_create_port(network)
- return ('tap' + port.id)[:self.driver.DEV_NAME_LEN]
+ return self.driver.get_device_name(port)
def get_device_id(self, network):
# There could be more than one dhcp server per network, so create
# from linux IF_NAMESIZE
DEV_NAME_LEN = 14
+ DEV_NAME_PREFIX = 'tap'
def __init__(self, conf):
self.conf = conf
if not ip_lib.device_exists(bridge):
raise exception.BridgeDoesNotExist(bridge=bridge)
+ def get_device_name(self, port):
+ return (self.DEV_NAME_PREFIX + port.id)[:self.DEV_NAME_LEN]
+
@abc.abstractmethod
def plug(self, network_id, port_id, device_name, mac_address):
"""Plug in the interface."""
mac_address],
self.conf.root_helper)
- device = ip_lib.IPDevice(device_name, self.conf.root_helper)
+ ip = ip_lib.IPWrapper(self.conf.root_helper)
+ device = ip.device(device_name)
device.link.set_address(mac_address)
if self.conf.network_device_mtu:
device.link.set_mtu(self.conf.network_device_mtu)
class BridgeInterfaceDriver(LinuxInterfaceDriver):
"""Driver for creating bridge interfaces."""
- BRIDGE_NAME_PREFIX = 'brq'
+ DEV_NAME_PREFIX = 'dhc'
def plug(self, network_id, port_id, device_name, mac_address):
"""Plugin the interface."""
if not ip_lib.device_exists(device_name):
- device = ip_lib.IPDevice(device_name, self.conf.root_helper)
- try:
- # First, try with 'ip'
- device.tuntap.add()
- except RuntimeError, e:
- # Second option: tunctl
- utils.execute(['tunctl', '-b', '-t', device_name],
- self.conf.root_helper)
+ ip = ip_lib.IPWrapper(self.conf.root_helper)
- device.link.set_address(mac_address)
- device.link.set_up()
+ tap_name = device_name.replace(self.DEV_NAME_PREFIX, 'tap')
+ root_veth, dhcp_veth = ip.add_veth(tap_name, device_name)
+ root_veth.link.set_address(mac_address)
+ root_veth.link.set_up()
+ dhcp_veth.link.set_up()
else:
LOG.warn(_("Device %s already exists") % device_name)
except RuntimeError:
LOG.error(_("Failed unplugging interface '%s'") %
device_name)
-
- def get_bridge(self, network_id):
- """Returns the name of the bridge interface."""
- bridge = self.BRIDGE_NAME_PREFIX + network_id[0:11]
- return bridge
# under the License.
from quantum.agent.linux import utils
+from quantum.common import exceptions
-class IPDevice(object):
- def __init__(self, name, root_helper=None):
- self.name = name
+class SubProcessBase(object):
+ def __init__(self, root_helper=None, namespace=None):
self.root_helper = root_helper
- self._commands = {}
+ self.namespace = namespace
- self.link = IpLinkCommand(self)
- self.tuntap = IpTuntapCommand(self)
- self.addr = IpAddrCommand(self)
+ def _run(self, options, command, args):
+ if self.namespace:
+ return self._as_root(options, command, args)
+ else:
+ return self._execute(options, command, args)
- def __eq__(self, other):
- return self.name == other.name
+ def _as_root(self, options, command, args):
+ if not self.root_helper:
+ raise exceptions.SudoRequired()
+ return self._execute(options,
+ command,
+ args,
+ self.root_helper,
+ self.namespace)
@classmethod
- def _execute(cls, options, command, args, root_helper=None):
+ def _execute(cls, options, command, args, root_helper=None,
+ namespace=None):
opt_list = ['-%s' % o for o in options]
- return utils.execute(['ip'] + opt_list + [command] + list(args),
+ if namespace:
+ ip_cmd = ['ip', 'netns', 'exec', namespace, 'ip']
+ else:
+ ip_cmd = ['ip']
+ return utils.execute(ip_cmd + opt_list + [command] + list(args),
root_helper=root_helper)
- @classmethod
- def get_devices(cls):
+
+class IPWrapper(SubProcessBase):
+ def __init__(self, root_helper=None, namespace=None):
+ super(IPWrapper, self).__init__(root_helper=root_helper,
+ namespace=namespace)
+ self.netns = IpNetnsCommand(self)
+
+ def device(self, name):
+ return IPDevice(name, self.root_helper, self.namespace)
+
+ def get_devices(self):
retval = []
- for line in cls._execute('o', 'link', ('list',)).split('\n'):
+ output = self._execute('o', 'link', ('list',),
+ self.root_helper, self.namespace)
+ for line in output.split('\n'):
if '<' not in line:
continue
- index, name, attrs = line.split(':', 2)
- retval.append(IPDevice(name.strip()))
+ tokens = line.split(':', 2)
+ if len(tokens) >= 3:
+ retval.append(IPDevice(tokens[1].strip(),
+ self.root_helper,
+ self.namespace))
return retval
+ def add_tuntap(self, name, mode='tap'):
+ self._as_root('', 'tuntap', ('add', name, 'mode', mode))
+ return IPDevice(name, self.root_helper, self.namespace)
+
+ def add_veth(self, name1, name2):
+ self._as_root('', 'link',
+ ('add', name1, 'type', 'veth', 'peer', 'name', name2))
+
+ return (IPDevice(name1, self.root_helper, self.namespace),
+ IPDevice(name2, self.root_helper, self.namespace))
+
+ def ensure_namespace(self, name):
+ if not self.netns.exists(name):
+ ip = self.netns.add(name)
+ lo = ip.device('lo')
+ lo.link.set_up()
+ else:
+ ip = IP(self.root_helper, name)
+ return ip
+
+ def add_device_to_namespace(self, device):
+ if self.namespace:
+ device.link.set_netns(self.namespace)
+
+ @classmethod
+ def get_namespaces(cls, root_helper):
+ output = cls._execute('netns', ('list',), root_helper=root_helper)
+ return [l.strip() for l in output.split('\n')]
+
+
+class IPDevice(SubProcessBase):
+ def __init__(self, name, root_helper=None, namespace=None):
+ super(IPDevice, self).__init__(root_helper=root_helper,
+ namespace=namespace)
+ self.name = name
+ self.link = IpLinkCommand(self)
+ self.addr = IpAddrCommand(self)
+
+ def __eq__(self, other):
+ return (other is not None and self.name == other.name
+ and self.namespace == other.namespace)
+
+ def __str__(self):
+ return self.name
+
class IpCommandBase(object):
COMMAND = ''
def __init__(self, parent):
self._parent = parent
- @property
- def name(self):
- return self._parent.name
-
def _run(self, *args, **kwargs):
- return self._parent._execute(kwargs.get('options', []),
- self.COMMAND,
- args)
+ return self._parent._run(kwargs.get('options', []), self.COMMAND, args)
def _as_root(self, *args, **kwargs):
- if not self._parent.root_helper:
- raise exceptions.SudoRequired()
- return self._parent._execute(kwargs.get('options', []),
+ return self._parent._as_root(kwargs.get('options', []),
self.COMMAND,
- args,
- self._parent.root_helper)
+ args)
-class IpLinkCommand(IpCommandBase):
+class IpDeviceCommandBase(IpCommandBase):
+ @property
+ def name(self):
+ return self._parent.name
+
+
+class IpLinkCommand(IpDeviceCommandBase):
COMMAND = 'link'
def set_address(self, mac_address):
def set_down(self):
self._as_root('set', self.name, 'down')
+ def set_netns(self, namespace):
+ self._as_root('set', self.name, 'netns', namespace)
+ self._parent.namespace = namespace
+
+ def set_name(self, name):
+ self._as_root('set', self.name, 'name', name)
+ self._parent.name = name
+
def delete(self):
self._as_root('delete', self.name)
return retval
-class IpTuntapCommand(IpCommandBase):
- COMMAND = 'tuntap'
-
- def add(self):
- self._as_root('add', self.name, 'mode', 'tap')
-
-
-class IpAddrCommand(IpCommandBase):
+class IpAddrCommand(IpDeviceCommandBase):
COMMAND = 'addr'
def add(self, ip_version, cidr, broadcast, scope='global'):
return retval
-def device_exists(device_name):
+class IpNetnsCommand(IpCommandBase):
+ COMMAND = 'netns'
+
+ def add(self, name):
+ self._as_root('add', name)
+ return IPWrapper(self._parent.root_helper, name)
+
+ def delete(self, name):
+ self._as_root('delete', name)
+
+ def execute(self, cmds):
+ if not self._parent.root_helper:
+ raise exceptions.SudoRequired()
+ elif not self._parent.namespace:
+ raise Exception(_('No namespace defined for parent'))
+ else:
+ return utils.execute(
+ ['ip', 'netns', 'exec', self._parent.namespace] + list(cmds),
+ root_helper=self._parent.root_helper)
+
+ def exists(self, name):
+ output = self._as_root('list', options='o')
+
+ for line in output.split('\n'):
+ if name == line.strip():
+ return True
+ return False
+
+
+def device_exists(device_name, root_helper=None, namespace=None):
try:
- address = IPDevice(device_name).link.address
+ address = IPDevice(device_name, root_helper, namespace).link.address
except RuntimeError:
return False
-
return True
def test_dhcp_agent_main(self):
with mock.patch('quantum.agent.dhcp_agent.DeviceManager') as dev_mgr:
with mock.patch('quantum.agent.dhcp_agent.DhcpAgent') as dhcp:
- dhcp_agent.main()
- dev_mgr.assert_called_once(mock.ANY, 'sudo')
- dhcp.assert_has_calls([
- mock.call(mock.ANY),
- mock.call().daemon_loop()])
+ with mock.patch('quantum.agent.dhcp_agent.sys') as mock_sys:
+ mock_sys.argv = []
+ dhcp_agent.main()
+ dev_mgr.assert_called_once(mock.ANY, 'sudo')
+ dhcp.assert_has_calls([
+ mock.call(mock.ANY),
+ mock.call().daemon_loop()])
def test_daemon_loop_survives_get_network_state_delta_failure(self):
def stop_loop(*args):
name='filter_by',
side_effect=get_filter_results)
+ self.mock_driver.get_device_name.return_value = 'tap12345678-12'
+
dh = dhcp_agent.DeviceManager(self.conf, mock_db)
dh.setup(fake_network)
mock_driver.DEV_NAME_LEN = (
interface.LinuxInterfaceDriver.DEV_NAME_LEN)
mock_driver.port = fake_port
+ mock_driver.get_device_name.return_value = 'tap12345678-12'
dvr_cls.return_value = mock_driver
dh = dhcp_agent.DeviceManager(self.conf, mock_db)
dvr_cls.assert_called_once_with(self.conf)
mock_driver.assert_has_calls(
- [mock.call.unplug('tap12345678-12')])
+ [mock.call.get_device_name(mock.ANY),
+ mock.call.unplug('tap12345678-12')])
class TestAugmentingWrapper(unittest.TestCase):
pass
+class FakeNetwork:
+ id = '12345678-1234-5678-90ab-ba0987654321'
+
+
class FakeSubnet:
cidr = '192.168.1.1/24'
ip_version = 4
-class FakePort(object):
+class FakePort:
+ id = 'abcdef01-1234-5678-90ab-ba0987654321'
fixed_ips = [FakeAllocation]
device_id = 'cccccccc-cccc-cccc-cccc-cccccccccccc'
+ network = FakeNetwork()
class TestBase(unittest.TestCase):
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()
except RuntimeError, e:
pass
self.ip_dev_p.stop()
+ self.ip_p.stop()
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')]
class TestOVSInterfaceDriver(TestBase):
def test_plug(self, additional_expectation=[]):
- def device_exists(dev, root_helper=None):
+ def device_exists(dev, root_helper=None, namespace=None):
return dev == 'br-int'
vsctl_cmd = ['ovs-vsctl', '--', '--may-exist', 'add-port',
'aa:bb:cc:dd:ee:ff')
execute.assert_called_once_with(vsctl_cmd, 'sudo')
- expected = [mock.call('tap0', 'sudo'),
- mock.call().link.set_address('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)
- expected.append(mock.call().link.set_up())
- self.ip_dev.assert_has_calls(expected)
+ 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().link.set_mtu(9000)])
+ self.test_plug([mock.call().device().link.set_mtu(9000)])
def test_unplug(self):
with mock.patch('quantum.agent.linux.ovs_lib.OVSBridge') as ovs_br:
class TestBridgeInterfaceDriver(TestBase):
- def test_get_bridge(self):
+ def test_get_device_name(self):
br = interface.BridgeInterfaceDriver(self.conf)
- self.assertEqual('brq12345678-11', br.get_bridge('12345678-1122-3344'))
+ device_name = br.get_device_name(FakePort())
+ self.assertEqual('dhcabcdef01-12', device_name)
def test_plug(self):
- def device_exists(device, root_helper=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))
+
expected = [mock.call(c, 'sudo') for c in [
['ip', 'tuntap', 'add', 'tap0', 'mode', 'tap'],
['ip', 'link', 'set', 'tap0', 'address', 'aa:bb:cc:dd:ee:ff'],
br = interface.BridgeInterfaceDriver(self.conf)
br.plug('01234567-1234-1234-99',
'port-1234',
- 'tap0',
+ 'dhc0',
'aa:bb:cc:dd:ee:ff')
- self.ip_dev.assert_has_calls(
- [mock.call('tap0', 'sudo'),
- mock.call().tuntap.add(),
- mock.call().link.set_address('aa:bb:cc:dd:ee:ff'),
- mock.call().link.set_up()])
+ self.ip.assert_has_calls(
+ [mock.call(),
+ mock.call('sudo'),
+ mock.call().add_veth('tap0', 'dhc0')])
+
+ 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
self.ip_dev.assert_has_calls([])
self.assertEquals(log.call_count, 1)
- def test_tunctl_failback(self):
- def device_exists(dev, root_helper=None):
- return dev.startswith('brq')
-
- expected = [mock.call(c, 'sudo') for c in [
- ['ip', 'tuntap', 'add', 'tap0', 'mode', 'tap'],
- ['tunctl', '-b', '-t', 'tap0'],
- ['ip', 'link', 'set', 'tap0', 'address', 'aa:bb:cc:dd:ee:ff'],
- ['ip', 'link', 'set', 'tap0', 'up']]
- ]
-
- self.device_exists.side_effect = device_exists
- self.ip_dev().tuntap.add.side_effect = RuntimeError
- self.ip_dev.reset_calls()
- with mock.patch.object(utils, 'execute') as execute:
- br = interface.BridgeInterfaceDriver(self.conf)
- br.plug('01234567-1234-1234-99',
- 'port-1234',
- 'tap0',
- 'aa:bb:cc:dd:ee:ff')
- execute.assert_called_once_with(['tunctl', '-b', '-t', 'tap0'],
- 'sudo')
- self.ip_dev.assert_has_calls(
- [mock.call('tap0', 'sudo'),
- mock.call().tuntap.add(),
- mock.call().link.set_address('aa:bb:cc:dd:ee:ff'),
- mock.call().link.set_up()])
-
def test_unplug(self):
self.device_exists.return_value = True
with mock.patch('quantum.agent.linux.interface.LOG.debug') as log:
from quantum.agent.linux import ip_lib
from quantum.agent.linux import utils
+from quantum.common import exceptions
+NETNS_SAMPLE = [
+ '12345678-1234-5678-abcd-1234567890ab',
+ 'bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb',
+ 'cccccccc-cccc-cccc-cccc-cccccccccccc']
LINK_SAMPLE = [
'1: lo: <LOOPBACK,UP,LOWER_UP> mtu 16436 qdisc noqueue state UNKNOWN \\'
""")
-class TestIPDevice(unittest.TestCase):
+class TestSubProcessBase(unittest.TestCase):
+ def setUp(self):
+ self.execute_p = mock.patch('quantum.agent.linux.utils.execute')
+ self.execute = self.execute_p.start()
+
+ def tearDown(self):
+ self.execute_p.stop()
+
def test_execute_wrapper(self):
- with mock.patch('quantum.agent.linux.utils.execute') as execute:
- ip_lib.IPDevice._execute('o', 'link', ('list',), 'sudo')
+ ip_lib.SubProcessBase._execute('o', 'link', ('list',), 'sudo')
- execute.assert_called_once_with(['ip', '-o', 'link', 'list'],
- root_helper='sudo')
+ self.execute.assert_called_once_with(['ip', '-o', 'link', 'list'],
+ root_helper='sudo')
def test_execute_wrapper_int_options(self):
- with mock.patch('quantum.agent.linux.utils.execute') as execute:
- ip_lib.IPDevice._execute([4], 'link', ('list',))
+ ip_lib.SubProcessBase._execute([4], 'link', ('list',))
- execute.assert_called_once_with(['ip', '-4', 'link', 'list'],
- root_helper=None)
+ self.execute.assert_called_once_with(['ip', '-4', 'link', 'list'],
+ root_helper=None)
def test_execute_wrapper_no_options(self):
- with mock.patch('quantum.agent.linux.utils.execute') as execute:
- ip_lib.IPDevice._execute([], 'link', ('list',))
+ 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.assertRaises(exceptions.SudoRequired,
+ base._as_root,
+ [], 'link', ('list',))
+
+
+class TestIpWrapper(unittest.TestCase):
+ def setUp(self):
+ self.execute_p = mock.patch.object(ip_lib.IPWrapper, '_execute')
+ self.execute = self.execute_p.start()
- execute.assert_called_once_with(['ip', 'link', 'list'],
- root_helper=None)
+ def tearDown(self):
+ self.execute_p.stop()
def test_get_devices(self):
- with mock.patch.object(ip_lib.IPDevice, '_execute') as _execute:
- _execute.return_value = '\n'.join(LINK_SAMPLE)
- retval = ip_lib.IPDevice.get_devices()
- self.assertEquals(retval,
- [ip_lib.IPDevice('lo'),
- ip_lib.IPDevice('eth0'),
- ip_lib.IPDevice('br-int'),
- ip_lib.IPDevice('gw-ddc717df-49')])
+ self.execute.return_value = '\n'.join(LINK_SAMPLE)
+ retval = ip_lib.IPWrapper('sudo').get_devices()
+ self.assertEquals(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.assertEquals(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.assertEquals(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:
+ ns = ip_lib.IPWrapper('sudo').ensure_namespace('ns')
+ self.execute.assert_has_calls([mock.call('o', 'netns', ('list',),
+ 'sudo', None)])
+ ip_dev.assert_has_calls([mock.call('lo', 'sudo', 'ns'),
+ mock.call().link.set_up()])
+
+ 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, [])
+
- _execute.assert_called_once_with('o', 'link', ('list',))
+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):
- 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.IpCommandBase(self.ip_dev)
+ 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.assertEqual(self.ip_cmd._run('link', 'show'), 'executed')
- self.ip_dev._execute.assert_called_once_with([], 'foo',
- ('link', 'show'))
+ self.ip_cmd._run('link', 'show')
+ self.ip.assert_has_calls([mock.call._run([], 'foo', ('link', 'show'))])
def test_run_with_options(self):
- self.assertEqual(self.ip_cmd._run('link', options='o'), 'executed')
- self.ip_dev._execute.assert_called_once_with('o', 'foo', ('link',))
+ self.ip_cmd._run('link', options='o')
+ self.ip.assert_has_calls([mock.call._run('o', 'foo', ('link', ))])
def test_as_root(self):
- self.assertEqual(self.ip_cmd._as_root('link'), 'executed')
- self.ip_dev._execute.assert_called_once_with([], 'foo',
- ('link',), 'sudo')
+ self.ip_cmd._as_root('link')
+ self.ip.assert_has_calls([mock.call._as_root([], 'foo', ('link', ))])
def test_as_root_with_options(self):
- self.assertEqual(self.ip_cmd._as_root('link', options='o'), 'executed')
- self.ip_dev._execute.assert_called_once_with('o', 'foo',
- ('link',), 'sudo')
+ self.ip_cmd._as_root('link', options='o')
+ self.ip.assert_has_calls([mock.call._as_root('o', 'foo', ('link', ))])
+
+
+class TestIPDeviceCommandBase(unittest.TestCase):
+ def setUp(self):
+ 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')
def _assert_call(self, options, args):
self.parent.assert_has_calls([
- mock.call._execute(options, self.command, args)])
+ mock.call._run(options, self.command, args)])
def _assert_sudo(self, options, args):
self.parent.assert_has_calls([
- mock.call._execute(options, self.command, args, 'sudo')])
+ mock.call._as_root(options, self.command, args)])
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)
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_delete(self):
self.link_cmd.delete()
self._assert_sudo([], ('delete', 'eth0'))
self.parent._execute = mock.Mock(return_value=LINK_SAMPLE[1])
self.assertEqual(self.link_cmd.qlen, 1000)
+ 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,
self._assert_call('o', ('show', 'eth0'))
-class TestIpTuntapCommand(TestIPCmdBase):
- def setUp(self):
- super(TestIpTuntapCommand, self).setUp()
- self.parent.name = 'tap0'
- self.command = 'tuntap'
- self.tuntap_cmd = ip_lib.IpTuntapCommand(self.parent)
-
- def test_add_tap(self):
- self.tuntap_cmd.add()
- self._assert_sudo([], ('add', 'tap0', 'mode', 'tap'))
-
-
class TestIpAddrCommand(TestIPCmdBase):
def setUp(self):
super(TestIpAddrCommand, self).setUp()
dict(ip_version=6, scope='link',
dynamic=False, cidr='fe80::dfcc:aaff:feb9:76ce/64')]
- self.parent._execute = mock.Mock(return_value=ADDR_SAMPLE)
+ self.parent._run = mock.Mock(return_value=ADDR_SAMPLE)
self.assertEquals(self.addr_cmd.list(), expected)
self._assert_call([], ('show', 'tap0'))
dynamic=False, cidr='172.16.77.240/24')]
output = '\n'.join(ADDR_SAMPLE.split('\n')[0:4])
- self.parent._execute = mock.Mock(return_value=output)
+ self.parent._run.return_value = output
self.assertEquals(self.addr_cmd.list('global', filters=['permanent']),
expected)
self._assert_call([], ('show', 'tap0', 'permanent', 'scope', 'global'))
+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'))
+ self.assertEqual(ns.namespace, 'ns')
+
+ def test_delete_namespace(self):
+ self.netns_cmd.delete('ns')
+ self._assert_sudo([], ('delete', 'ns'))
+
+ 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',))
+
+ 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',))
+
+ def test_execute(self):
+ self.parent.namespace = 'ns'
+ with mock.patch('quantum.agent.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')
+
+
class TestDeviceExists(unittest.TestCase):
def test_device_exists(self):
with mock.patch.object(ip_lib.IPDevice, '_execute') as _execute:
Code Review / openstack-build / neutron-build.git / commitdiff