# Allow overlapping IP (Must have kernel build with CONFIG_NET_NS=y and
# iproute2 package that supports namespaces).
# use_namespaces = True
-
-# Use "sudo quantum-rootwrap /etc/quantum/rootwrap.conf" to use the real
-# root filter facility.
-# Change to "sudo" to skip the filtering and just run the comand directly
-root_helper = sudo
# LinuxBridge
#interface_driver = quantum.agent.linux.interface.BridgeInterfaceDriver
-# Use "sudo quantum-rootwrap /etc/quantum/rootwrap.conf" to use the real
-# root filter facility.
-# Change to "sudo" to skip the filtering and just run the comand directly
-root_helper = sudo
-
# Allow overlapping IP (Must have kernel build with CONFIG_NET_NS=y and
# iproute2 package that supports namespaces).
# use_namespaces = True
admin_user = %SERVICE_USER%
admin_password = %SERVICE_PASSWORD%
-# Use "sudo quantum-rootwrap /etc/quantum/rootwrap.conf" to use the real
-# root filter facility.
-# Change to "sudo" to skip the filtering and just run the comand directly
-root_helper = sudo
-
# Where to store metadata state files. This directory must be writable by the
# user executing the agent.
# state_path = /var/lib/quantum
[SECURITYGROUP]
# If set to true this allows quantum to receive proxied security group calls from nova
# proxy_mode = False
+
+[AGENT]
+# Use "sudo quantum-rootwrap /etc/quantum/rootwrap.conf" to use the real
+# root filter facility.
+# Change to "sudo" to skip the filtering and just run the comand directly
+# root_helper = sudo
[AGENT]
# Agent's polling interval in seconds
polling_interval = 2
-# Use "sudo quantum-rootwrap /etc/quantum/rootwrap.conf" to use the real
-# root filter facility.
-# Change to "sudo" to skip the filtering and just run the comand directly
-root_helper = "sudo"
[AGENT]
# Agent's polling interval in seconds
polling_interval = 2
-# Use "sudo quantum-rootwrap /etc/quantum/rootwrap.conf" to use the real
-# root filter facility.
-# Change to "sudo" to skip the filtering and just run the comand directly
-root_helper = sudo
#-----------------------------------------------------------------------------
# Sample Configurations.
# integration_bridge = br-int
# bridge_mappings = default:br-eth1
# [AGENT]
-# root_helper = sudo
# Add the following setting, if you want to log to a file
#
# 2. With tunneling.
# integration_bridge = br-int
# tunnel_bridge = br-tun
# local_ip = 10.0.0.3
-# [AGENT]
-# root_helper = sudo
# ovsdb_ip =
# ovsdb_interface =
ovsdb_interface = eth0
-
-[AGENT]
-# Use "sudo quantum-rootwrap /etc/quantum/rootwrap.conf" to use the real
-# root filter facility.
-# Change to "sudo" to skip the filtering and just run the comand directly
-root_helper = sudo
from quantum.common import config
from quantum.openstack.common import cfg
+from quantum.openstack.common import log as logging
+
+
+LOG = logging.getLogger(__name__)
+
+
+ROOT_HELPER_OPTS = [
+ cfg.StrOpt('root_helper', default='sudo',
+ help=_('Root helper application.')),
+]
+
+
+def register_root_helper(conf):
+ # The first call is to ensure backward compatibility
+ conf.register_opts(ROOT_HELPER_OPTS)
+ conf.register_opts(ROOT_HELPER_OPTS, 'AGENT')
+
+
+def get_root_helper(conf):
+ root_helper = conf.AGENT.root_helper
+ if root_helper is not 'sudo':
+ return root_helper
+
+ root_helper = conf.root_helper
+ if root_helper is not 'sudo':
+ LOG.deprecated(_('DEFAULT.root_helper is deprecated!'))
+ return root_helper
+
+ return 'sudo'
def setup_conf():
class DhcpAgent(object):
OPTS = [
- cfg.StrOpt('root_helper', default='sudo',
- help=_("Root helper application.")),
cfg.IntOpt('resync_interval', default=30,
help=_("Interval to resync.")),
cfg.StrOpt('dhcp_driver',
self.needs_resync = False
self.conf = conf
self.cache = NetworkCache()
+ self.root_helper = config.get_root_helper(conf)
self.dhcp_driver_cls = importutils.import_class(conf.dhcp_driver)
ctx = context.get_admin_context_without_session()
# the base models.
driver = self.dhcp_driver_cls(self.conf,
network,
- self.conf.root_helper,
+ self.root_helper,
self.device_manager,
namespace)
getattr(driver, action)()
def __init__(self, conf, plugin):
self.conf = conf
+ self.root_helper = config.get_root_helper(conf)
self.plugin = plugin
if not conf.interface_driver:
LOG.error(_('You must specify an interface driver'))
namespace = None
if ip_lib.device_exists(interface_name,
- self.conf.root_helper,
+ self.root_helper,
namespace):
if not reuse_existing:
raise exceptions.PreexistingDeviceFailure(
# ensure that the dhcp interface is first in the list
if namespace is None:
- device = ip_lib.IPDevice(interface_name, self.conf.root_helper)
+ device = ip_lib.IPDevice(interface_name,
+ self.root_helper)
device.route.pullup_route(interface_name)
return interface_name
def main():
eventlet.monkey_patch()
cfg.CONF.register_opts(DhcpAgent.OPTS)
+ config.register_root_helper(cfg.CONF)
cfg.CONF.register_opts(DeviceManager.OPTS)
cfg.CONF.register_opts(DhcpLeaseRelay.OPTS)
cfg.CONF.register_opts(dhcp.OPTS)
class L3NATAgent(manager.Manager):
OPTS = [
- cfg.StrOpt('root_helper', default='sudo',
- help=_("Root helper application.")),
cfg.StrOpt('external_network_bridge', default='br-ex',
help=_("Name of bridge used for external network "
"traffic.")),
self.conf = conf
else:
self.conf = cfg.CONF
+ self.root_helper = config.get_root_helper(self.conf)
self.router_info = {}
if not self.conf.interface_driver:
"""Destroy all router namespaces on the host to eliminate
all stale linux devices, iptables rules, and namespaces.
"""
- root_ip = ip_lib.IPWrapper(self.conf.root_helper)
- for ns in root_ip.get_namespaces(self.conf.root_helper):
+ root_ip = ip_lib.IPWrapper(self.root_helper)
+ for ns in root_ip.get_namespaces(self.root_helper):
if ns.startswith(NS_PREFIX):
try:
self._destroy_router_namespace(ns)
LOG.exception(_("Failed deleting namespace '%s'"), ns)
def _destroy_router_namespace(self, namespace):
- ns_ip = ip_lib.IPWrapper(self.conf.root_helper,
- namespace=namespace)
+ ns_ip = ip_lib.IPWrapper(self.root_helper, namespace=namespace)
for d in ns_ip.get_devices(exclude_loopback=True):
if d.name.startswith(INTERNAL_DEV_PREFIX):
# device is on default bridge
#(TODO) Address the failure for the deletion of the namespace
def _create_router_namespace(self, ri):
- ip_wrapper_root = ip_lib.IPWrapper(self.conf.root_helper)
+ ip_wrapper_root = ip_lib.IPWrapper(self.root_helper)
ip_wrapper = ip_wrapper_root.ensure_namespace(ri.ns_name())
ip_wrapper.netns.execute(['sysctl', '-w', 'net.ipv4.ip_forward=1'])
raise
def _router_added(self, router_id, router=None):
- ri = RouterInfo(router_id, self.conf.root_helper,
+ ri = RouterInfo(router_id, self.root_helper,
self.conf.use_namespaces, router)
self.router_info[router_id] = ri
if self.conf.use_namespaces:
pm = external_process.ProcessManager(
self.conf,
router_info.router_id,
- self.conf.root_helper,
+ self.root_helper,
router_info.ns_name())
pm.enable(callback)
pm = external_process.ProcessManager(
self.conf,
router_info.router_id,
- self.conf.root_helper,
+ self.root_helper,
router_info.ns_name())
pm.disable()
ip_address]
try:
if self.conf.use_namespaces:
- ip_wrapper = ip_lib.IPWrapper(self.conf.root_helper,
+ ip_wrapper = ip_lib.IPWrapper(self.root_helper,
namespace=ri.ns_name())
ip_wrapper.netns.execute(arping_cmd, check_exit_code=True)
else:
utils.execute(arping_cmd, check_exit_code=True,
- root_helper=self.conf.root_helper)
+ root_helper=self.root_helper)
except Exception as e:
LOG.error(_("Failed sending gratuitous ARP: %s"), str(e))
interface_name = self.get_external_device_name(ex_gw_port['id'])
ex_gw_ip = ex_gw_port['fixed_ips'][0]['ip_address']
if not ip_lib.device_exists(interface_name,
- root_helper=self.conf.root_helper,
+ root_helper=self.root_helper,
namespace=ri.ns_name()):
self.driver.plug(ex_gw_port['network_id'],
ex_gw_port['id'], interface_name,
if ex_gw_port['subnet']['gateway_ip']:
cmd = ['route', 'add', 'default', 'gw', gw_ip]
if self.conf.use_namespaces:
- ip_wrapper = ip_lib.IPWrapper(self.conf.root_helper,
+ ip_wrapper = ip_lib.IPWrapper(self.root_helper,
namespace=ri.ns_name())
ip_wrapper.netns.execute(cmd, check_exit_code=False)
else:
utils.execute(cmd, check_exit_code=False,
- root_helper=self.conf.root_helper)
+ root_helper=self.root_helper)
for (c, r) in self.external_gateway_nat_rules(ex_gw_ip,
internal_cidrs,
interface_name = self.get_external_device_name(ex_gw_port['id'])
if ip_lib.device_exists(interface_name,
- root_helper=self.conf.root_helper,
+ root_helper=self.root_helper,
namespace=ri.ns_name()):
self.driver.unplug(interface_name,
bridge=self.conf.external_network_bridge,
internal_cidr, mac_address):
interface_name = self.get_internal_device_name(port_id)
if not ip_lib.device_exists(interface_name,
- root_helper=self.conf.root_helper,
+ root_helper=self.root_helper,
namespace=ri.ns_name()):
self.driver.plug(network_id, port_id, interface_name, mac_address,
namespace=ri.ns_name(),
def internal_network_removed(self, ri, ex_gw_port, port_id, internal_cidr):
interface_name = self.get_internal_device_name(port_id)
if ip_lib.device_exists(interface_name,
- root_helper=self.conf.root_helper,
+ root_helper=self.root_helper,
namespace=ri.ns_name()):
self.driver.unplug(interface_name, namespace=ri.ns_name(),
prefix=INTERNAL_DEV_PREFIX)
def floating_ip_added(self, ri, ex_gw_port, floating_ip, fixed_ip):
ip_cidr = str(floating_ip) + '/32'
interface_name = self.get_external_device_name(ex_gw_port['id'])
- device = ip_lib.IPDevice(interface_name, self.conf.root_helper,
+ device = ip_lib.IPDevice(interface_name, self.root_helper,
namespace=ri.ns_name())
if ip_cidr not in [addr['cidr'] for addr in device.addr.list()]:
net = netaddr.IPNetwork(ip_cidr)
interface_name = self.get_external_device_name(ex_gw_port['id'])
- device = ip_lib.IPDevice(interface_name, self.conf.root_helper,
+ device = ip_lib.IPDevice(interface_name, self.root_helper,
namespace=ri.ns_name())
device.addr.delete(net.version, ip_cidr)
eventlet.monkey_patch()
conf = cfg.CONF
conf.register_opts(L3NATAgent.OPTS)
+ config.register_root_helper(conf)
conf.register_opts(interface.OPTS)
conf.register_opts(external_process.OPTS)
conf()
import netaddr
+from quantum.agent.common import config
from quantum.agent.linux import ip_lib
from quantum.agent.linux import ovs_lib
from quantum.agent.linux import utils
def __init__(self, conf):
self.conf = conf
+ self.root_helper = config.get_root_helper(conf)
def init_l3(self, device_name, ip_cidrs, namespace=None):
"""Set the L3 settings for the interface using data from the port.
ip_cidrs: list of 'X.X.X.X/YY' strings
"""
- device = ip_lib.IPDevice(device_name, self.conf.root_helper,
+ device = ip_lib.IPDevice(device_name,
+ self.root_helper,
namespace=namespace)
previous = {}
'external-ids:iface-status=active',
'--', 'set', 'Interface', device_name,
'external-ids:attached-mac=%s' % mac_address]
- utils.execute(cmd, self.conf.root_helper)
+ utils.execute(cmd, self.root_helper)
def plug(self, network_id, port_id, device_name, mac_address,
bridge=None, namespace=None, prefix=None):
self.check_bridge_exists(bridge)
if not ip_lib.device_exists(device_name,
- self.conf.root_helper,
+ self.root_helper,
namespace=namespace):
- ip = ip_lib.IPWrapper(self.conf.root_helper)
+ ip = ip_lib.IPWrapper(self.root_helper)
tap_name = self._get_tap_name(device_name, prefix)
if self.conf.ovs_use_veth:
tap_name = self._get_tap_name(device_name, prefix)
self.check_bridge_exists(bridge)
- ovs = ovs_lib.OVSBridge(bridge, self.conf.root_helper)
+ ovs = ovs_lib.OVSBridge(bridge, self.root_helper)
try:
ovs.delete_port(tap_name)
if self.conf.ovs_use_veth:
- device = ip_lib.IPDevice(device_name, self.conf.root_helper,
+ device = ip_lib.IPDevice(device_name,
+ self.root_helper,
namespace)
device.link.delete()
LOG.debug(_("Unplugged interface '%s'"), device_name)
bridge=None, namespace=None, prefix=None):
"""Plugin the interface."""
if not ip_lib.device_exists(device_name,
- self.conf.root_helper,
+ self.root_helper,
namespace=namespace):
- ip = ip_lib.IPWrapper(self.conf.root_helper)
+ ip = ip_lib.IPWrapper(self.root_helper)
# Enable agent to define the prefix
if prefix:
def unplug(self, device_name, bridge=None, namespace=None, prefix=None):
"""Unplug the interface."""
- device = ip_lib.IPDevice(device_name, self.conf.root_helper, namespace)
+ device = ip_lib.IPDevice(device_name, self.root_helper, namespace)
try:
device.link.delete()
LOG.debug(_("Unplugged interface '%s'"), device_name)
return self.flavor_driver_map[flavor]
def _get_driver_by_device_name(self, device_name, namespace=None):
- device = ip_lib.IPDevice(device_name, self.conf.root_helper, namespace)
+ device = ip_lib.IPDevice(device_name, self.root_helper, namespace)
mac_address = device.link.address
ports = self.quantum.list_ports(mac_address=mac_address)
if not ports.get('ports'):
from quantum.agent import dhcp_agent
from quantum.agent import l3_agent
+from quantum.agent.common import config as agent_config
from quantum.agent.linux import dhcp
from quantum.agent.linux import ip_lib
from quantum.agent.linux import ovs_lib
"""
opts = [
- cfg.StrOpt('root_helper', default='sudo',
- help=_("Root helper application.")),
cfg.StrOpt('dhcp_driver',
default='quantum.agent.linux.dhcp.Dnsmasq',
help=_("The driver used to manage the DHCP server.")),
default=False,
help=_('Delete the namespace by removing all devices.')),
]
+
conf = cfg.ConfigOpts()
conf.register_opts(opts)
+ agent_config.register_root_helper(conf)
conf.register_opts(dhcp.OPTS)
config.setup_logging(conf)
return conf
def kill_dhcp(conf, namespace):
"""Disable DHCP for a network if DHCP is still active."""
+ root_helper = agent_config.get_root_helper(conf)
network_id = namespace.replace(dhcp_agent.NS_PREFIX, '')
null_delegate = NullDelegate()
conf.dhcp_driver,
conf,
FakeNetwork(network_id),
- conf.root_helper,
+ root_helper,
null_delegate)
if dhcp_driver.active:
if not re.match(NS_MANGLING_PATTERN, namespace):
return False
- ip = ip_lib.IPWrapper(conf.root_helper, namespace)
+ root_helper = agent_config.get_root_helper(conf)
+ ip = ip_lib.IPWrapper(root_helper, namespace)
return force or ip.namespace_is_empty()
try:
device.link.delete()
except RuntimeError:
+ root_helper = agent_config.get_root_helper(conf)
# Maybe the device is OVS port, so try to delete
- bridge_name = ovs_lib.get_bridge_for_iface(conf.root_helper,
- device.name)
+ bridge_name = ovs_lib.get_bridge_for_iface(root_helper, device.name)
if bridge_name:
- bridge = ovs_lib.OVSBridge(bridge_name,
- conf.root_helper)
+ bridge = ovs_lib.OVSBridge(bridge_name, root_helper)
bridge.delete_port(device.name)
else:
LOG.debug(_('Unable to find bridge for device: %s'), device.name)
"""
try:
- ip = ip_lib.IPWrapper(conf.root_helper, namespace)
+ root_helper = agent_config.get_root_helper(conf)
+ ip = ip_lib.IPWrapper(root_helper, namespace)
if force:
kill_dhcp(conf, namespace)
conf = setup_conf()
conf()
+ 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(conf.root_helper)
+ ip_lib.IPWrapper.get_namespaces(root_helper)
if eligible_for_deletion(conf, ns, conf.force)]
if candidates:
# under the License.
from quantum.agent import l3_agent
+from quantum.agent.common import config as agent_config
from quantum.agent.linux import interface
from quantum.agent.linux import ip_lib
from quantum.agent.linux import ovs_lib
'bridges.'))
]
- agent_opts = [
- cfg.StrOpt('root_helper', default='sudo',
- help=_("Root helper application.")),
- ]
-
conf = cfg.ConfigOpts()
conf.register_cli_opts(opts)
conf.register_opts(l3_agent.L3NATAgent.OPTS)
conf.register_opts(interface.OPTS)
- conf.register_opts(agent_opts, 'AGENT')
+ agent_config.register_root_helper(conf)
config.setup_logging(conf)
return conf
import netaddr
+from quantum.agent.common import config
from quantum.agent.dhcp_agent import DictModel
from quantum.agent.linux import ip_lib
from quantum.agent.linux import utils
class QuantumDebugAgent():
OPTS = [
- cfg.StrOpt('root_helper', default='sudo',
- help=_("Root helper application.")),
# Needed for drivers
cfg.StrOpt('admin_user',
help=_("Admin user")),
def __init__(self, conf, client, driver):
self.conf = conf
+ self.root_helper = config.get_root_helper(conf)
self.client = client
self.driver = driver
if self.conf.use_namespaces:
namespace = self._get_namespace(port)
- if ip_lib.device_exists(interface_name,
- self.conf.root_helper, namespace):
+ if ip_lib.device_exists(interface_name, self.root_helper, namespace):
LOG.debug(_('Reusing existing device: %s.'), interface_name)
else:
self.driver.plug(network.id,
bridge = None
if network.external:
bridge = self.conf.external_network_bridge
- ip = ip_lib.IPWrapper(self.conf.root_helper)
+ ip = ip_lib.IPWrapper(self.root_helper)
namespace = self._get_namespace(port)
if self.conf.use_namespaces and ip.netns.exists(namespace):
self.driver.unplug(self.driver.get_device_name(port),
def exec_command(self, port_id, command=None):
port = DictModel(self.client.show_port(port_id)['port'])
- ip = ip_lib.IPWrapper(self.conf.root_helper)
+ ip = ip_lib.IPWrapper(self.root_helper)
namespace = self._get_namespace(port)
if self.conf.use_namespaces:
if not command:
client = self.client_manager.quantum
cfg.CONF.register_opts(interface.OPTS)
cfg.CONF.register_opts(QuantumDebugAgent.OPTS)
+ config.register_root_helper(cfg.CONF)
cfg.CONF(['--config-file', self.options.config_file])
config.setup_logging(cfg.CONF)
driver = importutils.import_object(cfg.CONF.interface_driver, cfg.CONF)
# @author: Sumit Naiksatam, Cisco Systems, Inc.
# @author: Rohit Agarwalla, Cisco Systems, Inc.
+from quantum.agent.common import config
from quantum.openstack.common import cfg
DEFAULT_VLAN_RANGES = []
cfg.IntOpt('polling_interval', default=2,
help=_("The number of seconds the agent will wait between "
"polling for local device changes.")),
- cfg.StrOpt('root_helper', default='sudo',
- help=_("Root helper application.")),
]
cfg.CONF.register_opts(vlan_opts, "VLANS")
cfg.CONF.register_opts(bridge_opts, "LINUX_BRIDGE")
cfg.CONF.register_opts(agent_opts, "AGENT")
+config.register_root_helper(cfg.CONF)
# under the License.
# @author: Ryota MIBU
+from quantum.agent.common import config
from quantum.openstack.common import cfg
# import rpc config options
from quantum.openstack.common import rpc
cfg.IntOpt('polling_interval', default=2,
help=_("The number of seconds the agent will wait between "
"polling for local device changes.")),
- cfg.StrOpt('root_helper', default='sudo',
- help=_("Root helper application.")),
]
ofc_opts = [
cfg.CONF.register_opts(ovs_opts, "OVS")
cfg.CONF.register_opts(agent_opts, "AGENT")
cfg.CONF.register_opts(ofc_opts, "OFC")
+config.register_root_helper(cfg.CONF)
# shortcuts
CONF = cfg.CONF
# License for the specific language governing permissions and limitations
# under the License.
+from quantum.agent.common import config
from quantum.openstack.common import cfg
cfg.IntOpt('polling_interval', default=2,
help=_("The number of seconds the agent will wait between "
"polling for local device changes.")),
- cfg.StrOpt('root_helper', default='sudo',
- help=_("Root helper application.")),
]
cfg.CONF.register_opts(ovs_opts, "OVS")
cfg.CONF.register_opts(agent_opts, "AGENT")
+config.register_root_helper(cfg.CONF)
# License for the specific language governing permissions and limitations
# under the License.
+from quantum.agent.common import config
from quantum.openstack.common import cfg
help=_("OVSDB interface to connect to")),
]
-agent_opts = [
- cfg.StrOpt('root_helper', default='sudo',
- help=_("Root helper application.")),
-]
-
cfg.CONF.register_opts(ovs_opts, "OVS")
-cfg.CONF.register_opts(agent_opts, "AGENT")
+config.register_root_helper(cfg.CONF)
# License for the specific language governing permissions and limitations
# under the License.
+import unittest2
+
from quantum.agent.common import config
def test_setup_conf():
conf = config.setup_conf()
assert conf.state_path.endswith('/var/lib/quantum')
+
+
+class TestRootHelper(unittest2.TestCase):
+
+ def test_agent_root_helper(self):
+ conf = config.setup_conf()
+ config.register_root_helper(conf)
+ conf.set_override('root_helper', 'my_root_helper', 'AGENT')
+ self.assertEquals(config.get_root_helper(conf), 'my_root_helper')
+
+ def test_root_helper(self):
+ conf = config.setup_conf()
+ config.register_root_helper(conf)
+ conf.set_override('root_helper', 'my_root_helper')
+ self.assertEquals(config.get_root_helper(conf), 'my_root_helper')
+
+ def test_root_default(self):
+ conf = config.setup_conf()
+ config.register_root_helper(conf)
+ self.assertEquals(config.get_root_helper(conf), 'sudo')
def test_kill_dhcp(self, dhcp_active=True):
conf = mock.Mock()
- conf.root_helper = 'sudo',
+ conf.AGENT.root_helper = 'sudo',
conf.dhcp_driver = 'driver'
method_to_patch = 'quantum.openstack.common.importutils.import_object'
util.kill_dhcp(conf, 'ns')
- import_object.called_once_with('driver', conf, mock.ANY, 'sudo',
+ import_object.called_once_with('driver', conf, mock.ANY,
+ conf.AGENT.root_helper,
mock.ANY)
if dhcp_active:
expected):
ns = prefix + '6e322ac7-ab50-4f53-9cdc-d1d3c1164b6d'
conf = mock.Mock()
- conf.root_helper = 'sudo'
with mock.patch('quantum.agent.linux.ip_lib.IPWrapper') as ip_wrap:
ip_wrap.return_value.namespace_is_empty.return_value = is_empty
self.assertEqual(util.eligible_for_deletion(conf, ns, force),
expected)
- expected_calls = [mock.call('sudo', ns)]
+ expected_calls = [mock.call(conf.AGENT.root_helper, ns)]
if not force:
expected_calls.append(mock.call().namespace_is_empty())
ip_wrap.assert_has_calls(expected_calls)
def test_unplug_device_ovs_port(self):
conf = mock.Mock()
conf.ovs_integration_bridge = 'br-int'
- conf.root_helper = 'sudo'
device = mock.Mock()
device.name = 'tap1'
util.unplug_device(conf, device)
mock_get_bridge_for_iface.assert_called_once_with(
- conf.root_helper, 'tap1')
- ovs_br_cls.called_once_with('br-int', 'sudo')
+ conf.AGENT.root_helper, 'tap1')
+ ovs_br_cls.called_once_with('br-int', conf.AGENT.root_helper)
ovs_bridge.assert_has_calls(
[mock.call.delete_port(device.name)])
def test_unplug_device_cannot_determine_bridge_port(self):
conf = mock.Mock()
conf.ovs_integration_bridge = 'br-int'
- conf.root_helper = 'sudo'
device = mock.Mock()
device.name = 'tap1'
util.unplug_device(conf, device)
mock_get_bridge_for_iface.assert_called_once_with(
- conf.root_helper, 'tap1')
+ conf.AGENT.root_helper, 'tap1')
self.assertEqual(ovs_br_cls.mock_calls, [])
self.assertTrue(debug.called)
def _test_destroy_namespace_helper(self, force, num_devices):
ns = 'qrouter-6e322ac7-ab50-4f53-9cdc-d1d3c1164b6d'
conf = mock.Mock()
- conf.root_helper = 'sudo'
+# conf.AGENT.root_helper = 'sudo'
lo_device = mock.Mock()
lo_device.name = 'lo'
with mock.patch.object(util, 'kill_dhcp') as kill_dhcp:
util.destroy_namespace(conf, ns, force)
- expected = [mock.call('sudo', ns)]
+ expected = [mock.call(conf.AGENT.root_helper, ns)]
if force:
expected.extend([
def test_destroy_namespace_exception(self):
ns = 'qrouter-6e322ac7-ab50-4f53-9cdc-d1d3c1164b6d'
conf = mock.Mock()
- conf.root_helper = 'sudo'
+ conf.AGENT.root_helper = 'sudo'
with mock.patch('quantum.agent.linux.ip_lib.IPWrapper') as ip_wrap:
ip_wrap.side_effect = Exception()
util.destroy_namespace(conf, ns)
with mock.patch('eventlet.sleep') as eventlet_sleep:
conf = mock.Mock()
- conf.root_helper = 'sudo'
conf.force = False
methods_to_mock = dict(
eligible_for_deletion=mock.DEFAULT,
mock.call(conf, 'ns2', False)])
ip_wrap.assert_has_calls(
- [mock.call.get_namespaces('sudo')])
+ [mock.call.get_namespaces(conf.AGENT.root_helper)])
eventlet_sleep.assert_called_once_with(2)
with mock.patch('eventlet.sleep') as eventlet_sleep:
conf = mock.Mock()
- conf.root_helper = 'sudo'
conf.force = False
methods_to_mock = dict(
eligible_for_deletion=mock.DEFAULT,
util.main()
ip_wrap.assert_has_calls(
- [mock.call.get_namespaces('sudo')])
+ [mock.call.get_namespaces(conf.AGENT.root_helper)])
mocks['eligible_for_deletion'].assert_has_calls(
[mock.call(conf, 'ns1', False),
import mock
from quantum.agent import l3_agent
+from quantum.agent.common import config as agent_config
from quantum.agent.linux import interface
from quantum.common import config as base_config
from quantum.common import constants as l3_constants
self.conf = cfg.ConfigOpts()
self.conf.register_opts(base_config.core_opts)
self.conf.register_opts(l3_agent.L3NATAgent.OPTS)
+ agent_config.register_root_helper(self.conf)
self.conf.register_opts(interface.OPTS)
self.conf.set_override('interface_driver',
'quantum.agent.linux.interface.NullDriver')
]
self.conf = config.setup_conf()
self.conf.register_opts(interface.OPTS)
- self.conf.register_opts(root_helper_opt)
+ config.register_root_helper(self.conf)
self.ip_dev_p = mock.patch.object(ip_lib, 'IPDevice')
self.ip_dev = self.ip_dev_p.start()
self.ip_p = mock.patch.object(ip_lib, 'IPWrapper')
Code Review / openstack-build / neutron-build.git / commitdiff