Change ip_lib network namespace code to use pyroute2

Change network namespace add/delete/list code to use
pyroute2 library instead of calling /sbin/ip.

Also changed all in-tree callers to use the new calls.

Closes-bug: #1717582
Related-bug: #1492714

Change-Id: Id802e77543177fbb95ff15c2c7361172e8824633
changes/01/505701/12
Brian Haley 5 years ago committed by Slawek Kaplonski
parent d2bf790a44
commit 4f627b4e8d
  1. 2
      neutron/agent/dhcp_agent.py
  2. 6
      neutron/agent/l3/dvr_local_router.py
  3. 3
      neutron/agent/l3/l3_agent_extension_api.py
  4. 3
      neutron/agent/l3/namespace_manager.py
  5. 5
      neutron/agent/linux/dhcp.py
  6. 62
      neutron/agent/linux/ip_lib.py
  7. 2
      neutron/agent/linux/iptables_manager.py
  8. 2
      neutron/cmd/netns_cleanup.py
  9. 21
      neutron/cmd/sanity/checks.py
  10. 5
      neutron/debug/debug_agent.py
  11. 32
      neutron/privileged/agent/linux/ip_lib.py
  12. 5
      neutron/services/metering/drivers/iptables/iptables_driver.py
  13. 3
      neutron/tests/common/net_helpers.py
  14. 2
      neutron/tests/fullstack/base.py
  15. 3
      neutron/tests/fullstack/resources/machine.py
  16. 5
      neutron/tests/fullstack/resources/process.py
  17. 4
      neutron/tests/fullstack/test_l3_agent.py
  18. 3
      neutron/tests/functional/agent/l3/framework.py
  19. 3
      neutron/tests/functional/agent/l3/test_namespace_manager.py
  20. 4
      neutron/tests/functional/agent/test_dhcp_agent.py
  21. 2
      neutron/tests/functional/cmd/test_netns_cleanup.py
  22. 16
      neutron/tests/unit/agent/l3/test_agent.py
  23. 7
      neutron/tests/unit/agent/l3/test_dvr_snat_ns.py
  24. 32
      neutron/tests/unit/agent/l3/test_l3_agent_extension_api.py
  25. 8
      neutron/tests/unit/agent/l3/test_namespace_manager.py
  26. 22
      neutron/tests/unit/agent/linux/test_dhcp.py
  27. 90
      neutron/tests/unit/agent/linux/test_ip_lib.py
  28. 4
      neutron/tests/unit/agent/linux/test_iptables_manager.py
  29. 16
      neutron/tests/unit/cmd/test_netns_cleanup.py
  30. 9
      neutron/tests/unit/debug/test_commands.py
  31. 2
      neutron/tests/unit/services/metering/drivers/test_iptables.py

@ -34,12 +34,14 @@ def register_options(conf):
dhcp_config.register_agent_dhcp_opts(conf)
meta_conf.register_meta_conf_opts(meta_conf.SHARED_OPTS, conf)
config.register_interface_opts(conf)
config.register_root_helper(conf)
def main():
register_options(cfg.CONF)
common_config.init(sys.argv[1:])
config.setup_logging()
config.setup_privsep()
server = neutron_service.Service.create(
binary='neutron-dhcp-agent',
topic=topics.DHCP_AGENT,

@ -596,8 +596,7 @@ class DvrLocalRouter(dvr_router_base.DvrRouterBase):
def _delete_interface_route_in_fip_ns(self, router_port):
rtr_2_fip_ip, fip_2_rtr_name = self.get_rtr_fip_ip_and_interface_name()
fip_ns_name = self.fip_ns.get_name()
ip_wrapper = ip_lib.IPWrapper(namespace=fip_ns_name)
if ip_wrapper.netns.exists(fip_ns_name):
if ip_lib.network_namespace_exists(fip_ns_name):
device = ip_lib.IPDevice(fip_2_rtr_name, namespace=fip_ns_name)
if not device.exists():
return
@ -608,8 +607,7 @@ class DvrLocalRouter(dvr_router_base.DvrRouterBase):
def _add_interface_route_to_fip_ns(self, router_port):
rtr_2_fip_ip, fip_2_rtr_name = self.get_rtr_fip_ip_and_interface_name()
fip_ns_name = self.fip_ns.get_name()
ip_wrapper = ip_lib.IPWrapper(namespace=fip_ns_name)
if ip_wrapper.netns.exists(fip_ns_name):
if ip_lib.network_namespace_exists(fip_ns_name):
device = ip_lib.IPDevice(fip_2_rtr_name, namespace=fip_ns_name)
if not device.exists():
return

@ -30,8 +30,7 @@ class L3AgentExtensionAPI(object):
self._router_info = router_info
def _local_namespaces(self):
root_ip = ip_lib.IPWrapper()
local_ns_list = root_ip.get_namespaces()
local_ns_list = ip_lib.list_network_namespaces()
return set(local_ns_list)
def get_router_hosting_port(self, port_id):

@ -114,8 +114,7 @@ class NamespaceManager(object):
def list_all(self):
"""Get a set of all namespaces on host managed by this manager."""
try:
root_ip = ip_lib.IPWrapper()
namespaces = root_ip.get_namespaces()
namespaces = ip_lib.list_network_namespaces()
return set(ns for ns in namespaces if self.is_managed(ns))
except RuntimeError:
LOG.exception('RuntimeError in obtaining namespace list for '

@ -241,12 +241,11 @@ class DhcpLocalProcess(DhcpBase):
LOG.warning('Failed trying to delete interface: %s',
self.interface_name)
ns_ip = ip_lib.IPWrapper(namespace=self.network.namespace)
if not ns_ip.netns.exists(self.network.namespace):
if not ip_lib.network_namespace_exists(self.network.namespace):
LOG.debug("Namespace already deleted: %s", self.network.namespace)
return
try:
ns_ip.netns.delete(self.network.namespace)
ip_lib.delete_network_namespace(self.network.namespace)
except RuntimeError:
LOG.warning('Failed trying to delete namespace: %s',
self.network.namespace)

@ -17,6 +17,7 @@ import os
import re
import time
from debtcollector import removals
import eventlet
import netaddr
from neutron_lib import constants
@ -24,6 +25,7 @@ from neutron_lib import exceptions
from oslo_config import cfg
from oslo_log import log as logging
from oslo_utils import excutils
from pyroute2 import netns
import six
from neutron._i18n import _
@ -256,12 +258,13 @@ class IPWrapper(SubProcessBase):
self._as_root([], 'link', cmd)
return (IPDevice(name, namespace=self.namespace))
@removals.remove(version='Queens', removal_version='Rocky',
message="This will be removed in the future. Please use "
"'neutron.agent.linux.ip_lib."
"list_network_namespaces' instead.")
@classmethod
def get_namespaces(cls):
output = cls._execute(
[], 'netns', ('list',),
run_as_root=cfg.CONF.AGENT.use_helper_for_ns_read)
return [l.split()[0] for l in output.splitlines()]
return list_network_namespaces()
class IPDevice(SubProcessBase):
@ -863,14 +866,14 @@ class IpNetnsCommand(IpCommandBase):
COMMAND = 'netns'
def add(self, name):
self._as_root([], ('add', name), use_root_namespace=True)
create_network_namespace(name)
wrapper = IPWrapper(namespace=name)
wrapper.netns.execute(['sysctl', '-w',
'net.ipv4.conf.all.promote_secondaries=1'])
return wrapper
def delete(self, name):
self._as_root([], ('delete', name), use_root_namespace=True)
delete_network_namespace(name)
def execute(self, cmds, addl_env=None, check_exit_code=True,
log_fail_as_error=True, extra_ok_codes=None,
@ -891,13 +894,7 @@ class IpNetnsCommand(IpCommandBase):
log_fail_as_error=log_fail_as_error, **kwargs)
def exists(self, name):
output = self._parent._execute(
['o'], 'netns', ['list'],
run_as_root=cfg.CONF.AGENT.use_helper_for_ns_read)
for line in [l.split()[0] for l in output.splitlines()]:
if name == line:
return True
return False
return network_namespace_exists(name)
def vlan_in_use(segmentation_id, namespace=None):
@ -1018,6 +1015,45 @@ def dump_neigh_entries(ip_version, device=None, namespace=None, **kwargs):
**kwargs))
def create_network_namespace(namespace, **kwargs):
"""Create a network namespace.
:param namespace: The name of the namespace to create
:param kwargs: Callers add any filters they use as kwargs
"""
privileged.create_netns(namespace, **kwargs)
def delete_network_namespace(namespace, **kwargs):
"""Delete a network namespace.
:param namespace: The name of the namespace to delete
:param kwargs: Callers add any filters they use as kwargs
"""
privileged.remove_netns(namespace, **kwargs)
def list_network_namespaces(**kwargs):
"""List all network namespace entries.
:param kwargs: Callers add any filters they use as kwargs
"""
if cfg.CONF.AGENT.use_helper_for_ns_read:
return privileged.list_netns(**kwargs)
else:
return netns.listnetns(**kwargs)
def network_namespace_exists(namespace, **kwargs):
"""Check if a network namespace exists.
:param namespace: The name of the namespace to check
:param kwargs: Callers add any filters they use as kwargs
"""
output = list_network_namespaces(**kwargs)
return namespace in output
def ensure_device_is_ready(device_name, namespace=None):
dev = IPDevice(device_name, namespace=namespace)
dev.set_log_fail_as_error(False)

@ -523,7 +523,7 @@ class IptablesManager(object):
# exist.
with excutils.save_and_reraise_exception() as ctx:
if (self.namespace and not
ip_lib.IPWrapper().netns.exists(self.namespace)):
ip_lib.network_namespace_exists(self.namespace)):
ctx.reraise = False
LOG.error("Namespace %s was deleted during IPTables "
"operations.", self.namespace)

@ -256,7 +256,7 @@ def destroy_namespace(conf, namespace, force=False):
def cleanup_network_namespaces(conf):
# Identify namespaces that are candidates for deletion.
candidates = [ns for ns in
ip_lib.IPWrapper.get_namespaces()
ip_lib.list_network_namespaces()
if eligible_for_deletion(conf, ns, conf.force)]
if candidates:

@ -181,15 +181,13 @@ def vf_extended_management_supported():
def netns_read_requires_helper():
ipw = ip_lib.IPWrapper()
nsname = "netnsreadtest-" + uuidutils.generate_uuid()
ipw.netns.add(nsname)
ip_lib.create_network_namespace(nsname)
try:
# read without root_helper. if exists, not required.
ipw_nohelp = ip_lib.IPWrapper()
exists = ipw_nohelp.netns.exists(nsname)
exists = ip_lib.network_namespace_exists(nsname)
finally:
ipw.netns.delete(nsname)
ip_lib.delete_network_namespace(nsname)
return not exists
@ -294,7 +292,7 @@ class KeepalivedIPv6Test(object):
common_utils.wait_until_true(_gw_vip_assigned)
def __enter__(self):
ip_lib.IPWrapper().netns.add(self.nsname)
ip_lib.create_network_namespace(self.nsname)
return self
def __exit__(self, exc_type, exc_value, exc_tb):
@ -304,7 +302,7 @@ class KeepalivedIPv6Test(object):
self.manager.disable()
if self.config_path:
shutil.rmtree(self.config_path, ignore_errors=True)
ip_lib.IPWrapper().netns.delete(self.nsname)
ip_lib.delete_network_namespace(self.nsname)
cfg.CONF.set_override('check_child_processes_interval',
self.orig_interval, 'AGENT')
@ -450,13 +448,12 @@ def _fix_ip_nonlocal_bind_root_value(original_value):
def ip_nonlocal_bind():
ipw = ip_lib.IPWrapper()
nsname1 = "ipnonlocalbind1-" + uuidutils.generate_uuid()
nsname2 = "ipnonlocalbind2-" + uuidutils.generate_uuid()
ipw.netns.add(nsname1)
ip_lib.create_network_namespace(nsname1)
try:
ipw.netns.add(nsname2)
ip_lib.create_network_namespace(nsname2)
try:
original_value = ip_lib.get_ip_nonlocal_bind(namespace=None)
try:
@ -470,7 +467,7 @@ def ip_nonlocal_bind():
"Exception: %s", e)
return False
finally:
ipw.netns.delete(nsname2)
ip_lib.delete_network_namespace(nsname2)
finally:
ipw.netns.delete(nsname1)
ip_lib.delete_network_namespace(nsname1)
return ns1_value == 0

@ -98,14 +98,13 @@ class NeutronDebugAgent(object):
bridge = None
if network.external:
bridge = self.conf.external_network_bridge
ip = ip_lib.IPWrapper()
namespace = self._get_namespace(port)
if ip.netns.exists(namespace):
if ip_lib.network_namespace_exists(namespace):
self.driver.unplug(self.driver.get_device_name(port),
bridge=bridge,
namespace=namespace)
try:
ip.netns.delete(namespace)
ip_lib.delete_network_namespace(namespace)
except Exception:
LOG.warning('Failed to delete namespace %s', namespace)
else:

@ -17,6 +17,7 @@ import pyroute2
from pyroute2.netlink import rtnl
from pyroute2.netlink.rtnl import ndmsg
from pyroute2 import NetlinkError
from pyroute2 import netns
from neutron._i18n import _
from neutron import privileged
@ -175,3 +176,34 @@ def dump_neigh_entries(ip_version, device, namespace, **kwargs):
'lladdr': attrs.get('NDA_LLADDR'),
'device': device}]
return entries
@privileged.default.entrypoint
def create_netns(name, **kwargs):
"""Create a network namespace.
:param name: The name of the namespace to create
"""
try:
netns.create(name, **kwargs)
except OSError as e:
if e.errno != errno.EEXIST:
raise
@privileged.default.entrypoint
def remove_netns(name, **kwargs):
"""Remove a network namespace.
:param name: The name of the namespace to remove
"""
netns.remove(name, **kwargs)
@privileged.default.entrypoint
def list_netns(**kwargs):
"""List network namespaces.
Caller requires raised priveleges to list namespaces
"""
return netns.listnetns(**kwargs)

@ -81,7 +81,7 @@ class RouterWithMetering(object):
self.id)
# Check for namespace existence before we assign the
# snat_iptables_manager
if ip_lib.IPWrapper().netns.exists(snat_ns_name):
if ip_lib.network_namespace_exists(snat_ns_name):
self.snat_iptables_manager = iptables_manager.IptablesManager(
namespace=snat_ns_name,
binary_name=WRAP_NAME,
@ -91,8 +91,7 @@ class RouterWithMetering(object):
# NOTE(Swami): If distributed routers, all external traffic on a
# compute node will flow through the rfp interface in the router
# namespace.
ip_wrapper = ip_lib.IPWrapper(namespace=self.ns_name)
if ip_wrapper.netns.exists(self.ns_name):
if ip_lib.network_namespace_exists(self.ns_name):
self.iptables_manager = iptables_manager.IptablesManager(
namespace=self.ns_name,
binary_name=WRAP_NAME,

@ -658,8 +658,7 @@ class MacvtapFixture(fixtures.Fixture):
self.addCleanup(self.destroy)
def destroy(self):
ip_wrapper = ip_lib.IPWrapper(self.ip_dev.namespace)
if (ip_wrapper.netns.exists(self.ip_dev.namespace) or
if (ip_lib.network_namespace_exists(self.ip_dev.namespace) or
self.ip_dev.namespace is None):
try:
self.ip_dev.link.delete()

@ -16,6 +16,7 @@ import os
from oslo_config import cfg
from neutron.conf.agent import common as config
from neutron.tests import base as tests_base
from neutron.tests.common import helpers
from neutron.tests.fullstack.resources import client as client_resource
@ -60,6 +61,7 @@ class BaseFullStackTestCase(testlib_api.MySQLTestCaseMixin,
# configure test runner to use rootwrap
self.setup_rootwrap()
config.setup_privsep()
self.environment = environment
self.environment.test_name = self.get_name()

@ -200,8 +200,7 @@ class FakeFullstackMachine(machine_fixtures.FakeMachineBase):
# All associated vlan interfaces are deleted too
self.bridge.delete_port(self.port.name)
ip_wrap = ip_lib.IPWrapper(self.namespace)
ip_wrap.netns.delete(self.namespace)
ip_lib.delete_network_namespace(self.namespace)
class FakeFullstackTrunkMachine(FakeFullstackMachine):

@ -348,11 +348,10 @@ class DhcpAgentFixture(fixtures.Fixture):
namespace suffix.
"""
ip_wrapper = ip_lib.IPWrapper()
for namespace in ip_wrapper.get_namespaces():
for namespace in ip_lib.list_network_namespaces():
if self.dhcp_namespace_pattern.match(namespace):
try:
ip_wrapper.netns.delete(namespace)
ip_lib.delete_network_namespace(namespace)
except RuntimeError:
# Continue cleaning even if namespace deletions fails
pass

@ -93,8 +93,8 @@ class TestLegacyL3Agent(TestL3Agent):
return namespaces.build_ns_name(namespaces.NS_PREFIX, router_id)
def _assert_namespace_exists(self, ns_name):
ip = ip_lib.IPWrapper(ns_name)
common_utils.wait_until_true(lambda: ip.netns.exists(ns_name))
common_utils.wait_until_true(
lambda: ip_lib.network_namespace_exists(ns_name))
def test_namespace_exists(self):
tenant_id = uuidutils.generate_uuid()

@ -340,8 +340,7 @@ class L3AgentTestFramework(base.BaseSudoTestCase):
ip_version, ipv6_subnet_modes, interface_id)
def _namespace_exists(self, namespace):
ip = ip_lib.IPWrapper(namespace=namespace)
return ip.netns.exists(namespace)
return ip_lib.network_namespace_exists(namespace)
def _metadata_proxy_exists(self, conf, router):
pm = external_process.ProcessManager(

@ -53,8 +53,7 @@ class NamespaceManagerTestFramework(base.BaseSudoTestCase):
raise e
def _namespace_exists(self, namespace):
ip = ip_lib.IPWrapper(namespace=namespace)
return ip.netns.exists(namespace)
return ip_lib.network_namespace_exists(namespace)
class NamespaceManagerTestCase(NamespaceManagerTestFramework):

@ -173,8 +173,8 @@ class DHCPAgentOVSTestFramework(base.BaseSudoTestCase):
self.assert_dhcp_device(network.namespace, iface_name, dhcp_enabled)
def assert_dhcp_namespace(self, namespace, dhcp_enabled):
ip = ip_lib.IPWrapper()
self.assertEqual(dhcp_enabled, ip.netns.exists(namespace))
self.assertEqual(dhcp_enabled,
ip_lib.network_namespace_exists(namespace))
def assert_accept_ra_disabled(self, namespace):
actual = ip_lib.IPWrapper(namespace=namespace).netns.execute(

@ -30,7 +30,7 @@ from neutron.tests.common import net_helpers
from neutron.tests.functional import base
from neutron.tests.functional.cmd import process_spawn
GET_NAMESPACES = 'neutron.agent.linux.ip_lib.IPWrapper.get_namespaces'
GET_NAMESPACES = 'neutron.agent.linux.ip_lib.list_network_namespaces'
TEST_INTERFACE_DRIVER = 'neutron.agent.linux.interface.OVSInterfaceDriver'
NUM_SUBPROCESSES = 6

@ -89,6 +89,10 @@ class BasicRouterOperationsFramework(base.BaseTestCase):
'neutron.agent.linux.ip_lib.device_exists')
self.device_exists = self.device_exists_p.start()
self.list_network_namespaces_p = mock.patch(
'neutron.agent.linux.ip_lib.list_network_namespaces')
self.list_network_namespaces = self.list_network_namespaces_p.start()
self.ensure_dir = mock.patch(
'oslo_utils.fileutils.ensure_tree').start()
@ -373,7 +377,7 @@ class TestBasicRouterOperations(BasicRouterOperationsFramework):
for r_id in stale_router_ids]
namespace_list += [namespaces.NS_PREFIX + r['id']
for r in active_routers]
self.mock_ip.get_namespaces.return_value = namespace_list
self.list_network_namespaces.return_value = namespace_list
driver = metadata_driver.MetadataDriver
with mock.patch.object(
driver, 'destroy_monitored_metadata_proxy') as destroy_proxy:
@ -2327,7 +2331,7 @@ class TestBasicRouterOperations(BasicRouterOperationsFramework):
def test_destroy_namespace(self):
namespace = 'qrouter-bar'
self.mock_ip.get_namespaces.return_value = [namespace]
self.list_network_namespaces.return_value = [namespace]
self.mock_ip.get_devices.return_value = [
l3_test_common.FakeDev('qr-aaaa'),
l3_test_common.FakeDev('rfp-aaaa')]
@ -2356,7 +2360,7 @@ class TestBasicRouterOperations(BasicRouterOperationsFramework):
def test_destroy_snat_namespace(self):
namespace = 'snat-bar'
self.mock_ip.get_namespaces.return_value = [namespace]
self.list_network_namespaces.return_value = [namespace]
self.mock_ip.get_devices.return_value = [
l3_test_common.FakeDev('qg-aaaa'),
l3_test_common.FakeDev('sg-aaaa')]
@ -2614,9 +2618,9 @@ class TestBasicRouterOperations(BasicRouterOperationsFramework):
for r in router_list]
good_namespace_list += [dvr_snat_ns.SNAT_NS_PREFIX + r['id']
for r in router_list]
self.mock_ip.get_namespaces.return_value = (stale_namespace_list +
good_namespace_list +
other_namespaces)
self.list_network_namespaces.return_value = (stale_namespace_list +
good_namespace_list +
other_namespaces)
agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)

@ -18,6 +18,7 @@ from oslo_utils import uuidutils
from neutron.agent.common import utils
from neutron.agent.l3 import dvr_snat_ns
from neutron.agent.linux import ip_lib
from neutron.tests import base
_uuid = uuidutils.generate_uuid
@ -37,7 +38,10 @@ class TestDvrSnatNs(base.BaseTestCase):
use_ipv6=False)
@mock.patch.object(utils, 'execute')
def test_create(self, execute):
@mock.patch.object(ip_lib, 'create_network_namespace')
@mock.patch.object(ip_lib, 'network_namespace_exists')
def test_create(self, exists, create, execute):
exists.return_value = False
self.snat_ns.create()
netns_cmd = ['ip', 'netns', 'exec', self.snat_ns.name]
@ -46,4 +50,5 @@ class TestDvrSnatNs(base.BaseTestCase):
check_exit_code=True, extra_ok_codes=None,
log_fail_as_error=True, run_as_root=True)]
create.assert_called_once_with(self.snat_ns.name)
execute.assert_has_calls(expected)

@ -43,14 +43,14 @@ class TestL3AgentExtensionApi(base.BaseTestCase):
ports = [{'id': pid} for pid in port_ids]
router_info, ri = self._prepare_router_data(ports)
with mock.patch.object(ip_lib.IPWrapper,
'get_namespaces') as mock_get_namespaces:
with mock.patch.object(ip_lib,
'list_network_namespaces') as mock_list_netns:
mock_get_namespaces.return_value = []
mock_list_netns.return_value = []
api_object = l3_agent_api.L3AgentExtensionAPI(router_info)
router = api_object.get_router_hosting_port(port_ids[0])
mock_get_namespaces.assert_called_once_with()
mock_list_netns.assert_called_once_with()
self.assertFalse(router)
def test_get_router_hosting_port_for_router_in_ns(self):
@ -58,9 +58,9 @@ class TestL3AgentExtensionApi(base.BaseTestCase):
ports = [{'id': pid} for pid in port_ids]
router_info, ri = self._prepare_router_data(ports)
with mock.patch.object(ip_lib.IPWrapper,
'get_namespaces') as mock_get_namespaces:
mock_get_namespaces.return_value = [ri.ns_name]
with mock.patch.object(ip_lib,
'list_network_namespaces') as mock_list_netns:
mock_list_netns.return_value = [ri.ns_name]
api_object = l3_agent_api.L3AgentExtensionAPI(router_info)
router = api_object.get_router_hosting_port(port_ids[0])
self.assertEqual(ri, router)
@ -68,9 +68,9 @@ class TestL3AgentExtensionApi(base.BaseTestCase):
def test_get_routers_in_project(self):
router_info, ri = self._prepare_router_data()
with mock.patch.object(ip_lib.IPWrapper,
'get_namespaces') as mock_get_namespaces:
mock_get_namespaces.return_value = [ri.ns_name]
with mock.patch.object(ip_lib,
'list_network_namespaces') as mock_list_netns:
mock_list_netns.return_value = [ri.ns_name]
api_object = l3_agent_api.L3AgentExtensionAPI(router_info)
routers = api_object.get_routers_in_project(self.project_id)
self.assertEqual([ri], routers)
@ -78,9 +78,9 @@ class TestL3AgentExtensionApi(base.BaseTestCase):
def test_is_router_in_namespace_for_in_ns(self):
router_info, ri = self._prepare_router_data()
with mock.patch.object(ip_lib.IPWrapper,
'get_namespaces') as mock_get_namespaces:
mock_get_namespaces.return_value = [ri.ns_name]
with mock.patch.object(ip_lib,
'list_network_namespaces') as mock_list_netns:
mock_list_netns.return_value = [ri.ns_name]
api_object = l3_agent_api.L3AgentExtensionAPI(router_info)
router_in_ns = api_object.is_router_in_namespace(ri.router_id)
self.assertTrue(router_in_ns)
@ -88,9 +88,9 @@ class TestL3AgentExtensionApi(base.BaseTestCase):
def test_is_router_in_namespace_for_not_in_ns(self):
router_info, ri = self._prepare_router_data()
with mock.patch.object(ip_lib.IPWrapper,
'get_namespaces') as mock_get_namespaces:
mock_get_namespaces.return_value = [uuidutils.generate_uuid()]
with mock.patch.object(ip_lib,
'list_network_namespaces') as mock_list_netns:
mock_list_netns.return_value = [uuidutils.generate_uuid()]
api_object = l3_agent_api.L3AgentExtensionAPI(router_info)
router_in_ns = api_object.is_router_in_namespace(ri.router_id)
self.assertFalse(router_in_ns)

@ -77,7 +77,7 @@ class TestNamespaceManager(NamespaceManagerTestCaseFramework):
'dhcp-' + _uuid(), ]
# Test the normal path
with mock.patch.object(ip_lib.IPWrapper, 'get_namespaces',
with mock.patch.object(ip_lib, 'list_network_namespaces',
return_value=ns_names):
retrieved_ns_names = self.ns_manager.list_all()
self.assertEqual(len(ns_names) - 1, len(retrieved_ns_names))
@ -85,8 +85,8 @@ class TestNamespaceManager(NamespaceManagerTestCaseFramework):
self.assertIn(ns_names[i], retrieved_ns_names)
self.assertNotIn(ns_names[-1], retrieved_ns_names)
# Test path where IPWrapper raises exception
with mock.patch.object(ip_lib.IPWrapper, 'get_namespaces',
# Test path where list_network_namespaces() raises exception
with mock.patch.object(ip_lib, 'list_network_namespaces',
side_effect=RuntimeError):
retrieved_ns_names = self.ns_manager.list_all()
self.assertFalse(retrieved_ns_names)
@ -104,7 +104,7 @@ class TestNamespaceManager(NamespaceManagerTestCaseFramework):
ns_names += [dvr_snat_ns.SNAT_NS_PREFIX + _uuid() for _ in range(5)]
ns_names += [namespaces.NS_PREFIX + router_id,
dvr_snat_ns.SNAT_NS_PREFIX + router_id]
with mock.patch.object(ip_lib.IPWrapper, 'get_namespaces',
with mock.patch.object(ip_lib, 'list_network_namespaces',
return_value=ns_names), \
mock.patch.object(self.ns_manager, '_cleanup') as mock_cleanup:
self.ns_manager.ensure_router_cleanup(router_id)

@ -1096,7 +1096,9 @@ class TestDhcpLocalProcess(TestBase):
self.assertTrue(lp.process_monitor.unregister.called)
self.assertTrue(self.external_process().disable.called)
def test_disable_not_active(self):
@mock.patch('neutron.agent.linux.ip_lib.network_namespace_exists')
def test_disable_not_active(self, namespace_exists):
namespace_exists.return_value = False
attrs_to_mock = dict([(a, mock.DEFAULT) for a in
['active', 'interface_name']])
with mock.patch.multiple(LocalChild, **attrs_to_mock) as mocks:
@ -1121,30 +1123,38 @@ class TestDhcpLocalProcess(TestBase):
lp.disable(retain_port=True)
self._assert_disabled(lp)
def test_disable(self):
@mock.patch('neutron.agent.linux.ip_lib.network_namespace_exists')
def test_disable(self, namespace_exists):
namespace_exists.return_value = True
attrs_to_mock = {'active': mock.DEFAULT}
with mock.patch.multiple(LocalChild, **attrs_to_mock) as mocks:
mocks['active'].__get__ = mock.Mock(return_value=False)
lp = LocalChild(self.conf, FakeDualNetwork())
with mock.patch('neutron.agent.linux.ip_lib.IPWrapper') as ip:
with mock.patch('neutron.agent.linux.ip_lib.'
'delete_network_namespace') as delete_ns:
lp.disable()
self._assert_disabled(lp)
ip.return_value.netns.delete.assert_called_with('qdhcp-ns')
delete_ns.assert_called_with('qdhcp-ns')
def test_disable_config_dir_removed_after_destroy(self):
@mock.patch('neutron.agent.linux.ip_lib.network_namespace_exists')
def test_disable_config_dir_removed_after_destroy(self, namespace_exists):
namespace_exists.return_value = True
parent = mock.MagicMock()
parent.attach_mock(self.rmtree, 'rmtree')
parent.attach_mock(self.mock_mgr, 'DeviceManager')
lp = LocalChild(self.conf, FakeDualNetwork())
lp.disable(retain_port=False)
with mock.patch('neutron.agent.linux.ip_lib.'
'delete_network_namespace') as delete_ns:
lp.disable(retain_port=False)
expected = [mock.call.DeviceManager().destroy(mock.ANY, mock.ANY),
mock.call.rmtree(mock.ANY, ignore_errors=True)]
parent.assert_has_calls(expected)
delete_ns.assert_called_with('qdhcp-ns')
def test_get_interface_name(self):
net = FakeDualNetwork()

@ -36,11 +36,6 @@ NETNS_SAMPLE = [
'bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb',
'cccccccc-cccc-cccc-cccc-cccccccccccc']
NETNS_SAMPLE_IPROUTE2_4 = [
'12345678-1234-5678-abcd-1234567890ab (id: 1)',
'bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb (id: 0)',
'cccccccc-cccc-cccc-cccc-cccccccccccc (id: 2)']
LINK_SAMPLE = [
'1: lo: <LOOPBACK,UP,LOWER_UP> mtu 16436 qdisc noqueue state UNKNOWN \\'
'link/loopback 00:00:00:00:00:00 brd 00:00:00:00:00:00 promiscuity 0',
@ -318,29 +313,31 @@ class TestIpWrapper(base.BaseTestCase):
self.assertEqual(device_name, somedevice.name)
self.assertFalse(devices)
def test_get_namespaces_non_root(self):
@mock.patch.object(pyroute2.netns, 'listnetns')
@mock.patch.object(priv_lib, 'list_netns')
def test_get_namespaces_non_root(self, priv_listnetns, listnetns):
self.config(group='AGENT', use_helper_for_ns_read=False)
self.execute.return_value = '\n'.join(NETNS_SAMPLE)
retval = ip_lib.IPWrapper.get_namespaces()
listnetns.return_value = NETNS_SAMPLE
retval = ip_lib.list_network_namespaces()
self.assertEqual(retval,
['12345678-1234-5678-abcd-1234567890ab',
'bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb',
'cccccccc-cccc-cccc-cccc-cccccccccccc'])
self.assertEqual(1, listnetns.call_count)
self.assertFalse(priv_listnetns.called)
self.execute.assert_called_once_with([], 'netns', ('list',),
run_as_root=False)
def test_get_namespaces_iproute2_4_root(self):
@mock.patch.object(pyroute2.netns, 'listnetns')
@mock.patch.object(priv_lib, 'list_netns')
def test_get_namespaces_root(self, priv_listnetns, listnetns):
self.config(group='AGENT', use_helper_for_ns_read=True)
self.execute.return_value = '\n'.join(NETNS_SAMPLE_IPROUTE2_4)
retval = ip_lib.IPWrapper.get_namespaces()
priv_listnetns.return_value = NETNS_SAMPLE
retval = ip_lib.list_network_namespaces()
self.assertEqual(retval,
['12345678-1234-5678-abcd-1234567890ab',
'bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb',
'cccccccc-cccc-cccc-cccc-cccccccccccc'])
self.execute.assert_called_once_with([], 'netns', ('list',),
run_as_root=True)
self.assertEqual(1, priv_listnetns.call_count)
self.assertFalse(listnetns.called)
def test_add_tuntap(self):
ip_lib.IPWrapper().add_tuntap('tap0')
@ -398,17 +395,16 @@ class TestIpWrapper(base.BaseTestCase):
self.assertEqual(dev.namespace, 'ns')
self.assertEqual(dev.name, 'eth0')
def test_ensure_namespace(self):
@mock.patch.object(priv_lib, 'create_netns')
def test_ensure_namespace(self, create):
with mock.patch.object(ip_lib, 'IPDevice') as ip_dev:
ip = ip_lib.IPWrapper()
with mock.patch.object(ip.netns, 'exists') as ns_exists:
with mock.patch('neutron.agent.common.utils.execute'):
ns_exists.return_value = False
ip.ensure_namespace('ns')
self.execute.assert_has_calls(
[mock.call([], 'netns', ('add', 'ns'),
run_as_root=True, namespace=None,
log_fail_as_error=True)])
create.assert_called_once_with('ns')
ns_exists.assert_called_once_with('ns')
ip_dev.assert_has_calls([mock.call('lo', namespace='ns'),
mock.call().link.set_up()])
@ -1235,10 +1231,11 @@ class TestIpNetnsCommand(TestIPCmdBase):
self.command = 'netns'
self.netns_cmd = ip_lib.IpNetnsCommand(self.parent)
def test_add_namespace(self):
@mock.patch.object(priv_lib, 'create_netns')
def test_add_namespace(self, create):
with mock.patch('neutron.agent.common.utils.execute') as execute:
ns = self.netns_cmd.add('ns')
self._assert_sudo([], ('add', 'ns'), use_root_namespace=True)
create.assert_called_once_with('ns')
self.assertEqual(ns.namespace, 'ns')
execute.assert_called_once_with(
['ip', 'netns', 'exec', 'ns',
@ -1246,36 +1243,35 @@ class TestIpNetnsCommand(TestIPCmdBase):
run_as_root=True, check_exit_code=True, extra_ok_codes=None,
log_fail_as_error=True)
def test_delete_namespace(self):
with mock.patch('neutron.agent.common.utils.execute'):
self.netns_cmd.delete('ns')
self._assert_sudo([], ('delete', 'ns'), use_root_namespace=True)
@mock.patch.object(priv_lib, 'remove_netns')
def test_delete_namespace(self, remove):
self.netns_cmd.delete('ns')
remove.assert_called_once_with('ns')
def test_namespace_exists_use_helper(self):
@mock.patch.object(pyroute2.netns, 'listnetns')
@mock.patch.object(priv_lib, 'list_netns')
def test_namespace_exists_use_helper(self, priv_listnetns, listnetns):
self.config(group='AGENT', use_helper_for_ns_read=True)
retval = '\n'.join(NETNS_SAMPLE)
priv_listnetns.return_value = NETNS_SAMPLE
# need another instance to avoid mocking
netns_cmd = ip_lib.IpNetnsCommand(ip_lib.SubProcessBase())
with mock.patch('neutron.agent.common.utils.execute') as execute:
execute.return_value = retval
self.assertTrue(
netns_cmd.exists('bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb'))
execute.assert_called_once_with(['ip', '-o', 'netns', 'list'],
run_as_root=True,
log_fail_as_error=True)
def test_namespace_doest_not_exist_no_helper(self):
self.assertTrue(
netns_cmd.exists('bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb'))
self.assertEqual(1, priv_listnetns.call_count)
self.assertFalse(listnetns.called)
@mock.patch.object(pyroute2.netns, 'listnetns')
@mock.patch.object(priv_lib, 'list_netns')
def test_namespace_does_not_exist_no_helper(self, priv_listnetns,
listnetns):
self.config(group='AGENT', use_helper_for_ns_read=False)
retval = '\n'.join(NETNS_SAMPLE)
listnetns.return_value = NETNS_SAMPLE
# need another instance to avoid mocking
netns_cmd = ip_lib.IpNetnsCommand(ip_lib.SubProcessBase())
with mock.patch('neutron.agent.common.utils.execute') as execute:
execute.return_value = retval
self.assertFalse(
netns_cmd.exists('bbbbbbbb-1111-2222-3333-bbbbbbbbbbbb'))
execute.assert_called_once_with(['ip', '-o', 'netns', 'list'],
run_as_root=False,
log_fail_as_error=True)
self.assertFalse(
netns_cmd.exists('bbbbbbbb-1111-2222-3333-bbbbbbbbbbbb'))
self.assertEqual(1, listnetns.call_count)
self.assertFalse(priv_listnetns.called)
def test_execute(self):
self.parent.namespace = 'ns'

@ -990,11 +990,11 @@ class IptablesManagerStateFulTestCase(base.BaseTestCase):
self.assertRaises(RuntimeError,
self.iptables._apply_synchronized)
self.iptables.namespace = 'test'
with mock.patch('neutron.agent.linux.ip_lib.IpNetnsCommand.exists',
with mock.patch('neutron.agent.linux.ip_lib.network_namespace_exists',
return_value=True):
self.assertRaises(RuntimeError,
self.iptables._apply_synchronized)
with mock.patch('neutron.agent.linux.ip_lib.IpNetnsCommand.exists',
with mock.patch('neutron.agent.linux.ip_lib.network_namespace_exists',
return_value=False):
self.assertEqual([], self.iptables._apply_synchronized())

@ -363,8 +363,9 @@ class TestNetnsCleanup(base.BaseTestCase):
def test_main(self):
namespaces = ['ns1', 'ns2']
with mock.patch('neutron.agent.linux.ip_lib.IPWrapper') as ip_wrap:
ip_wrap.get_namespaces.return_value = namespaces
with mock.patch('neutron.agent.linux.ip_lib.'
'list_network_namespaces') as listnetns:
listnetns.return_value = namespaces
with mock.patch('time.sleep') as time_sleep:
conf = mock.Mock()
@ -388,15 +389,15 @@ class TestNetnsCleanup(base.BaseTestCase):
[mock.call(conf, 'ns1', False),
mock.call(conf, 'ns2', False)])
ip_wrap.assert_has_calls(
[mock.call.get_namespaces()])
self.assertEqual(1, listnetns.call_count)
time_sleep.assert_called_once_with(2)
def test_main_no_candidates(self):
namespaces = ['ns1', 'ns2']
with mock.patch('neutron.agent.linux.ip_lib.IPWrapper') as ip_wrap:
ip_wrap.get_namespaces.return_value = namespaces
with mock.patch('neutron.agent.linux.ip_lib.'
'list_network_namespaces') as listnetns:
listnetns.return_value = namespaces
with mock.patch('time.sleep') as time_sleep:
conf = mock.Mock()
@ -412,8 +413,7 @@ class TestNetnsCleanup(base.BaseTestCase):
with mock.patch('neutron.common.config.setup_logging'):
util.main()
ip_wrap.assert_has_calls(
[mock.call.get_namespaces()])
self.assertEqual(1, listnetns.call_count)
mocks['eligible_for_deletion'].assert_has_calls(
[mock.call(conf, 'ns1', False),

@ -43,9 +43,12 @@ class TestDebugCommands(base.BaseTestCase):
device_exists_p = mock.patch(
'neutron.agent.linux.ip_lib.device_exists', return_value=False)
device_exists_p.start()
namespace_p = mock.patch(
'neutron.agent.linux.ip_lib.IpNetnsCommand')
namespace_p.start()
namespace_e_p = mock.patch(
'neutron.agent.linux.ip_lib.network_namespace_exists')
namespace_e_p.start()
namespace_d_p = mock.patch(
'neutron.agent.linux.ip_lib.delete_network_namespace')
namespace_d_p.start()
ensure_namespace_p = mock.patch(
'neutron.agent.linux.ip_lib.IPWrapper.ensure_namespace')
ensure_namespace_p.start()

@ -134,7 +134,7 @@ class IptablesDriverTestCase(base.BaseTestCase):
self.v4filter_inst = mock.Mock()
self.v6filter_inst = mock.Mock()
self.namespace_exists_p = mock.patch(
'neutron.agent.linux.ip_lib.IpNetnsCommand.exists')
'neutron.agent.linux.ip_lib.network_namespace_exists')
self.namespace_exists = self.namespace_exists_p.start()
self.snat_ns_name_p = mock.patch(
'neutron.agent.l3.dvr_snat_ns.SnatNamespace.get_snat_ns_name')

Loading…
Cancel
Save