From 55874d06b1ce80b44356310c930f80e834d83a8a Mon Sep 17 00:00:00 2001 From: Akihiro Motoki Date: Fri, 18 Jan 2019 06:29:33 +0900 Subject: [PATCH] Change netns tests with oslo.privsep to check netns links It turns out that pyroute2.netns.setns() changes a network namespace of a thread instead of that of a process when it is called in a thread [1]. What we actually would like to check in test_in_namespace test is whether operations against a network namespace work with oslo.privsep expectedly. There is no need to check namespace inode. This commit changes test_in_namespace test to check a list of network devices in a namespace to check netns operation works correctly. What the new test does are: - create a network namespace for testing - create a veth pair and move one of them to the network namespace - call oslo.privsep entrypoint function to retrieve a list of network devices inside the netns [1] http://lists.openstack.org/pipermail/openstack-discuss/2019-January/001761.html Closes-Bug: #1811506 Change-Id: Ie5b238f1df707ea3ce50b5711ff791bac2681a2f --- .../privileged/tests/functional/utils.py | 23 +++++----- .../tests/functional/privileged/test_utils.py | 43 ++++++++++++------- 2 files changed, 40 insertions(+), 26 deletions(-) diff --git a/neutron_fwaas/privileged/tests/functional/utils.py b/neutron_fwaas/privileged/tests/functional/utils.py index ea3a872f1..2e58a27d8 100644 --- a/neutron_fwaas/privileged/tests/functional/utils.py +++ b/neutron_fwaas/privileged/tests/functional/utils.py @@ -13,24 +13,27 @@ # License for the specific language governing permissions and limitations # under the License. -import os -import re +import pyroute2 from neutron_fwaas import privileged from neutron_fwaas.privileged import utils -def get_my_netns_inode(): - link = os.readlink(utils.PROCESS_NETNS) +def _get_ifname(link): + attr_dict = dict(link['attrs']) + return attr_dict['IFLA_IFNAME'] - # NOTE(cby): link respects the format "net:[]" - return int(re.match('net:\[(\d+)\]', link).group(1)) + +def list_interface_names(): + iproute = pyroute2.IPRoute() + result = iproute.get_links() + return [_get_ifname(link) for link in result] @privileged.default.entrypoint -def get_in_namespace_netns_inodes(namespace): - before = get_my_netns_inode() +def get_in_namespace_interfaces(namespace): + before = list_interface_names() with utils.in_namespace(namespace): - inside = get_my_netns_inode() - after = get_my_netns_inode() + inside = list_interface_names() + after = list_interface_names() return before, inside, after diff --git a/neutron_fwaas/tests/functional/privileged/test_utils.py b/neutron_fwaas/tests/functional/privileged/test_utils.py index 4c0222fe4..3d62d4473 100644 --- a/neutron_fwaas/tests/functional/privileged/test_utils.py +++ b/neutron_fwaas/tests/functional/privileged/test_utils.py @@ -13,30 +13,41 @@ # License for the specific language governing permissions and limitations # under the License. -import os -import unittest - +from neutron.agent.linux import ip_lib +from neutron.common import utils as neutron_utils from neutron.tests.common import net_helpers from neutron.tests.functional import base from neutron_fwaas.privileged.tests.functional import utils -def get_netns_inode(namespace): - return os.stat('/var/run/netns/%s' % namespace).st_ino - - class InNamespaceTest(base.BaseSudoTestCase): - @unittest.skip('Temporarily skipped until a fix against oslo.privsep 1.31') + def setUp(self): + super(InNamespaceTest, self).setUp() + self.namespace = self.useFixture(net_helpers.NamespaceFixture()).name + + ip = ip_lib.IPWrapper() + root_dev_name = neutron_utils.get_rand_device_name() + netns_dev_name = neutron_utils.get_rand_device_name() + self.root_dev, self.netns_dev = ip.add_veth( + root_dev_name, netns_dev_name, namespace2=self.namespace) + self.addCleanup(self.root_dev.link.delete) + def test_in_namespace(self): - namespace = self.useFixture(net_helpers.NamespaceFixture()).name - expected = get_netns_inode(namespace) - before, observed, after = utils.get_in_namespace_netns_inodes( - namespace) - self.assertEqual(expected, observed) - self.assertEqual(before, after) + before, observed, after = utils.get_in_namespace_interfaces( + self.namespace) + expected = ['lo', self.netns_dev.name] + self.assertItemsEqual(expected, observed) + # Other tests can create/delete devices, so we just checks + # self.root_dev_name is included in the root namespace result. + self.assertIn(self.root_dev.name, before) + self.assertIn(self.root_dev.name, after) def test_in_no_namespace(self): - inodes = utils.get_in_namespace_netns_inodes(None) - self.assertEqual(1, len(set(inodes))) + before, observed, after = utils.get_in_namespace_interfaces(None) + # Other tests can create/delete devices, so we just checks + # self.root_dev_name is included in the root namespace result. + self.assertIn(self.root_dev.name, observed) + self.assertIn(self.root_dev.name, before) + self.assertIn(self.root_dev.name, after)