[Python-checkins] bpo-35537: Rewrite setsid test for os.posix_spawn (GH-11721)

[Miss Islington (bot)]

https://github.com/python/cpython/commit/e696b15a62dd0c85fe6ed3c9698b5f889c0bb1b3
commit: e696b15a62dd0c85fe6ed3c9698b5f889c0bb1b3
branch: 3.8
author: Miss Islington (bot) <31488909+miss-islington at users.noreply.github.com>
committer: GitHub <noreply at github.com>
date: 2019-06-14T10:49:22-07:00
summary:

bpo-35537: Rewrite setsid test for os.posix_spawn (GH-11721)


bpo-35537, bpo-35876: Fix also test_start_new_session() of
test_subprocess: use os.getsid() rather than os.getpgid().
(cherry picked from commit 5884043252473ac733aba1d3251d4debe72511e5)

Co-authored-by: Victor Stinner <vstinner at redhat.com>

files:
M Lib/test/test_posix.py
M Lib/test/test_subprocess.py

diff --git a/Lib/test/test_posix.py b/Lib/test/test_posix.py
index 0f07a8f2e68d..afa1398d4edc 100644
--- a/Lib/test/test_posix.py
+++ b/Lib/test/test_posix.py
@@ -1639,23 +1639,35 @@ def test_setsigmask_wrong_type(self):
                             os.environ, setsigmask=[signal.NSIG,
                                                     signal.NSIG+1])
 
-    @unittest.skipIf(True,
-                     "FIXME: bpo-35537: test fails is setsid is supported")
-    def test_start_new_session(self):
-        # For code coverage of calling setsid().  We don't care if we get an
-        # EPERM error from it depending on the test execution environment, that
-        # still indicates that it was called.
-        code = "import os; print(os.getpgid(os.getpid()))"
+    def test_setsid(self):
+        rfd, wfd = os.pipe()
+        self.addCleanup(os.close, rfd)
         try:
-            self.spawn_func(sys.executable,
-                            [sys.executable, "-c", code],
-                            os.environ, setsid=True)
-        except NotImplementedError as exc:
-            self.skipTest("setsid is not supported: %s" % exc)
-        else:
-            parent_pgid = os.getpgid(os.getpid())
-            child_pgid = int(output)
-            self.assertNotEqual(parent_pgid, child_pgid)
+            os.set_inheritable(wfd, True)
+
+            code = textwrap.dedent(f"""
+                import os
+                fd = {wfd}
+                sid = os.getsid(0)
+                os.write(fd, str(sid).encode())
+            """)
+
+            try:
+                pid = self.spawn_func(sys.executable,
+                                      [sys.executable, "-c", code],
+                                      os.environ, setsid=True)
+            except NotImplementedError as exc:
+                self.skipTest(f"setsid is not supported: {exc!r}")
+            except PermissionError as exc:
+                self.skipTest(f"setsid failed with: {exc!r}")
+        finally:
+            os.close(wfd)
+
+        self.assertEqual(os.waitpid(pid, 0), (pid, 0))
+        output = os.read(rfd, 100)
+        child_sid = int(output)
+        parent_sid = os.getsid(os.getpid())
+        self.assertNotEqual(parent_sid, child_sid)
 
     @unittest.skipUnless(hasattr(signal, 'pthread_sigmask'),
                          'need signal.pthread_sigmask()')
diff --git a/Lib/test/test_subprocess.py b/Lib/test/test_subprocess.py
index fca3ed62099b..97d21904b9ce 100644
--- a/Lib/test/test_subprocess.py
+++ b/Lib/test/test_subprocess.py
@@ -1705,16 +1705,15 @@ def test_start_new_session(self):
         # still indicates that it was called.
         try:
             output = subprocess.check_output(
-                    [sys.executable, "-c",
-                     "import os; print(os.getpgid(os.getpid()))"],
+                    [sys.executable, "-c", "import os; print(os.getsid(0))"],
                     start_new_session=True)
         except OSError as e:
             if e.errno != errno.EPERM:
                 raise
         else:
-            parent_pgid = os.getpgid(os.getpid())
-            child_pgid = int(output)
-            self.assertNotEqual(parent_pgid, child_pgid)
+            parent_sid = os.getsid(0)
+            child_sid = int(output)
+            self.assertNotEqual(parent_sid, child_sid)
 
     def test_run_abort(self):
         # returncode handles signal termination