From 49fd7fa4431da299196d74087df4a04f99f9c46f Mon Sep 17 00:00:00 2001
From: Thomas Wouters <thomas@python.org>
Date: Fri, 21 Apr 2006 10:40:58 +0000
Subject: Merge p3yk branch with the trunk up to revision 45595. This breaks a
 fair number of tests, all because of the codecs/_multibytecodecs issue
 described here (it's not a Py3K issue, just something Py3K discovers):
 http://mail.python.org/pipermail/python-dev/2006-April/064051.html

Hye-Shik Chang promised to look for a fix, so no need to fix it here. The
tests that are expected to break are:

test_codecencodings_cn
test_codecencodings_hk
test_codecencodings_jp
test_codecencodings_kr
test_codecencodings_tw
test_codecs
test_multibytecodec

This merge fixes an actual test failure (test_weakref) in this branch,
though, so I believe merging is the right thing to do anyway.
---
 Lib/test/fork_wait.py | 71 +++++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 71 insertions(+)
 create mode 100644 Lib/test/fork_wait.py

(limited to 'Lib/test/fork_wait.py')

diff --git a/Lib/test/fork_wait.py b/Lib/test/fork_wait.py
new file mode 100644
index 0000000000..5600bdbc51
--- /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
-- 
cgit v1.2.1