Patch #1309579: wait3 and wait4 were added to the posix module by Chad J. Schroeder.
This was a fair amount of rework of the patch. Refactored test_fork1 so it
could be reused by the new tests for wait3/4. Also made them into new style
unittests (derive from unittest.TestCase).
diff --git a/Lib/test/fork_wait.py b/Lib/test/fork_wait.py
new file mode 100644
index 0000000..5600bdb
--- /dev/null
+++ b/Lib/test/fork_wait.py
@@ -0,0 +1,71 @@
+"""This test case provides support for checking forking and wait behavior.
+
+To test different wait behavior, overrise the wait_impl method.
+
+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, unittest
+from test.test_support import TestSkipped
+
+LONGSLEEP = 2
+SHORTSLEEP = 0.5
+NUM_THREADS = 4
+
+class ForkWait(unittest.TestCase):
+
+ def setUp(self):
+ self.alive = {}
+ self.stop = 0
+
+ def f(self, id):
+ while not self.stop:
+ self.alive[id] = os.getpid()
+ try:
+ time.sleep(SHORTSLEEP)
+ except IOError:
+ pass
+
+ def wait_impl(self, cpid):
+ spid, status = os.waitpid(cpid, 0)
+ self.assertEquals(spid, cpid)
+ self.assertEquals(status, 0, "cause = %d, exit = %d" % (status&0xff, status>>8))
+
+ def test_wait(self):
+ for i in range(NUM_THREADS):
+ thread.start_new(self.f, (i,))
+
+ time.sleep(LONGSLEEP)
+
+ a = self.alive.keys()
+ a.sort()
+ self.assertEquals(a, range(NUM_THREADS))
+
+ prefork_lives = self.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 self.alive:
+ if self.alive[key] != prefork_lives[key]:
+ n += 1
+ os._exit(n)
+ else:
+ # Parent
+ self.wait_impl(cpid)
+ # Tell threads to die
+ self.stop = 1
+ time.sleep(2*SHORTSLEEP) # Wait for threads to die
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))
-main()
+def test_main():
+ run_unittest(ForkTest)
+
+if __name__ == "__main__":
+ test_main()
diff --git a/Lib/test/test_wait3.py b/Lib/test/test_wait3.py
new file mode 100644
index 0000000..a1cbd7b
--- /dev/null
+++ b/Lib/test/test_wait3.py
@@ -0,0 +1,32 @@
+"""This test checks for correct wait3() behavior.
+"""
+
+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_wait3"
+
+try:
+ os.wait3
+except AttributeError:
+ raise TestSkipped, "os.wait3 not defined -- skipping test_wait3"
+
+class Wait3Test(ForkWait):
+ def wait_impl(self, cpid):
+ while 1:
+ spid, status, rusage = os.wait3(0)
+ if spid == cpid:
+ break
+ self.assertEqual(spid, cpid)
+ self.assertEqual(status, 0, "cause = %d, exit = %d" % (status&0xff, status>>8))
+ self.assertTrue(rusage)
+
+def test_main():
+ run_unittest(Wait3Test)
+
+if __name__ == "__main__":
+ test_main()
diff --git a/Lib/test/test_wait4.py b/Lib/test/test_wait4.py
new file mode 100644
index 0000000..027e5c3
--- /dev/null
+++ b/Lib/test/test_wait4.py
@@ -0,0 +1,29 @@
+"""This test checks for correct wait4() behavior.
+"""
+
+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_wait4"
+
+try:
+ os.wait4
+except AttributeError:
+ raise TestSkipped, "os.wait4 not defined -- skipping test_wait4"
+
+class Wait4Test(ForkWait):
+ def wait_impl(self, cpid):
+ spid, status, rusage = os.wait4(cpid, 0)
+ self.assertEqual(spid, cpid)
+ self.assertEqual(status, 0, "cause = %d, exit = %d" % (status&0xff, status>>8))
+ self.assertTrue(rusage)
+
+def test_main():
+ run_unittest(Wait4Test)
+
+if __name__ == "__main__":
+ test_main()