Merge "Check the namespace is ready in test_mtu_update tests" into stable/rocky

This commit is contained in:
Zuul 2019-09-18 18:54:01 +00:00 committed by Gerrit Code Review
commit de7c225670
6 changed files with 57 additions and 10 deletions

View File

@ -1123,14 +1123,24 @@ def list_network_namespaces(**kwargs):
return netns.listnetns(**kwargs)
def network_namespace_exists(namespace, **kwargs):
def network_namespace_exists(namespace, try_is_ready=False, **kwargs):
"""Check if a network namespace exists.
:param namespace: The name of the namespace to check
:param try_is_ready: Try to open the namespace to know if the namespace
is ready to be operated.
:param kwargs: Callers add any filters they use as kwargs
"""
output = list_network_namespaces(**kwargs)
return namespace in output
if not try_is_ready:
output = list_network_namespaces(**kwargs)
return namespace in output
try:
privileged.open_namespace(namespace)
return True
except (RuntimeError, OSError):
pass
return False
def ensure_device_is_ready(device_name, namespace=None):

View File

@ -131,6 +131,13 @@ def _get_iproute(namespace):
return pyroute2.IPRoute()
@privileged.default.entrypoint
def open_namespace(namespace):
"""Open namespace to test if the namespace is ready to be manipulated"""
with pyroute2.NetNS(namespace, flags=0):
pass
def _translate_ip_device_exception(e, device=None, namespace=None):
if e.code == errno.ENODEV:
raise NetworkInterfaceNotFound(device=device, namespace=namespace)

View File

@ -18,6 +18,7 @@ import os
from oslo_config import cfg
from oslo_log import log as logging
from neutron.agent.linux import ip_lib
from neutron.common import utils as common_utils
from neutron.conf.agent import common as config
from neutron.tests import base as tests_base
@ -121,3 +122,8 @@ class BaseFullStackTestCase(testlib_api.MySQLTestCaseMixin,
exception=RuntimeError("Could not ping the other VM, "
"re-starting %s leads to network "
"disruption" % agent_names))
def assert_namespace_exists(self, ns_name):
common_utils.wait_until_true(
lambda: ip_lib.network_namespace_exists(ns_name,
try_is_ready=True))

View File

@ -107,8 +107,9 @@ class TestDhcpAgentNoHA(BaseDhcpAgentTest):
namespace = cmd._get_namespace_name(
self.network['id'],
suffix=self.environment.hosts[0].dhcp_agent.get_namespace_suffix())
ip = ip_lib.IPWrapper(namespace)
self.assert_namespace_exists(namespace)
ip = ip_lib.IPWrapper(namespace)
devices = ip.get_devices()
self.assertEqual(1, len(devices))

View File

@ -92,10 +92,6 @@ class TestLegacyL3Agent(TestL3Agent):
def _get_namespace(self, router_id):
return namespaces.build_ns_name(namespaces.NS_PREFIX, router_id)
def _assert_namespace_exists(self, 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()
@ -108,7 +104,7 @@ class TestLegacyL3Agent(TestL3Agent):
namespace = "%s@%s" % (
self._get_namespace(router['id']),
self.environment.hosts[0].l3_agent.get_namespace_suffix(), )
self._assert_namespace_exists(namespace)
self.assert_namespace_exists(namespace)
def test_mtu_update(self):
tenant_id = uuidutils.generate_uuid()
@ -122,7 +118,7 @@ class TestLegacyL3Agent(TestL3Agent):
namespace = "%s@%s" % (
self._get_namespace(router['id']),
self.environment.hosts[0].l3_agent.get_namespace_suffix(), )
self._assert_namespace_exists(namespace)
self.assert_namespace_exists(namespace)
ip = ip_lib.IPWrapper(namespace)
common_utils.wait_until_true(lambda: ip.get_devices())

View File

@ -21,6 +21,7 @@ from neutron_lib.utils import net
from oslo_config import cfg
from oslo_log import log as logging
from oslo_utils import importutils
from oslo_utils import uuidutils
import testtools
from neutron.agent.linux import ip_lib
@ -620,3 +621,29 @@ class TestSetIpNonlocalBind(functional_base.BaseSudoTestCase):
self.assertFalse(failed)
self.assertEqual(expected, observed)
class NamespaceTestCase(functional_base.BaseSudoTestCase):
def setUp(self):
super(NamespaceTestCase, self).setUp()
self.namespace = 'test_ns_' + uuidutils.generate_uuid()
ip_lib.create_network_namespace(self.namespace)
self.addCleanup(self._delete_namespace)
def _delete_namespace(self):
ip_lib.delete_network_namespace(self.namespace)
def test_network_namespace_exists_ns_exists(self):
self.assertTrue(ip_lib.network_namespace_exists(self.namespace))
def test_network_namespace_exists_ns_doesnt_exists(self):
self.assertFalse(ip_lib.network_namespace_exists('another_ns'))
def test_network_namespace_exists_ns_exists_try_is_ready(self):
self.assertTrue(ip_lib.network_namespace_exists(self.namespace,
try_is_ready=True))
def test_network_namespace_exists_ns_doesnt_exists_try_is_ready(self):
self.assertFalse(ip_lib.network_namespace_exists('another_ns',
try_is_ready=True))