Implement "find_parent_pid" natively

Use Python native libraries instead of shell calls.

Closes-Bug: #1900268

Change-Id: Iaad9e2b3374b5c81f83c780d306685773af719ed

neutron/agent/linux/utils.py

@@ -33,6 +33,7 @@ from oslo_rootwrap import client
 from oslo_utils import encodeutils
 from oslo_utils import excutils
 from oslo_utils import fileutils
+import psutil
 
 from neutron._i18n import _
 from neutron.agent.linux import xenapi_root_helper
@@ -186,18 +187,10 @@ def find_parent_pid(pid):
     If the pid doesn't exist in the system, this function will return
     None
     """
-    try:
+    process = psutil.Process(pid=int(pid))
-        ppid = execute(['ps', '-o', 'ppid=', pid],
+    if process:
-                       log_fail_as_error=False)
+        return str(process.parent().pid)
-    except exceptions.ProcessExecutionError as e:
+    return None
-        # Unexpected errors are the responsibility of the caller
-        with excutils.save_and_reraise_exception() as ctxt:
-            # Exception has already been logged by execute
-            no_such_pid = e.returncode == 1
-            if no_such_pid:
-                ctxt.reraise = False
-                return
-    return ppid.strip()
 
 
 def get_process_count_by_name(name):

neutron/tests/functional/agent/linux/test_utils.py

@@ -13,6 +13,8 @@
 #    under the License.
 
 import functools
+import os
+import signal
 
 from neutron.agent.common import async_process
 from neutron.agent.linux import utils
@@ -91,3 +93,28 @@ class TestGetRootHelperChildPid(functional_base.BaseSudoTestCase):
         with open('/proc/%s/cmdline' % child_pid, 'r') as f_proc_cmdline:
             cmdline = f_proc_cmdline.readline().split('\0')[0]
         self.assertIn('bash', cmdline)
+
+
+class TestFindParentPid(functional_base.BaseSudoTestCase):
+
+    def _stop_process(self, process):
+        process.stop(kill_signal=signal.SIGKILL)
+
+    def _test_process(self, run_as_root):
+        test_pid = str(os.getppid())
+        cmd = ['bash', '-c', '(sleep 10)']
+        proc = async_process.AsyncProcess(cmd, run_as_root=run_as_root)
+        proc.start()
+        self.addCleanup(self._stop_process, proc)
+        common_utils.wait_until_true(lambda: proc._process.pid,
+                                     sleep=0.5, timeout=10)
+
+        bash_pid = utils.find_parent_pid(proc._process.pid)
+        testcase_pid = utils.find_parent_pid(bash_pid)
+        self.assertEqual(test_pid, testcase_pid)
+
+    def test_root_process(self):
+        self._test_process(run_as_root=True)
+
+    def test_non_root_process(self):
+        self._test_process(run_as_root=False)

neutron/tests/unit/agent/linux/test_utils.py

@@ -188,32 +188,6 @@ class AgentUtilsExecuteEncodeTest(base.BaseTestCase):
         self.assertEqual((str_data, ''), result)
 
 
-class TestFindParentPid(base.BaseTestCase):
-    def setUp(self):
-        super(TestFindParentPid, self).setUp()
-        self.m_execute = mock.patch.object(utils, 'execute').start()
-
-    def test_returns_none_for_no_valid_pid(self):
-        self.m_execute.side_effect = exceptions.ProcessExecutionError(
-            '', returncode=1)
-        self.assertIsNone(utils.find_parent_pid(-1))
-
-    def test_returns_parent_id_for_good_ouput(self):
-        self.m_execute.return_value = '123 \n'
-        self.assertEqual(utils.find_parent_pid(-1), '123')
-
-    def test_raises_exception_returncode_0(self):
-        with testtools.ExpectedException(exceptions.ProcessExecutionError):
-            self.m_execute.side_effect = \
-                exceptions.ProcessExecutionError('', returncode=0)
-            utils.find_parent_pid(-1)
-
-    def test_raises_unknown_exception(self):
-        with testtools.ExpectedException(RuntimeError):
-            self.m_execute.side_effect = RuntimeError()
-            utils.find_parent_pid(-1)
-
-
 class TestFindForkTopParent(base.BaseTestCase):
     def _test_find_fork_top_parent(self, expected=_marker,
                                    find_parent_pid_retvals=None,