summaryrefslogtreecommitdiffstats
path: root/Lib/test/test_posix.py
diff options
context:
space:
mode:
authorPablo Galindo <Pablogsal@gmail.com>2018-09-07 15:44:24 (GMT)
committerGitHub <noreply@github.com>2018-09-07 15:44:24 (GMT)
commit254a4663d8c5970ae2928185c50ebaa6c7e62c80 (patch)
tree30dfeb3927b8693f2ed5317917824c56033a620a /Lib/test/test_posix.py
parent5e922658fb55734bf8b4c6246033ea93af172ff7 (diff)
downloadcpython-254a4663d8c5970ae2928185c50ebaa6c7e62c80.zip
cpython-254a4663d8c5970ae2928185c50ebaa6c7e62c80.tar.gz
cpython-254a4663d8c5970ae2928185c50ebaa6c7e62c80.tar.bz2
bpo-20104: Add flag capabilities to posix_spawn (GH-6693)
Implement the "attributes objects" parameter of `os.posix_spawn` to complete the implementation and fully cover the underlying API.
Diffstat (limited to 'Lib/test/test_posix.py')
-rw-r--r--Lib/test/test_posix.py143
1 files changed, 143 insertions, 0 deletions
diff --git a/Lib/test/test_posix.py b/Lib/test/test_posix.py
index ac18aa3..dafe9c1 100644
--- a/Lib/test/test_posix.py
+++ b/Lib/test/test_posix.py
@@ -8,6 +8,7 @@ posix = support.import_module('posix')
import errno
import sys
+import signal
import time
import os
import platform
@@ -16,6 +17,7 @@ import stat
import tempfile
import unittest
import warnings
+import textwrap
_DUMMY_SYMLINK = os.path.join(tempfile.gettempdir(),
support.TESTFN + '-dummy-symlink')
@@ -1540,6 +1542,147 @@ class TestPosixSpawn(unittest.TestCase):
)
self.assertEqual(os.waitpid(pid, 0), (pid, 0))
+ def test_resetids_explicit_default(self):
+ pid = posix.posix_spawn(
+ sys.executable,
+ [sys.executable, '-c', 'pass'],
+ os.environ,
+ resetids=False
+ )
+ self.assertEqual(os.waitpid(pid, 0), (pid, 0))
+
+ def test_resetids(self):
+ pid = posix.posix_spawn(
+ sys.executable,
+ [sys.executable, '-c', 'pass'],
+ os.environ,
+ resetids=True
+ )
+ self.assertEqual(os.waitpid(pid, 0), (pid, 0))
+
+ def test_resetids_wrong_type(self):
+ with self.assertRaises(TypeError):
+ posix.posix_spawn(sys.executable,
+ [sys.executable, "-c", "pass"],
+ os.environ, resetids=None)
+
+ def test_setpgroup(self):
+ pid = posix.posix_spawn(
+ sys.executable,
+ [sys.executable, '-c', 'pass'],
+ os.environ,
+ setpgroup=os.getpgrp()
+ )
+ self.assertEqual(os.waitpid(pid, 0), (pid, 0))
+
+ def test_setpgroup_wrong_type(self):
+ with self.assertRaises(TypeError):
+ posix.posix_spawn(sys.executable,
+ [sys.executable, "-c", "pass"],
+ os.environ, setpgroup="023")
+
+ @unittest.skipUnless(hasattr(signal, 'pthread_sigmask'),
+ 'need signal.pthread_sigmask()')
+ def test_setsigmask(self):
+ code = textwrap.dedent("""\
+ import _testcapi, signal
+ _testcapi.raise_signal(signal.SIGUSR1)""")
+
+ pid = posix.posix_spawn(
+ sys.executable,
+ [sys.executable, '-c', code],
+ os.environ,
+ setsigmask=[signal.SIGUSR1]
+ )
+ self.assertEqual(os.waitpid(pid, 0), (pid, 0))
+
+ def test_setsigmask_wrong_type(self):
+ with self.assertRaises(TypeError):
+ posix.posix_spawn(sys.executable,
+ [sys.executable, "-c", "pass"],
+ os.environ, setsigmask=34)
+ with self.assertRaises(TypeError):
+ posix.posix_spawn(sys.executable,
+ [sys.executable, "-c", "pass"],
+ os.environ, setsigmask=["j"])
+ with self.assertRaises(ValueError):
+ posix.posix_spawn(sys.executable,
+ [sys.executable, "-c", "pass"],
+ os.environ, setsigmask=[signal.NSIG,
+ signal.NSIG+1])
+
+ @unittest.skipUnless(hasattr(signal, 'pthread_sigmask'),
+ 'need signal.pthread_sigmask()')
+ def test_setsigdef(self):
+ original_handler = signal.signal(signal.SIGUSR1, signal.SIG_IGN)
+ code = textwrap.dedent("""\
+ import _testcapi, signal
+ _testcapi.raise_signal(signal.SIGUSR1)""")
+ try:
+ pid = posix.posix_spawn(
+ sys.executable,
+ [sys.executable, '-c', code],
+ os.environ,
+ setsigdef=[signal.SIGUSR1]
+ )
+ finally:
+ signal.signal(signal.SIGUSR1, original_handler)
+
+ pid2, status = os.waitpid(pid, 0)
+ self.assertEqual(pid2, pid)
+ self.assertTrue(os.WIFSIGNALED(status), status)
+ self.assertEqual(os.WTERMSIG(status), signal.SIGUSR1)
+
+ def test_setsigdef_wrong_type(self):
+ with self.assertRaises(TypeError):
+ posix.posix_spawn(sys.executable,
+ [sys.executable, "-c", "pass"],
+ os.environ, setsigdef=34)
+ with self.assertRaises(TypeError):
+ posix.posix_spawn(sys.executable,
+ [sys.executable, "-c", "pass"],
+ os.environ, setsigdef=["j"])
+ with self.assertRaises(ValueError):
+ posix.posix_spawn(sys.executable,
+ [sys.executable, "-c", "pass"],
+ os.environ, setsigdef=[signal.NSIG, signal.NSIG+1])
+
+ @unittest.skipUnless(hasattr(posix, 'sched_setscheduler'), "can't change scheduler")
+ def test_setscheduler_only_param(self):
+ policy = os.sched_getscheduler(0)
+ priority = os.sched_get_priority_min(policy)
+ code = textwrap.dedent(f"""\
+ import os
+ if os.sched_getscheduler(0) != {policy}:
+ os.exit(101)
+ if os.sched_getparam(0).sched_priority != {priority}:
+ os.exit(102)""")
+ pid = posix.posix_spawn(
+ sys.executable,
+ [sys.executable, '-c', code],
+ os.environ,
+ scheduler=(None, os.sched_param(priority))
+ )
+ self.assertEqual(os.waitpid(pid, 0), (pid, 0))
+
+ @unittest.skipUnless(hasattr(posix, 'sched_setscheduler'), "can't change scheduler")
+ def test_setscheduler_with_policy(self):
+ policy = os.sched_getscheduler(0)
+ priority = os.sched_get_priority_min(policy)
+ code = textwrap.dedent(f"""\
+ import os
+ if os.sched_getscheduler(0) != {policy}:
+ os.exit(101)
+ if os.sched_getparam(0).sched_priority != {priority}:
+ os.exit(102)""")
+ pid = posix.posix_spawn(
+ sys.executable,
+ [sys.executable, '-c', code],
+ os.environ,
+ scheduler=(policy, os.sched_param(priority))
+ )
+ self.assertEqual(os.waitpid(pid, 0), (pid, 0))
+
def test_multiple_file_actions(self):
file_actions = [
(os.POSIX_SPAWN_OPEN, 3, os.path.realpath(__file__), os.O_RDONLY, 0),