diff options
author | Gregory P. Smith <greg@mad-scientist.com> | 2010-03-01 06:18:41 (GMT) |
---|---|---|
committer | Gregory P. Smith <greg@mad-scientist.com> | 2010-03-01 06:18:41 (GMT) |
commit | 24cec9fe07a1f7e408a0ff846c6491e0cf8295c4 (patch) | |
tree | 771ec4efea43ffc13b6dfb71229ec50148a70f8b /Lib/test/test_thread.py | |
parent | c78d79ce30b8303a428f075770c8fcf44cd48a53 (diff) | |
download | cpython-24cec9fe07a1f7e408a0ff846c6491e0cf8295c4.zip cpython-24cec9fe07a1f7e408a0ff846c6491e0cf8295c4.tar.gz cpython-24cec9fe07a1f7e408a0ff846c6491e0cf8295c4.tar.bz2 |
Merged revisions 78527,78550 via svnmerge from
svn+ssh://pythondev@svn.python.org/python/trunk
........
r78527 | gregory.p.smith | 2010-02-28 17:22:39 -0800 (Sun, 28 Feb 2010) | 4 lines
Issue #7242: On Solaris 9 and earlier calling os.fork() from within a
thread could raise an incorrect RuntimeError about not holding the import
lock. The import lock is now reinitialized after fork.
........
r78550 | gregory.p.smith | 2010-02-28 22:01:02 -0800 (Sun, 28 Feb 2010) | 2 lines
Fix test to be skipped on windows.
........
Diffstat (limited to 'Lib/test/test_thread.py')
-rw-r--r-- | Lib/test/test_thread.py | 42 |
1 files changed, 41 insertions, 1 deletions
diff --git a/Lib/test/test_thread.py b/Lib/test/test_thread.py index 70d89fe..2601c36 100644 --- a/Lib/test/test_thread.py +++ b/Lib/test/test_thread.py @@ -4,6 +4,7 @@ import random from test import support import _thread as thread import time +import sys import weakref from test import lock_tests @@ -193,8 +194,47 @@ class LockTests(lock_tests.LockTests): locktype = thread.allocate_lock +class TestForkInThread(unittest.TestCase): + def setUp(self): + self.read_fd, self.write_fd = os.pipe() + + @unittest.skipIf(sys.platform.startswith('win'), + "This test is only appropriate for POSIX-like systems.") + def test_forkinthread(self): + def thread1(): + try: + pid = os.fork() # fork in a thread + except RuntimeError: + os._exit(1) # exit the child + + if pid == 0: # child + try: + os.close(self.read_fd) + os.write(self.write_fd, b"OK") + finally: + os._exit(0) + else: # parent + os.close(self.write_fd) + + thread.start_new_thread(thread1, ()) + self.assertEqual(os.read(self.read_fd, 2), b"OK", + "Unable to fork() in thread") + + def tearDown(self): + try: + os.close(self.read_fd) + except OSError: + pass + + try: + os.close(self.write_fd) + except OSError: + pass + + def test_main(): - support.run_unittest(ThreadRunningTests, BarrierTest, LockTests) + support.run_unittest(ThreadRunningTests, BarrierTest, LockTests, + TestForkInThread) if __name__ == "__main__": test_main() |