summaryrefslogtreecommitdiffstats
path: root/Lib/test/test_fork1.py
diff options
context:
space:
mode:
Diffstat (limited to 'Lib/test/test_fork1.py')
-rw-r--r--Lib/test/test_fork1.py76
1 files changed, 12 insertions, 64 deletions
diff --git a/Lib/test/test_fork1.py b/Lib/test/test_fork1.py
index aca7a84..cba5fc7 100644
--- a/Lib/test/test_fork1.py
+++ b/Lib/test/test_fork1.py
@@ -1,75 +1,23 @@
"""This test checks for correct fork() behavior.
-
-We want fork1() semantics -- only the forking thread survives in the
-child after a fork().
-
-On some systems (e.g. Solaris without posix threads) we find that all
-active threads survive in the child after a fork(); this is an error.
-
-While BeOS doesn't officially support fork and native threading in
-the same application, the present example should work just fine. DC
"""
-import os, sys, time, thread
-from test.test_support import verify, verbose, TestSkipped
+import os
+from test.fork_wait import ForkWait
+from test.test_support import TestSkipped, run_unittest
try:
os.fork
except AttributeError:
raise TestSkipped, "os.fork not defined -- skipping test_fork1"
-LONGSLEEP = 2
-
-SHORTSLEEP = 0.5
-
-NUM_THREADS = 4
-
-alive = {}
-
-stop = 0
-
-def f(id):
- while not stop:
- alive[id] = os.getpid()
- try:
- time.sleep(SHORTSLEEP)
- except IOError:
- pass
-
-def main():
- for i in range(NUM_THREADS):
- thread.start_new(f, (i,))
-
- time.sleep(LONGSLEEP)
-
- a = alive.keys()
- a.sort()
- verify(a == range(NUM_THREADS))
-
- prefork_lives = alive.copy()
-
- if sys.platform in ['unixware7']:
- cpid = os.fork1()
- else:
- cpid = os.fork()
-
- if cpid == 0:
- # Child
- time.sleep(LONGSLEEP)
- n = 0
- for key in alive.keys():
- if alive[key] != prefork_lives[key]:
- n = n+1
- os._exit(n)
- else:
- # Parent
+class ForkTest(ForkWait):
+ def wait_impl(self, cpid):
spid, status = os.waitpid(cpid, 0)
- verify(spid == cpid)
- verify(status == 0,
- "cause = %d, exit = %d" % (status&0xff, status>>8) )
- global stop
- # Tell threads to die
- stop = 1
- time.sleep(2*SHORTSLEEP) # Wait for threads to die
+ self.assertEqual(spid, cpid)
+ self.assertEqual(status, 0, "cause = %d, exit = %d" % (status&0xff, status>>8))
+
+def test_main():
+ run_unittest(ForkTest)
-main()
+if __name__ == "__main__":
+ test_main()