openstack
/
neutron
Code Issues Proposed changes

Remove ip_lib SubProcessBase._execute() as class method 

There is only a single out-of-class caller of ip_lib
SubProcessBase._execute() in the SRIOV mechanism driver
library.  Since it should be calling _as_root(), fix-up
the call and remove the @classmethod.  Cleaned-up tests
in the process.

Change-Id: I79f0d9a35d9cec4771f6990bd3f7a8f427dd311e
This commit is contained in:
Brian Haley 2017-09-29 12:56:48 -04:00
parent 4f627b4e8d
commit 0b1131e85f
5 changed files with 38 additions and 49 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
neutron/agent/linux/ip_lib.py
View File

@ -82,27 +82,23 @@ class SubProcessBase(object):
elif self.force_root:
# Force use of the root helper to ensure that commands
# will execute in dom0 when running under XenServer/XCP.
return self._execute(options, command, args, run_as_root=True,
log_fail_as_error=self.log_fail_as_error)
return self._execute(options, command, args, run_as_root=True)
else:
return self._execute(options, command, args,
log_fail_as_error=self.log_fail_as_error)
return self._execute(options, command, args)
def _as_root(self, options, command, args, use_root_namespace=False):
namespace = self.namespace if not use_root_namespace else None
return self._execute(options, command, args, run_as_root=True,
namespace=namespace,
log_fail_as_error=self.log_fail_as_error)
namespace=namespace)
@classmethod
def _execute(cls, options, command, args, run_as_root=False,
namespace=None, log_fail_as_error=True):
def _execute(self, options, command, args, run_as_root=False,
namespace=None):
opt_list = ['-%s' % o for o in options]
ip_cmd = add_namespace_to_cmd(['ip'], namespace)
cmd = ip_cmd + opt_list + [command] + list(args)
return utils.execute(cmd, run_as_root=run_as_root,
log_fail_as_error=log_fail_as_error)
log_fail_as_error=self.log_fail_as_error)
def set_log_fail_as_error(self, fail_with_error):
self.log_fail_as_error = fail_with_error

2
neutron/plugins/ml2/drivers/mech_sriov/agent/eswitch_manager.py
View File

@ -241,7 +241,7 @@ class EmbSwitch(object):
vf_index = self.pci_slot_map.get(pci_slot)
mac = None
if vf_index is not None:
ls = pci_lib.PciDeviceIPWrapper.link_show()
ls = self.pci_dev_wrapper.link_show()
if PciOsWrapper.is_assigned_vf(self.dev_name, vf_index, ls):
macs = self.pci_dev_wrapper.get_assigned_macs([vf_index])
mac = macs.get(vf_index)

5
neutron/plugins/ml2/drivers/mech_sriov/agent/pci_lib.py
View File

@ -184,10 +184,9 @@ class PciDeviceIPWrapper(ip_lib.IPWrapper):
{'line': vf_line, 'device': self.dev_name})
return vf_details
@classmethod
def link_show(cls):
def link_show(self):
try:
out = cls._execute([], "link", ("show", ), run_as_root=True)
out = self._as_root([], "link", ("show", ))
except Exception as e:
LOG.error("Failed executing ip command: %s", e)
raise exc.IpCommandError(reason=e)

46
neutron/tests/unit/agent/linux/test_ip_lib.py
View File

@ -214,22 +214,24 @@ class TestSubProcessBase(base.BaseTestCase):
self.execute = self.execute_p.start()
def test_execute_wrapper(self):
ip_lib.SubProcessBase._execute(['o'], 'link', ('list',),
run_as_root=True)
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):
ip_lib.SubProcessBase._execute([4], 'link', ('list',))
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):
ip_lib.SubProcessBase._execute([], 'link', ('list',))
base = ip_lib.SubProcessBase()
base._execute([], 'link', ('list',))
self.execute.assert_called_once_with(['ip', 'link', 'list'],
run_as_root=False,
@ -343,16 +345,14 @@ class TestIpWrapper(base.BaseTestCase):
ip_lib.IPWrapper().add_tuntap('tap0')
self.execute.assert_called_once_with([], 'tuntap',
('add', 'tap0', 'mode', 'tap'),
run_as_root=True, namespace=None,
log_fail_as_error=True)
run_as_root=True, namespace=None)
def test_add_veth(self):
ip_lib.IPWrapper().add_veth('tap0', 'tap1')
self.execute.assert_called_once_with([], 'link',
('add', 'tap0', 'type', 'veth',
'peer', 'name', 'tap1'),
run_as_root=True, namespace=None,
log_fail_as_error=True)
run_as_root=True, namespace=None)
def test_add_macvtap(self):
ip_lib.IPWrapper().add_macvtap('macvtap0', 'eth0', 'bridge')
@ -360,15 +360,13 @@ class TestIpWrapper(base.BaseTestCase):
('add', 'link', 'eth0', 'name',
'macvtap0', 'type', 'macvtap',
'mode', 'bridge'),
run_as_root=True, namespace=None,
log_fail_as_error=True)
run_as_root=True, namespace=None)
def test_del_veth(self):
ip_lib.IPWrapper().del_veth('fpr-1234')
self.execute.assert_called_once_with([], 'link',
('del', 'fpr-1234'),
run_as_root=True, namespace=None,
log_fail_as_error=True)
run_as_root=True, namespace=None)
def test_add_veth_with_namespaces(self):
ns2 = 'ns2'
@ -379,16 +377,14 @@ class TestIpWrapper(base.BaseTestCase):
('add', 'tap0', 'type', 'veth',
'peer', 'name', 'tap1',
'netns', ns2),
run_as_root=True, namespace=None,
log_fail_as_error=True)
run_as_root=True, namespace=None)
def test_add_dummy(self):
ip_lib.IPWrapper().add_dummy('dummy0')
self.execute.assert_called_once_with([], 'link',
('add', 'dummy0',
'type', 'dummy'),
run_as_root=True, namespace=None,
log_fail_as_error=True)
run_as_root=True, namespace=None)
def test_get_device(self):
dev = ip_lib.IPWrapper(namespace='ns').device('eth0')
@ -489,8 +485,7 @@ class TestIpWrapper(base.BaseTestCase):
['add', 'link', 'eth0',
'name', 'eth0.1',
'type', 'vlan', 'id', '1'],
run_as_root=True, namespace=None,
log_fail_as_error=True)
run_as_root=True, namespace=None)
def test_add_vxlan_valid_srcport_length(self):
retval = ip_lib.IPWrapper().add_vxlan('vxlan0', 'vni0',
@ -508,8 +503,7 @@ class TestIpWrapper(base.BaseTestCase):
'ttl', 'ttl0', 'tos', 'tos0',
'local', 'local0', 'proxy',
'srcport', '1', '2'],
run_as_root=True, namespace=None,
log_fail_as_error=True)
run_as_root=True, namespace=None)
def test_add_vxlan_invalid_srcport_length(self):
wrapper = ip_lib.IPWrapper()
@ -546,8 +540,7 @@ class TestIpWrapper(base.BaseTestCase):
'local', 'local0', 'proxy',
'srcport', '1', '2',
'dstport', '4789'],
run_as_root=True, namespace=None,
log_fail_as_error=True)
run_as_root=True, namespace=None)
def test_add_device_to_namespace(self):
dev = mock.Mock()
@ -1309,11 +1302,12 @@ class TestIpNetnsCommand(TestIPCmdBase):
class TestDeviceExists(base.BaseTestCase):
def test_device_exists(self):
with mock.patch.object(ip_lib.IPDevice, '_execute') as _execute:
_execute.return_value = LINK_SAMPLE[1]
with mock.patch('neutron.agent.common.utils.execute') as execute:
execute.return_value = LINK_SAMPLE[1]
self.assertTrue(ip_lib.device_exists('eth0'))
_execute.assert_called_once_with(['o'], 'link', ('show', 'eth0'),
log_fail_as_error=False)
execute.assert_called_once_with(
['ip', '-o', 'link', 'show', 'eth0'],
run_as_root=False, log_fail_as_error=False)
def test_device_exists_reset_fail(self):
device = ip_lib.IPDevice('eth0')

18
neutron/tests/unit/plugins/ml2/drivers/mech_sriov/agent/test_pci_lib.py
View File

@ -114,15 +114,15 @@ class TestPciLib(base.BaseTestCase):
True)
def test_set_vf_spoofcheck(self):
with mock.patch.object(self.pci_wrapper, "_execute"):
with mock.patch.object(self.pci_wrapper, "_as_root"):
result = self.pci_wrapper.set_vf_spoofcheck(self.VF_INDEX,
True)
self.assertIsNone(result)
def test_set_vf_spoofcheck_fail(self):
with mock.patch.object(self.pci_wrapper,
"_execute") as mock_exec:
mock_exec.side_effect = Exception()
"_as_root") as mock_as_root:
mock_as_root.side_effect = Exception()
self.assertRaises(exc.IpCommandDeviceError,
self.pci_wrapper.set_vf_spoofcheck,
self.VF_INDEX,
@ -141,7 +141,7 @@ class TestPciLib(base.BaseTestCase):
[], "link", ("set", self.DEV_NAME, "vf",
str(self.VF_INDEX), "rate", '1000'))
else:
with mock.patch.object(self.pci_wrapper, "_execute",
with mock.patch.object(self.pci_wrapper, "_as_root",
side_effect=Exception()):
self.assertRaises(exc.IpCommandDeviceError,
self.pci_wrapper.set_vf_rate,
@ -167,8 +167,8 @@ class TestPciLib(base.BaseTestCase):
def test_set_vf_state_not_supported(self):
with mock.patch.object(self.pci_wrapper,
"_execute") as mock_exec:
mock_exec.side_effect = Exception(
"_as_root") as mock_as_root:
mock_as_root.side_effect = Exception(
pci_lib.PciDeviceIPWrapper.IP_LINK_OP_NOT_SUPPORTED)
self.assertRaises(exc.IpCommandOperationNotSupportedError,
self.pci_wrapper.set_vf_state,
@ -189,7 +189,7 @@ class TestPciLib(base.BaseTestCase):
def test_link_show_command_failed(self):
with mock.patch.object(pci_lib.PciDeviceIPWrapper,
"_execute") as mock_exec:
mock_exec.side_effect = Exception()
"_as_root") as mock_as_root:
mock_as_root.side_effect = Exception()
self.assertRaises(exc.IpCommandError,
pci_lib.PciDeviceIPWrapper.link_show)
self.pci_wrapper.link_show)