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

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:[<inode>]"
-    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

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)