Check the namespace is ready in test_mtu_update tests
Conflicts:
neutron/tests/fullstack/base.py
neutron/privileged/agent/linux/ip_lib.py
neutron/tests/fullstack/base.py
Change-Id: Ifdeac05b485a079e2288b413829bce0e39aef6fd
Closes-Bug: #1820865
(cherry picked from commit 434b29a2cb
)
This commit is contained in:
parent
28f3e371aa
commit
261880aea5
|
@ -1059,14 +1059,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):
|
||||
|
|
|
@ -102,6 +102,13 @@ def _run_iproute(command, device, namespace, **kwargs):
|
|||
raise
|
||||
|
||||
|
||||
@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
|
||||
|
||||
|
||||
@privileged.default.entrypoint
|
||||
def add_neigh_entry(ip_version, ip_address, mac_address, device, namespace,
|
||||
**kwargs):
|
||||
|
|
|
@ -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
|
||||
|
@ -118,3 +119,8 @@ class BaseFullStackTestCase(testlib_api.MySQLTestCaseMixin,
|
|||
timeout=count * (ping_timeout + 1),
|
||||
exception=RuntimeError("Could not ping the other VM, L2 agent "
|
||||
"restart leads to network disruption"))
|
||||
|
||||
def assert_namespace_exists(self, ns_name):
|
||||
common_utils.wait_until_true(
|
||||
lambda: ip_lib.network_namespace_exists(ns_name,
|
||||
try_is_ready=True))
|
||||
|
|
|
@ -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))
|
||||
|
||||
|
|
|
@ -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())
|
||||
|
|
|
@ -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
|
||||
|
@ -317,3 +318,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))
|
||||
|
|
Loading…
Reference in New Issue