openstack
/
neutron
Code Issues Proposed changes

Remove remaining uses of passing root_helper 

This should remove the last bits of passing root_helper around
for execute(). Some tests store and pass the root_helper argument
outside of this, and have been left alone.

Depends-On: I1e3b64e5a1d6cff2aebc638710487bbdbdba61d4
Depends-On: I5d51fa5edc7f427d84d7ae49d3a4c972dca8ac37
Depends-On: I21c6e7795c8b2e8f87bbd2b00a8eb7cb215ece74
Depends-On: I6566e17a1ce07a3bebf2a1a3653ad7e7f397f0b9
Change-Id: I38c901c65fb0c8598589ec56d5afd50d2f882f4f
Partially-Implements: blueprint rootwrap-daemon-mode
This commit is contained in:
Terry Wilson 2015-02-11 00:53:57 -06:00 committed by Henry Gessau
parent 8d5b57c20d
commit 326e25f932
27 changed files with 93 additions and 161 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
neutron/agent/l3/router_info.py
View File

@ -28,7 +28,6 @@ class RouterInfo(object):
agent_conf,
interface_driver,
use_ipv6=False,
root_helper=None,
ns_name=None):
self.router_id = router_id
self.ex_gw_port = None

1
neutron/agent/l3_agent.py
View File

@ -37,7 +37,6 @@ def register_opts(conf):
config.register_interface_driver_opts_helper(conf)
config.register_use_namespaces_opts_helper(conf)
config.register_agent_state_opts_helper(conf)
config.register_root_helper(conf)
conf.register_opts(interface.OPTS)
conf.register_opts(external_process.OPTS)

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

@ -41,7 +41,7 @@ VLAN_INTERFACE_DETAIL = ['vlan protocol 802.1q',
class SubProcessBase(object):
def __init__(self, root_helper=None, namespace=None,
def __init__(self, namespace=None,
log_fail_as_error=True):
self.namespace = namespace
self.log_fail_as_error = log_fail_as_error
@ -88,7 +88,7 @@ class SubProcessBase(object):
class IPWrapper(SubProcessBase):
def __init__(self, root_helper=None, namespace=None):
def __init__(self, namespace=None):
super(IPWrapper, self).__init__(namespace=namespace)
self.netns = IpNetnsCommand(self)
@ -185,7 +185,7 @@ class IPWrapper(SubProcessBase):
return (IPDevice(name, namespace=self.namespace))
@classmethod
def get_namespaces(cls, root_helper=None):
def get_namespaces(cls):
output = cls._execute('', 'netns', ('list',))
return [l.strip() for l in output.split('\n')]
@ -218,7 +218,7 @@ class IpRule(IPWrapper):
class IPDevice(SubProcessBase):
def __init__(self, name, root_helper=None, namespace=None):
def __init__(self, name, namespace=None):
super(IPDevice, self).__init__(namespace=namespace)
self.name = name
self.link = IpLinkCommand(self)
@ -566,7 +566,7 @@ class IpNetnsCommand(IpCommandBase):
return False
def device_exists(device_name, root_helper=None, namespace=None):
def device_exists(device_name, namespace=None):
"""Return True if the device exists in the namespace."""
try:
dev = IPDevice(device_name, namespace=namespace)
@ -577,8 +577,7 @@ def device_exists(device_name, root_helper=None, namespace=None):
return bool(address)
def device_exists_with_ip_mac(device_name, ip_cidr, mac, namespace=None,
root_helper=None):
def device_exists_with_ip_mac(device_name, ip_cidr, mac, namespace=None):
"""Return True if the device with the given IP and MAC addresses
exists in the namespace.
"""
@ -594,7 +593,7 @@ def device_exists_with_ip_mac(device_name, ip_cidr, mac, namespace=None,
return True
def get_routing_table(root_helper=None, namespace=None):
def get_routing_table(namespace=None):
"""Return a list of dictionaries, each representing a route.
The dictionary format is: {'destination': cidr,
@ -624,7 +623,7 @@ def get_routing_table(root_helper=None, namespace=None):
return routes
def ensure_device_is_ready(device_name, root_helper=None, namespace=None):
def ensure_device_is_ready(device_name, namespace=None):
dev = IPDevice(device_name, namespace=namespace)
dev.set_log_fail_as_error(False)
try:
@ -636,7 +635,7 @@ def ensure_device_is_ready(device_name, root_helper=None, namespace=None):
return True
def iproute_arg_supported(command, arg, root_helper=None):
def iproute_arg_supported(command, arg):
command += ['help']
stdout, stderr = utils.execute(command, check_exit_code=False,
return_stderr=True)
@ -656,7 +655,7 @@ def _arping(ns_name, iface_name, address, count):
'ns': ns_name})
def send_gratuitous_arp(ns_name, iface_name, address, count, root_helper=None):
def send_gratuitous_arp(ns_name, iface_name, address, count):
"""Send a gratuitous arp using given namespace, interface, and address"""
def arping():
@ -666,8 +665,7 @@ def send_gratuitous_arp(ns_name, iface_name, address, count, root_helper=None):
eventlet.spawn_n(arping)
def send_garp_for_proxyarp(ns_name, iface_name, address, count,
root_helper=None):
def send_garp_for_proxyarp(ns_name, iface_name, address, count):
"""
Send a gratuitous arp using given namespace, interface, and address

9
neutron/agent/linux/utils.py
View File

@ -61,16 +61,9 @@ def create_process(cmd, run_as_root=False, addl_env=None):
return obj, cmd
# TODO(twilson) Remove root_helper argument in favor of run_as_root
# root_helper= is only intended to handle old-style code that passes the
# root_helper option exclusively, the run_as_root option is for new-style code
# that leaves root_helper handling to the execute() function. The two options
# should not be used in conjunction.
def execute(cmd, root_helper=None, process_input=None, addl_env=None,
def execute(cmd, process_input=None, addl_env=None,
check_exit_code=True, return_stderr=False, log_fail_as_error=True,
extra_ok_codes=None, run_as_root=False):
if root_helper:
run_as_root = True
try:
obj, cmd = create_process(cmd, run_as_root=run_as_root,
addl_env=addl_env)

10
neutron/cmd/netns_cleanup.py
View File

@ -63,7 +63,6 @@ def setup_conf():
conf.register_cli_opts(cli_opts)
agent_config.register_interface_driver_opts_helper(conf)
agent_config.register_use_namespaces_opts_helper(conf)
agent_config.register_root_helper(conf)
conf.register_opts(dhcp_config.DHCP_AGENT_OPTS)
conf.register_opts(dhcp_config.DHCP_OPTS)
conf.register_opts(dhcp_config.DNSMASQ_OPTS)
@ -103,8 +102,7 @@ def eligible_for_deletion(conf, namespace, force=False):
if not re.match(NS_MANGLING_PATTERN, namespace):
return False
root_helper = agent_config.get_root_helper(conf)
ip = ip_lib.IPWrapper(root_helper, namespace)
ip = ip_lib.IPWrapper(namespace=namespace)
return force or ip.namespace_is_empty()
@ -130,8 +128,7 @@ def destroy_namespace(conf, namespace, force=False):
"""
try:
root_helper = agent_config.get_root_helper(conf)
ip = ip_lib.IPWrapper(root_helper, namespace)
ip = ip_lib.IPWrapper(namespace=namespace)
if force:
kill_dhcp(conf, namespace)
@ -167,10 +164,9 @@ def main():
conf()
config.setup_logging()
root_helper = agent_config.get_root_helper(conf)
# Identify namespaces that are candidates for deletion.
candidates = [ns for ns in
ip_lib.IPWrapper.get_namespaces(root_helper)
ip_lib.IPWrapper.get_namespaces()
if eligible_for_deletion(conf, ns, conf.force)]
if candidates:

8
neutron/debug/debug_agent.py
View File

@ -19,7 +19,6 @@ import socket
import netaddr
from oslo_config import cfg
from neutron.agent.common import config
from neutron.agent.linux import dhcp
from neutron.agent.linux import ip_lib
from neutron.agent.linux import utils
@ -45,7 +44,6 @@ class NeutronDebugAgent(object):
def __init__(self, conf, client, driver):
self.conf = conf
self.root_helper = config.get_root_helper(conf)
self.client = client
self.driver = driver
@ -64,7 +62,7 @@ class NeutronDebugAgent(object):
if self.conf.use_namespaces:
namespace = self._get_namespace(port)
if ip_lib.device_exists(interface_name, self.root_helper, namespace):
if ip_lib.device_exists(interface_name, namespace=namespace):
LOG.debug('Reusing existing device: %s.', interface_name)
else:
self.driver.plug(network.id,
@ -111,7 +109,7 @@ class NeutronDebugAgent(object):
bridge = None
if network.external:
bridge = self.conf.external_network_bridge
ip = ip_lib.IPWrapper(self.root_helper)
ip = ip_lib.IPWrapper()
namespace = self._get_namespace(port)
if self.conf.use_namespaces and ip.netns.exists(namespace):
self.driver.unplug(self.driver.get_device_name(port),
@ -138,7 +136,7 @@ class NeutronDebugAgent(object):
def exec_command(self, port_id, command=None):
port = dhcp.DictModel(self.client.show_port(port_id)['port'])
ip = ip_lib.IPWrapper(self.root_helper)
ip = ip_lib.IPWrapper()
namespace = self._get_namespace(port)
if self.conf.use_namespaces:
if not command:

1
neutron/debug/shell.py
View File

@ -73,7 +73,6 @@ class NeutronDebugShell(shell.NeutronShell):
cfg.CONF.register_opts(debug_agent.NeutronDebugAgent.OPTS)
config.register_interface_driver_opts_helper(cfg.CONF)
config.register_use_namespaces_opts_helper(cfg.CONF)
config.register_root_helper(cfg.CONF)
cfg.CONF(['--config-file', self.options.config_file])
config.setup_logging()
driver = importutils.import_object(cfg.CONF.interface_driver, cfg.CONF)

2
neutron/services/metering/agents/metering_agent.py
View File

@ -76,7 +76,6 @@ class MeteringAgent(MeteringPluginRpc, manager.Manager):
def __init__(self, host, conf=None):
self.conf = conf or cfg.CONF
self._load_drivers()
self.root_helper = config.get_root_helper(self.conf)
self.context = context.get_admin_context_without_session()
self.metering_info = {}
self.metering_loop = loopingcall.FixedIntervalLoopingCall(
@ -283,7 +282,6 @@ def main():
conf = cfg.CONF
conf.register_opts(MeteringAgent.Opts)
config.register_agent_state_opts_helper(conf)
config.register_root_helper(conf)
common_config.init(sys.argv[1:])
config.setup_logging()
server = neutron_service.Service.create(

2
neutron/services/metering/drivers/iptables/iptables_driver.py
View File

@ -37,7 +37,6 @@ LABEL = '-l-'
config.register_interface_driver_opts_helper(cfg.CONF)
config.register_use_namespaces_opts_helper(cfg.CONF)
config.register_root_helper(cfg.CONF)
cfg.CONF.register_opts(interface.OPTS)
@ -70,7 +69,6 @@ class RouterWithMetering(object):
self.conf = conf
self.id = router['id']
self.router = router
self.root_helper = config.get_root_helper(self.conf)
self.ns_name = NS_PREFIX + self.id if conf.use_namespaces else None
self.iptables_manager = iptables_manager.IptablesManager(
namespace=self.ns_name,

9
neutron/tests/functional/agent/linux/base.py
View File

@ -13,10 +13,8 @@
# under the License.
import netaddr
from oslo_config import cfg
import testscenarios
from neutron.agent.common import config
from neutron.agent.linux import ip_lib
from neutron.agent.linux import ovs_lib
from neutron.agent.linux import utils
@ -56,7 +54,6 @@ class BaseLinuxTestCase(functional_base.BaseSudoTestCase):
def setUp(self):
super(BaseLinuxTestCase, self).setUp()
config.register_root_helper(cfg.CONF)
def check_command(self, cmd, error_text, skip_msg, run_as_root=False):
try:
@ -67,7 +64,7 @@ class BaseLinuxTestCase(functional_base.BaseSudoTestCase):
raise
def _create_namespace(self):
ip_cmd = ip_lib.IPWrapper(self.root_helper)
ip_cmd = ip_lib.IPWrapper()
name = "func-%s" % uuidutils.generate_uuid()
namespace = ip_cmd.ensure_namespace(name)
self.addCleanup(namespace.netns.delete, namespace.namespace)
@ -92,7 +89,7 @@ class BaseLinuxTestCase(functional_base.BaseSudoTestCase):
continue
def create_veth(self):
ip_wrapper = ip_lib.IPWrapper(self.root_helper)
ip_wrapper = ip_lib.IPWrapper()
name1 = get_rand_veth_name()
name2 = get_rand_veth_name()
self.addCleanup(ip_wrapper.del_veth, name1)
@ -127,7 +124,7 @@ class BaseOVSLinuxTestCase(testscenarios.WithScenarios, BaseLinuxTestCase):
super(BaseOVSLinuxTestCase, self).setUp()
self.config(group='OVS', ovsdb_interface=self.ovsdb_interface)
self.ovs = ovs_lib.BaseOVS()
self.ip = ip_lib.IPWrapper(self.root_helper)
self.ip = ip_lib.IPWrapper()
def create_ovs_bridge(self, br_prefix=BR_PREFIX):
br = self.create_resource(br_prefix, self.ovs.add_bridge)

31
neutron/tests/functional/agent/linux/test_ip_lib.py
View File

@ -68,7 +68,7 @@ class IpLibTestFramework(base.BaseLinuxTestCase):
:param attr: A Device namedtuple
:return: A tuntap ip_lib.IPDevice
"""
ip = ip_lib.IPWrapper(self.root_helper, namespace=attr.namespace)
ip = ip_lib.IPWrapper(namespace=attr.namespace)
ip.netns.add(attr.namespace)
self.addCleanup(ip.netns.delete, attr.namespace)
tap_device = ip.add_tuntap(attr.name)
@ -85,50 +85,39 @@ class IpLibTestCase(IpLibTestFramework):
attr = self.generate_device_details()
self.assertFalse(
ip_lib.device_exists(attr.name, self.root_helper,
attr.namespace))
ip_lib.device_exists(attr.name, namespace=attr.namespace))
device = self.manage_device(attr)
self.assertTrue(
ip_lib.device_exists(device.name, self.root_helper,
attr.namespace))
ip_lib.device_exists(device.name, namespace=attr.namespace))
device.link.delete()
self.assertFalse(
ip_lib.device_exists(attr.name, self.root_helper,
attr.namespace))
ip_lib.device_exists(attr.name, namespace=attr.namespace))
def test_device_exists_with_ip_mac(self):
attr = self.generate_device_details()
device = self.manage_device(attr)
self.assertTrue(
ip_lib.device_exists_with_ip_mac(
*attr, root_helper=self.root_helper))
ip_lib.device_exists_with_ip_mac(*attr))
wrong_ip_cidr = '10.0.0.1/8'
wrong_mac_address = 'aa:aa:aa:aa:aa:aa'
attr = self.generate_device_details(name='wrong_name')
self.assertFalse(
ip_lib.device_exists_with_ip_mac(
*attr, root_helper=self.root_helper))
ip_lib.device_exists_with_ip_mac(*attr))
attr = self.generate_device_details(ip_cidr=wrong_ip_cidr)
self.assertFalse(
ip_lib.device_exists_with_ip_mac(
*attr, root_helper=self.root_helper))
self.assertFalse(ip_lib.device_exists_with_ip_mac(*attr))
attr = self.generate_device_details(mac_address=wrong_mac_address)
self.assertFalse(
ip_lib.device_exists_with_ip_mac(
*attr, root_helper=self.root_helper))
self.assertFalse(ip_lib.device_exists_with_ip_mac(*attr))
attr = self.generate_device_details(namespace='wrong_namespace')
self.assertFalse(
ip_lib.device_exists_with_ip_mac(
*attr, root_helper=self.root_helper))
self.assertFalse(ip_lib.device_exists_with_ip_mac(*attr))
device.link.delete()
@ -147,5 +136,5 @@ class IpLibTestCase(IpLibTestFramework):
'destination': str(
netaddr.IPNetwork(attr.ip_cidr).cidr)}]
routes = ip_lib.get_routing_table(self.root_helper, attr.namespace)
routes = ip_lib.get_routing_table(namespace=attr.namespace)
self.assertEqual(expected_routes, routes)

7
neutron/tests/functional/agent/linux/test_ovsdb_monitor.py
View File

@ -23,6 +23,7 @@ Tests in this module will be skipped unless:
"""
import eventlet
from oslo_config import cfg
from neutron.agent.linux import ovsdb_monitor
from neutron.agent.linux import utils
@ -35,14 +36,14 @@ class BaseMonitorTest(linux_base.BaseOVSLinuxTestCase):
def setUp(self):
super(BaseMonitorTest, self).setUp()
rootwrap_not_configured = (self.root_helper ==
rootwrap_not_configured = (cfg.CONF.AGENT.root_helper ==
functional_base.SUDO_CMD)
if rootwrap_not_configured:
# The monitor tests require a nested invocation that has
# to be emulated by double sudo if rootwrap is not
# configured.
self.root_helper = '%s %s' % (self.root_helper, self.root_helper)
self.config(group='AGENT', root_helper=self.root_helper)
self.config(group='AGENT',
root_helper=" ".join([functional_base.SUDO_CMD] * 2))
self._check_test_requirements()
self.bridge = self.create_ovs_bridge()

30
neutron/tests/functional/agent/test_l3_agent.py
View File

@ -76,7 +76,6 @@ class L3AgentTestFramework(base.BaseOVSLinuxTestCase):
'interface_driver',
'neutron.agent.linux.interface.OVSInterfaceDriver')
conf.set_override('router_delete_namespaces', True)
conf.set_override('root_helper', self.root_helper, group='AGENT')
br_int = self.create_ovs_bridge()
br_ex = self.create_ovs_bridge()
@ -125,7 +124,7 @@ class L3AgentTestFramework(base.BaseOVSLinuxTestCase):
router.router[l3_constants.FLOATINGIP_KEY].append(fip)
def _namespace_exists(self, namespace):
ip = ip_lib.IPWrapper(self.root_helper, namespace)
ip = ip_lib.IPWrapper(namespace=namespace)
return ip.netns.exists(namespace)
def _metadata_proxy_exists(self, conf, router):
@ -138,10 +137,8 @@ class L3AgentTestFramework(base.BaseOVSLinuxTestCase):
def device_exists_with_ip_mac(self, expected_device, name_getter,
namespace):
return ip_lib.device_exists_with_ip_mac(
name_getter(expected_device['id']),
expected_device['ip_cidr'],
expected_device['mac_address'],
namespace, self.root_helper)
name_getter(expected_device['id']), expected_device['ip_cidr'],
expected_device['mac_address'], namespace)
def get_expected_keepalive_configuration(self, router):
ha_confs_path = self.agent.conf.ha_confs_path
@ -248,7 +245,7 @@ class L3AgentTestFramework(base.BaseOVSLinuxTestCase):
device, self.agent.get_internal_device_name, router.ns_name))
def _assert_extra_routes(self, router):
routes = ip_lib.get_routing_table(self.root_helper, router.ns_name)
routes = ip_lib.get_routing_table(namespace=router.ns_name)
routes = [{'nexthop': route['nexthop'],
'destination': route['destination']} for route in routes]
@ -435,8 +432,7 @@ class L3AgentTestCase(L3AgentTestFramework):
external_device_name = self.agent.get_external_device_name(
external_port['id'])
external_device = ip_lib.IPDevice(external_device_name,
self.root_helper,
router.ns_name)
namespace=router.ns_name)
existing_gateway = (
external_device.route.get_gateway().get('gateway'))
expected_gateway = external_port['subnet']['gateway_ip']
@ -449,7 +445,7 @@ class L3AgentTestCase(L3AgentTestFramework):
self.agent.get_external_device_name(external_port['id']),
'%s/32' % fip['floating_ip_address'],
external_port['mac_address'],
router.ns_name, self.root_helper) for fip in floating_ips)
namespace=router.ns_name) for fip in floating_ips)
def _assert_ha_device(self, router):
self.assertTrue(self.device_exists_with_ip_mac(
@ -457,7 +453,7 @@ class L3AgentTestCase(L3AgentTestFramework):
router.get_ha_device_name, router.ns_name))
def _assert_no_ip_addresses_on_interface(self, router, interface):
device = ip_lib.IPDevice(interface, self.root_helper, router.ns_name)
device = ip_lib.IPDevice(interface, namespace=router.ns_name)
self.assertEqual([], device.addr.list())
def test_ha_router_conf_on_restarted_agent(self):
@ -500,8 +496,7 @@ class L3HATestFramework(L3AgentTestFramework):
device_name = router1.get_ha_device_name(
router1.router[l3_constants.HA_INTERFACE_KEY]['id'])
ha_device = ip_lib.IPDevice(device_name, self.root_helper,
router1.ns_name)
ha_device = ip_lib.IPDevice(device_name, namespace=router1.ns_name)
ha_device.link.set_down()
helpers.wait_until_true(lambda: router2.ha_state == 'master')
@ -750,8 +745,7 @@ class TestDvrRouter(L3AgentTestFramework):
external_device_name = self.agent.get_external_device_name(
external_port['id'])
external_device = ip_lib.IPDevice(external_device_name,
self.root_helper,
namespace)
namespace=namespace)
existing_gateway = (
external_device.route.get_gateway().get('gateway'))
expected_gateway = external_port['subnet']['gateway_ip']
@ -780,12 +774,12 @@ class TestDvrRouter(L3AgentTestFramework):
fip_ns.get_ext_device_name(external_gw_port['id']),
external_gw_port['ip_cidr'],
external_gw_port['mac_address'],
fip_ns_name, self.root_helper)
namespace=fip_ns_name)
self.assertTrue(fg_port_created_succesfully)
# Check fpr-router device has been created
device_name = fip_ns.get_int_device_name(router.router_id)
fpr_router_device_created_succesfully = ip_lib.device_exists(
device_name, self.root_helper, fip_ns_name)
device_name, namespace=fip_ns_name)
self.assertTrue(fpr_router_device_created_succesfully)
# In the router namespace
@ -793,4 +787,4 @@ class TestDvrRouter(L3AgentTestFramework):
for fip in floating_ips:
device_name = fip_ns.get_rtr_ext_device_name(router.router_id)
self.assertTrue(ip_lib.device_exists(
device_name, self.root_helper, router.ns_name))
device_name, namespace=router.ns_name))

4
neutron/tests/functional/base.py
View File

@ -52,9 +52,9 @@ class BaseSudoTestCase(base.BaseTestCase):
self.fail_on_missing_deps = (
base.bool_from_env('OS_FAIL_ON_MISSING_DEPS'))
self.root_helper = os.environ.get('OS_ROOTWRAP_CMD', SUDO_CMD)
config.register_root_helper(cfg.CONF)
cfg.CONF.set_override('root_helper', self.root_helper, group='AGENT')
self.config(group='AGENT',
root_helper=os.environ.get('OS_ROOTWRAP_CMD', SUDO_CMD))
def check_sudo_enabled(self):
if not self.sudo_enabled:

4
neutron/tests/tools.py
View File

@ -21,10 +21,10 @@ expected_calls_and_values is a list of (expected_call, return_value):
expected_calls_and_values = [
(mock.call(["ovs-vsctl", self.TO, '--', "--may-exist", "add-port",
self.BR_NAME, pname], root_helper=self.root_helper),
self.BR_NAME, pname]),
None),
(mock.call(["ovs-vsctl", self.TO, "set", "Interface",
pname, "type=gre"], root_helper=self.root_helper),
pname, "type=gre"]),
None),
....
]

2
neutron/tests/unit/agent/linux/test_ovsdb_monitor.py
View File

@ -23,7 +23,6 @@ class TestOvsdbMonitor(base.BaseTestCase):
def setUp(self):
super(TestOvsdbMonitor, self).setUp()
self.root_helper = 'sudo'
self.monitor = ovsdb_monitor.OvsdbMonitor('Interface')
def read_output_queues_and_returns_result(self, output_type, output):
@ -55,7 +54,6 @@ class TestSimpleInterfaceMonitor(base.BaseTestCase):
def setUp(self):
super(TestSimpleInterfaceMonitor, self).setUp()
self.root_helper = 'sudo'
self.monitor = ovsdb_monitor.SimpleInterfaceMonitor()
def test_is_active_is_false_by_default(self):

5
neutron/tests/unit/agent/linux/test_utils.py
View File

@ -24,7 +24,6 @@ _marker = object()
class AgentUtilsExecuteTest(base.BaseTestCase):
def setUp(self):
super(AgentUtilsExecuteTest, self).setUp()
self.root_helper = "echo"
self.test_file = self.get_temp_file_path('test_execute.tmp')
open(self.test_file, 'w').close()
self.process = mock.patch('eventlet.green.subprocess.Popen').start()
@ -40,8 +39,8 @@ class AgentUtilsExecuteTest(base.BaseTestCase):
def test_with_helper(self):
expected = "ls %s\n" % self.test_file
self.mock_popen.return_value = [expected, ""]
result = utils.execute(["ls", self.test_file],
self.root_helper)
self.config(group='AGENT', root_helper='echo')
result = utils.execute(["ls", self.test_file], run_as_root=True)
self.assertEqual(result, expected)
def test_stderr_true(self):

1
neutron/tests/unit/nec/test_config.py
View File

@ -21,7 +21,6 @@ class ConfigurationTest(base.BaseTestCase):
def test_defaults(self):
self.assertEqual('br-int', config.CONF.OVS.integration_bridge)
self.assertEqual(2, config.CONF.AGENT.polling_interval)
self.assertEqual('sudo', config.CONF.AGENT.root_helper)
self.assertEqual('127.0.0.1', config.CONF.OFC.host)
self.assertEqual('8888', config.CONF.OFC.port)

1
neutron/tests/unit/ofagent/ofa_test_base.py
View File

@ -68,5 +68,4 @@ class OFAAgentTestBase(OFATestBase):
cfg.IntOpt('ofp-tcp-listen-port', default=6633,
help='openflow tcp listen port')
])
cfg.CONF.set_override('root_helper', 'fake_helper', group='AGENT')
super(OFATestBase, self).setup_config()

1
neutron/tests/unit/ofagent/test_ofa_neutron_agent.py
View File

@ -91,7 +91,6 @@ class TestOFANeutronAgentBridge(ofa_test_base.OFAAgentTestBase):
def setUp(self):
super(TestOFANeutronAgentBridge, self).setUp()
self.br_name = 'bridge1'
self.root_helper = 'fake_helper'
self.ovs = self.mod_agent.Bridge(
self.br_name, self.ryuapp)

3
neutron/tests/unit/services/metering/drivers/test_iptables_driver.py
View File

@ -75,9 +75,6 @@ class IptablesDriverTestCase(base.BaseTestCase):
self.iptables_cls.return_value = self.iptables_inst
cfg.CONF.set_override('interface_driver',
'neutron.agent.linux.interface.NullDriver')
cfg.CONF.set_override('root_helper',
'fake_sudo',
'AGENT')
self.metering = iptables_driver.IptablesMeteringDriver('metering',
cfg.CONF)

3
neutron/tests/unit/services/metering/test_metering_agent.py
View File

@ -15,7 +15,6 @@
import mock
from oslo_config import cfg
from neutron.agent.common import config
from neutron.openstack.common import uuidutils
from neutron.services.metering.agents import metering_agent
from neutron.tests import base
@ -43,7 +42,6 @@ class TestMeteringOperations(base.BaseTestCase,
def setUp(self):
super(TestMeteringOperations, self).setUp()
cfg.CONF.register_opts(metering_agent.MeteringAgent.Opts)
config.register_root_helper(cfg.CONF)
self.noop_driver = ('neutron.services.metering.drivers.noop.'
'noop_driver.NoopMeteringDriver')
@ -130,7 +128,6 @@ class TestMeteringDriver(base.BaseTestCase):
def setUp(self):
super(TestMeteringDriver, self).setUp()
cfg.CONF.register_opts(metering_agent.MeteringAgent.Opts)
config.register_root_helper(cfg.CONF)
self.noop_driver = ('neutron.services.metering.drivers.noop.'
'noop_driver.NoopMeteringDriver')

1
neutron/tests/unit/test_debug_commands.py
View File

@ -39,7 +39,6 @@ class TestDebugCommands(base.BaseTestCase):
common_config.init([])
config.register_interface_driver_opts_helper(cfg.CONF)
config.register_use_namespaces_opts_helper(cfg.CONF)
config.register_root_helper(cfg.CONF)
cfg.CONF.set_override('use_namespaces', True)
device_exists_p = mock.patch(

1
neutron/tests/unit/test_iptables_manager.py
View File

@ -230,7 +230,6 @@ class IptablesManagerStateFulTestCase(base.BaseTestCase):
super(IptablesManagerStateFulTestCase, self).setUp()
cfg.CONF.register_opts(a_cfg.IPTABLES_OPTS, 'AGENT')
cfg.CONF.set_override('comment_iptables_rules', False, 'AGENT')
self.root_helper = 'sudo'
self.iptables = iptables_manager.IptablesManager()
self.execute = mock.patch.object(self.iptables, "execute").start()

81
neutron/tests/unit/test_linux_ip_lib.py
View File

@ -185,14 +185,14 @@ class TestSubProcessBase(base.BaseTestCase):
log_fail_as_error=True)
def test_run_no_namespace(self):
base = ip_lib.SubProcessBase('sudo')
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('sudo', 'ns')
base = ip_lib.SubProcessBase(namespace='ns')
base._run([], 'link', ('list',))
self.execute.assert_called_once_with(['ip', 'netns', 'exec', 'ns',
'ip', 'link', 'list'],
@ -200,7 +200,7 @@ class TestSubProcessBase(base.BaseTestCase):
log_fail_as_error=True)
def test_as_root_namespace(self):
base = ip_lib.SubProcessBase('sudo', 'ns')
base = ip_lib.SubProcessBase(namespace='ns')
base._as_root([], 'link', ('list',))
self.execute.assert_called_once_with(['ip', 'netns', 'exec', 'ns',
'ip', 'link', 'list'],
@ -216,7 +216,7 @@ class TestIpWrapper(base.BaseTestCase):
def test_get_devices(self):
self.execute.return_value = '\n'.join(LINK_SAMPLE)
retval = ip_lib.IPWrapper('sudo').get_devices()
retval = ip_lib.IPWrapper().get_devices()
self.assertEqual(retval,
[ip_lib.IPDevice('lo'),
ip_lib.IPDevice('eth0'),
@ -238,7 +238,7 @@ class TestIpWrapper(base.BaseTestCase):
def test_get_devices_malformed_line(self):
self.execute.return_value = '\n'.join(LINK_SAMPLE + ['gibberish'])
retval = ip_lib.IPWrapper('sudo').get_devices()
retval = ip_lib.IPWrapper().get_devices()
self.assertEqual(retval,
[ip_lib.IPDevice('lo'),
ip_lib.IPDevice('eth0'),
@ -260,7 +260,7 @@ class TestIpWrapper(base.BaseTestCase):
def test_get_namespaces(self):
self.execute.return_value = '\n'.join(NETNS_SAMPLE)
retval = ip_lib.IPWrapper.get_namespaces('sudo')
retval = ip_lib.IPWrapper.get_namespaces()
self.assertEqual(retval,
['12345678-1234-5678-abcd-1234567890ab',
'bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb',
@ -269,14 +269,14 @@ class TestIpWrapper(base.BaseTestCase):
self.execute.assert_called_once_with('', 'netns', ('list',))
def test_add_tuntap(self):
ip_lib.IPWrapper('sudo').add_tuntap('tap0')
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)
def test_add_veth(self):
ip_lib.IPWrapper('sudo').add_veth('tap0', 'tap1')
ip_lib.IPWrapper().add_veth('tap0', 'tap1')
self.execute.assert_called_once_with('', 'link',
('add', 'tap0', 'type', 'veth',
'peer', 'name', 'tap1'),
@ -284,7 +284,7 @@ class TestIpWrapper(base.BaseTestCase):
log_fail_as_error=True)
def test_del_veth(self):
ip_lib.IPWrapper('sudo').del_veth('fpr-1234')
ip_lib.IPWrapper().del_veth('fpr-1234')
self.execute.assert_called_once_with('', 'link',
('del', 'fpr-1234'),
run_as_root=True, namespace=None,
@ -293,7 +293,7 @@ class TestIpWrapper(base.BaseTestCase):
def test_add_veth_with_namespaces(self):
ns2 = 'ns2'
with mock.patch.object(ip_lib.IPWrapper, 'ensure_namespace') as en:
ip_lib.IPWrapper('sudo').add_veth('tap0', 'tap1', namespace2=ns2)
ip_lib.IPWrapper().add_veth('tap0', 'tap1', namespace2=ns2)
en.assert_has_calls([mock.call(ns2)])
self.execute.assert_called_once_with('', 'link',
('add', 'tap0', 'type', 'veth',
@ -303,13 +303,13 @@ class TestIpWrapper(base.BaseTestCase):
log_fail_as_error=True)
def test_get_device(self):
dev = ip_lib.IPWrapper('sudo', 'ns').device('eth0')
dev = ip_lib.IPWrapper(namespace='ns').device('eth0')
self.assertEqual(dev.namespace, 'ns')
self.assertEqual(dev.name, 'eth0')
def test_ensure_namespace(self):
with mock.patch.object(ip_lib, 'IPDevice') as ip_dev:
ip = ip_lib.IPWrapper('sudo')
ip = ip_lib.IPWrapper()
with mock.patch.object(ip.netns, 'exists') as ns_exists:
with mock.patch('neutron.agent.linux.utils.execute'):
ns_exists.return_value = False
@ -324,12 +324,12 @@ class TestIpWrapper(base.BaseTestCase):
def test_ensure_namespace_existing(self):
with mock.patch.object(ip_lib, 'IpNetnsCommand') as ip_ns_cmd:
ip_ns_cmd.exists.return_value = True
ns = ip_lib.IPWrapper('sudo').ensure_namespace('ns')
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('sudo', 'ns')
ip = ip_lib.IPWrapper(namespace='ns')
with mock.patch.object(ip, 'get_devices') as get_devices:
get_devices.return_value = []
@ -337,7 +337,7 @@ class TestIpWrapper(base.BaseTestCase):
get_devices.assert_called_once_with(exclude_loopback=True)
def test_namespace_is_empty(self):
ip = ip_lib.IPWrapper('sudo', 'ns')
ip = ip_lib.IPWrapper(namespace='ns')
with mock.patch.object(ip, 'get_devices') as get_devices:
get_devices.return_value = [mock.Mock()]
@ -347,7 +347,7 @@ class TestIpWrapper(base.BaseTestCase):
def test_garbage_collect_namespace_does_not_exist(self):
with mock.patch.object(ip_lib, 'IpNetnsCommand') as ip_ns_cmd_cls:
ip_ns_cmd_cls.return_value.exists.return_value = False
ip = ip_lib.IPWrapper('sudo', 'ns')
ip = ip_lib.IPWrapper(namespace='ns')
with mock.patch.object(ip, 'namespace_is_empty') as mock_is_empty:
self.assertFalse(ip.garbage_collect_namespace())
@ -360,7 +360,7 @@ class TestIpWrapper(base.BaseTestCase):
with mock.patch.object(ip_lib, 'IpNetnsCommand') as ip_ns_cmd_cls:
ip_ns_cmd_cls.return_value.exists.return_value = True
ip = ip_lib.IPWrapper('sudo', 'ns')
ip = ip_lib.IPWrapper(namespace='ns')
with mock.patch.object(ip, 'namespace_is_empty') as mock_is_empty:
mock_is_empty.return_value = True
@ -380,7 +380,7 @@ class TestIpWrapper(base.BaseTestCase):
with mock.patch.object(ip_lib, 'IpNetnsCommand') as ip_ns_cmd_cls:
ip_ns_cmd_cls.return_value.exists.return_value = True
ip = ip_lib.IPWrapper('sudo', 'ns')
ip = ip_lib.IPWrapper(namespace='ns')
with mock.patch.object(ip, 'namespace_is_empty') as mock_is_empty:
mock_is_empty.return_value = False
@ -395,12 +395,12 @@ class TestIpWrapper(base.BaseTestCase):
ip_ns_cmd_cls.mock_calls)
def test_add_vxlan_valid_port_length(self):
retval = ip_lib.IPWrapper('sudo').add_vxlan('vxlan0', 'vni0',
group='group0',
dev='dev0', ttl='ttl0',
tos='tos0',
local='local0', proxy=True,
port=('1', '2'))
retval = ip_lib.IPWrapper().add_vxlan('vxlan0', 'vni0',
group='group0',
dev='dev0', ttl='ttl0',
tos='tos0',
local='local0', proxy=True,
port=('1', '2'))
self.assertIsInstance(retval, ip_lib.IPDevice)
self.assertEqual(retval.name, 'vxlan0')
self.execute.assert_called_once_with('', 'link',
@ -414,7 +414,7 @@ class TestIpWrapper(base.BaseTestCase):
log_fail_as_error=True)
def test_add_vxlan_invalid_port_length(self):
wrapper = ip_lib.IPWrapper('sudo')
wrapper = ip_lib.IPWrapper()
self.assertRaises(exceptions.NetworkVxlanPortRangeError,
wrapper.add_vxlan, 'vxlan0', 'vni0', group='group0',
dev='dev0', ttl='ttl0', tos='tos0',
@ -423,12 +423,12 @@ class TestIpWrapper(base.BaseTestCase):
def test_add_device_to_namespace(self):
dev = mock.Mock()
ip_lib.IPWrapper('sudo', 'ns').add_device_to_namespace(dev)
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('sudo').add_device_to_namespace(dev)
ip_lib.IPWrapper().add_device_to_namespace(dev)
self.assertEqual(dev.mock_calls, [])
@ -440,7 +440,7 @@ class TestIpRule(base.BaseTestCase):
def _test_add_rule(self, ip, table, priority):
ip_version = netaddr.IPNetwork(ip).version
ip_lib.IpRule('sudo').add(ip, table, priority)
ip_lib.IpRule().add(ip, table, priority)
call_1 = mock.call([ip_version], 'rule', ['show'],
run_as_root=True, namespace=None,
log_fail_as_error=True)
@ -457,7 +457,7 @@ class TestIpRule(base.BaseTestCase):
def _test_add_rule_exists(self, ip, table, priority, output):
self.execute.return_value = output
ip_version = netaddr.IPNetwork(ip).version
ip_lib.IpRule('sudo').add(ip, table, priority)
ip_lib.IpRule().add(ip, table, priority)
self.execute.assert_called_once_with([ip_version], 'rule',
['show'],
run_as_root=True, namespace=None,
@ -465,7 +465,7 @@ class TestIpRule(base.BaseTestCase):
def _test_delete_rule(self, ip, table, priority):
ip_version = netaddr.IPNetwork(ip).version
ip_lib.IpRule('sudo').delete(ip, table, priority)
ip_lib.IpRule().delete(ip, table, priority)
self.execute.assert_called_once_with([ip_version], 'rule',
('del', 'table', table,
'priority', priority),
@ -508,12 +508,12 @@ class TestIPDevice(base.BaseTestCase):
self.assertEqual(dev1, dev2)
def test_eq_diff_namespace(self):
dev1 = ip_lib.IPDevice('tap0', 'sudo', 'ns1')
dev2 = ip_lib.IPDevice('tap0', 'sudo', 'ns2')
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', 'sudo', 'ns1')
dev1 = ip_lib.IPDevice('tap0', namespace='ns1')
self.assertIsNotNone(dev1)
def test_str(self):
@ -524,7 +524,6 @@ class TestIPCommandBase(base.BaseTestCase):
def setUp(self):
super(TestIPCommandBase, self).setUp()
self.ip = mock.Mock()
self.ip.root_helper = 'sudo'
self.ip.namespace = 'namespace'
self.ip_cmd = ip_lib.IpCommandBase(self.ip)
self.ip_cmd.COMMAND = 'foo'
@ -553,7 +552,6 @@ class TestIPDeviceCommandBase(base.BaseTestCase):
super(TestIPDeviceCommandBase, self).setUp()
self.ip_dev = mock.Mock()
self.ip_dev.name = 'eth0'
self.ip_dev.root_helper = 'sudo'
self.ip_dev._execute = mock.Mock(return_value='executed')
self.ip_cmd = ip_lib.IpDeviceCommandBase(self.ip_dev)
self.ip_cmd.COMMAND = 'foo'
@ -567,7 +565,6 @@ class TestIPCmdBase(base.BaseTestCase):
super(TestIPCmdBase, self).setUp()
self.parent = mock.Mock()
self.parent.name = 'eth0'
self.parent.root_helper = 'sudo'
def _assert_call(self, options, args):
self.parent.assert_has_calls([
@ -895,7 +892,6 @@ class TestIpNetnsCommand(TestIPCmdBase):
def test_execute_nosudo_with_no_namespace(self):
with mock.patch('neutron.agent.linux.utils.execute') as execute:
self.parent.namespace = None
self.parent.root_helper = None
self.netns_cmd.execute(['test'])
execute.assert_called_once_with(['test'],
check_exit_code=True,
@ -952,14 +948,12 @@ class TestArpPing(TestIPCmdBase):
function(mock.sentinel.ns_name,
mock.sentinel.iface_name,
address,
mock.sentinel.count,
mock.sentinel.root_helper)
mock.sentinel.count)
self.assertTrue(spawn_n.called)
mIPWrapper.assert_called_once_with(namespace=mock.sentinel.ns_name)
ip_wrapper = mIPWrapper(mock.sentinel.root_helper,
mock.sentinel.ns_name)
ip_wrapper = mIPWrapper(namespace=mock.sentinel.ns_name)
# Just test that arping is called with the right arguments
arping_cmd = ['arping', '-A',
@ -980,11 +974,9 @@ class TestArpPing(TestIPCmdBase):
@mock.patch('eventlet.spawn_n')
def test_send_garp_for_proxy_arp(self, spawn_n, mIPWrapper, mIPDevice):
addr = '20.0.0.1'
ip_wrapper = mIPWrapper(mock.sentinel.root_helper,
mock.sentinel.ns_name)
ip_wrapper = mIPWrapper(namespace=mock.sentinel.ns_name)
mIPWrapper.reset_mock()
device = mIPDevice(mock.sentinel.iface_name,
mock.sentinel.root_helper,
namespace=mock.sentinel.ns_name)
mIPDevice.reset_mock()
@ -1003,7 +995,6 @@ class TestArpPing(TestIPCmdBase):
# Test that the address was removed after arping
device = mIPDevice(mock.sentinel.iface_name,
mock.sentinel.root_helper,
namespace=mock.sentinel.ns_name)
device.addr.delete.assert_called_once_with(4, addr + '/32')

10
neutron/tests/unit/test_netns_cleanup.py
View File

@ -71,7 +71,7 @@ class TestNetnsCleanup(base.BaseTestCase):
self.assertEqual(util.eligible_for_deletion(conf, ns, force),
expected)
expected_calls = [mock.call(conf.AGENT.root_helper, ns)]
expected_calls = [mock.call(namespace=ns)]
if not force:
expected_calls.append(mock.call().namespace_is_empty())
ip_wrap.assert_has_calls(expected_calls)
@ -141,7 +141,6 @@ class TestNetnsCleanup(base.BaseTestCase):
def _test_destroy_namespace_helper(self, force, num_devices):
ns = 'qrouter-6e322ac7-ab50-4f53-9cdc-d1d3c1164b6d'
conf = mock.Mock()
# conf.AGENT.root_helper = 'sudo'
lo_device = mock.Mock()
lo_device.name = 'lo'
@ -162,7 +161,7 @@ class TestNetnsCleanup(base.BaseTestCase):
with mock.patch.object(util, 'kill_dhcp') as kill_dhcp:
util.destroy_namespace(conf, ns, force)
expected = [mock.call(conf.AGENT.root_helper, ns)]
expected = [mock.call(namespace=ns)]
if force:
expected.extend([
@ -188,7 +187,6 @@ class TestNetnsCleanup(base.BaseTestCase):
def test_destroy_namespace_exception(self):
ns = 'qrouter-6e322ac7-ab50-4f53-9cdc-d1d3c1164b6d'
conf = mock.Mock()
conf.AGENT.root_helper = 'sudo'
with mock.patch('neutron.agent.linux.ip_lib.IPWrapper') as ip_wrap:
ip_wrap.side_effect = Exception()
util.destroy_namespace(conf, ns)
@ -221,7 +219,7 @@ class TestNetnsCleanup(base.BaseTestCase):
mock.call(conf, 'ns2', False)])
ip_wrap.assert_has_calls(
[mock.call.get_namespaces(conf.AGENT.root_helper)])
[mock.call.get_namespaces()])
time_sleep.assert_called_once_with(2)
@ -245,7 +243,7 @@ class TestNetnsCleanup(base.BaseTestCase):
util.main()
ip_wrap.assert_has_calls(
[mock.call.get_namespaces(conf.AGENT.root_helper)])
[mock.call.get_namespaces()])
mocks['eligible_for_deletion'].assert_has_calls(
[mock.call(conf, 'ns1', False),

2
neutron/tests/unit/test_ovs_cleanup.py
View File

@ -35,13 +35,11 @@ class TestOVSCleanup(base.BaseTestCase):
self.assertEqual(conf.external_network_bridge, 'br-ex')
self.assertEqual(conf.ovs_integration_bridge, 'br-int')
self.assertFalse(conf.ovs_all_ports)
self.assertEqual(conf.AGENT.root_helper, 'sudo')
def test_main(self):
bridges = ['br-int', 'br-ex']
ports = ['p1', 'p2', 'p3']
conf = mock.Mock()
conf.AGENT.root_helper = 'dummy_sudo'
conf.ovs_all_ports = False
conf.ovs_integration_bridge = 'br-int'
conf.external_network_bridge = 'br-ex'