[Python-checkins] cpython: Try doing a raw test of os.fork()/os.kill().
richard.oudkerk
python-checkins at python.org
Thu Oct 17 15:33:02 CEST 2013
http://hg.python.org/cpython/rev/9558e9360afc
changeset: 86401:9558e9360afc
parent: 86399:9cd88b39ef62
user: Richard Oudkerk <shibturn at gmail.com>
date: Thu Oct 17 14:24:06 2013 +0100
summary:
Try doing a raw test of os.fork()/os.kill().
files:
Lib/test/_test_multiprocessing.py | 41 ++++++++++++++++--
1 files changed, 36 insertions(+), 5 deletions(-)
diff --git a/Lib/test/_test_multiprocessing.py b/Lib/test/_test_multiprocessing.py
--- a/Lib/test/_test_multiprocessing.py
+++ b/Lib/test/_test_multiprocessing.py
@@ -273,10 +273,11 @@
@classmethod
def _test_terminate(cls):
- print('signal.getsignal(SIGTERM) =', signal.getsignal(signal.SIGTERM))
- print('starting sleep')
+ print('signal.getsignal(SIGTERM) =',
+ signal.getsignal(signal.SIGTERM), file=sys.stderr)
+ print('starting sleep', file=sys.stderr)
time.sleep(100)
- print('finished sleep')
+ print('finished sleep', file=sys.stderr)
def test_terminate(self):
if self.TYPE == 'threads':
@@ -314,11 +315,12 @@
try:
signal.alarm(10)
self.assertEqual(join(), None)
- signal.alarm(0)
except RuntimeError:
- print('os.waitpid() =', os.waitpid(p.pid, os.WNOHANG))
+ print('os.waitpid() =',
+ os.waitpid(p.pid, os.WNOHANG), file=sys.stderr)
raise
finally:
+ signal.alarm(0)
signal.signal(signal.SIGALRM, old_handler)
else:
self.assertEqual(join(), None)
@@ -333,6 +335,35 @@
# XXX sometimes get p.exitcode == 0 on Windows ...
#self.assertEqual(p.exitcode, -signal.SIGTERM)
+ @unittest.skipIf(WIN32, 'Unix only')
+ def test_sigterm(self):
+ # A test for the Gentoo build bot which does not directly use
+ # multiprocessing. Start and terminate child processes.
+ if self.TYPE != 'processes':
+ return
+ for i in range(10):
+ pid = os.fork()
+ if pid == 0:
+ try:
+ print('sleeping', file=sys.stderr)
+ time.sleep(100)
+ print('waking', file=sys.stderr)
+ finally:
+ sys.stderr.flush()
+ os._exit(0)
+ else:
+ os.kill(pid, signal.SIGTERM)
+ def handler(*args):
+ raise RuntimeError('join took too long: %s' % p)
+ old_handler = signal.signal(signal.SIGALRM, handler)
+ try:
+ signal.alarm(10)
+ pid_status = os.waitpid(pid, 0)
+ self.assertEqual(pid_status[0], pid)
+ finally:
+ signal.alarm(0)
+ signal.signal(signal.SIGALRM, old_handler)
+
def test_cpu_count(self):
try:
cpus = multiprocessing.cpu_count()
--
Repository URL: http://hg.python.org/cpython
More information about the Python-checkins
mailing list