Changeset 391 for python/trunk/Lib/test/test_thread.py

Timestamp:

Mar 19, 2014, 11:31:01 PM (11 years ago)

Author:

dmik

Message:

python: Merge vendor 2.7.6 to trunk.

Location:

python/trunk

Files:

• . (modified) (1 prop)
• Lib/test/test_thread.py (modified) (6 diffs)

python/trunk/Lib/test/test_thread.py

@@ -3 +3 @@
 import random
 from test import test_support
-import thread
+thread = test_support.import_module('thread')
 import time
 import sys
+import weakref
 
 from test import lock_tests
@@ -65 +66 @@
     def test_stack_size(self):
         # Various stack size tests.
-        self.assertEquals(thread.stack_size(), 0, "intial stack size is not 0")
+        self.assertEqual(thread.stack_size(), 0, "initial stack size is not 0")
 
         thread.stack_size(0)
-        self.assertEquals(thread.stack_size(), 0, "stack_size not reset to default")
+        self.assertEqual(thread.stack_size(), 0, "stack_size not reset to default")
 
         if os.name not in ("nt", "os2", "posix"):
@@ -88 +89 @@
             for tss in (262144, 0x100000, 0):
                 thread.stack_size(tss)
-                self.assertEquals(thread.stack_size(), tss, fail_msg % tss)
+                self.assertEqual(thread.stack_size(), tss, fail_msg % tss)
                 verbose_print("successfully set stack_size(%d)" % tss)
 
@@ -103 +104 @@
 
             thread.stack_size(0)
+
+    def test__count(self):
+        # Test the _count() function.
+        orig = thread._count()
+        mut = thread.allocate_lock()
+        mut.acquire()
+        started = []
+        def task():
+            started.append(None)
+            mut.acquire()
+            mut.release()
+        thread.start_new_thread(task, ())
+        while not started:
+            time.sleep(0.01)
+        self.assertEqual(thread._count(), orig + 1)
+        # Allow the task to finish.
+        mut.release()
+        # The only reliable way to be sure that the thread ended from the
+        # interpreter's point of view is to wait for the function object to be
+        # destroyed.
+        done = []
+        wr = weakref.ref(task, lambda _: done.append(None))
+        del task
+        while not done:
+            time.sleep(0.01)
+        self.assertEqual(thread._count(), orig)
+
+    def test_save_exception_state_on_error(self):
+        # See issue #14474
+        def task():
+            started.release()
+            raise SyntaxError
+        def mywrite(self, *args):
+            try:
+                raise ValueError
+            except ValueError:
+                pass
+            real_write(self, *args)
+        c = thread._count()
+        started = thread.allocate_lock()
+        with test_support.captured_output("stderr") as stderr:
+            real_write = stderr.write
+            stderr.write = mywrite
+            started.acquire()
+            thread.start_new_thread(task, ())
+            started.acquire()
+            while thread._count() > c:
+                time.sleep(0.01)
+        self.assertIn("Traceback", stderr.getvalue())
 
 
@@ -175 +225 @@
         self.read_fd, self.write_fd = os.pipe()
 
-    def _test_forkinthread(self):
+    @unittest.skipIf(sys.platform.startswith('win'),
+                     "This test is only appropriate for POSIX-like systems.")
+    @test_support.reap_threads
+    def test_forkinthread(self):
         def thread1():
             try:
@@ -193 +246 @@
                          "Unable to fork() in thread")
 
-    if not sys.platform.startswith('win'):
-        test_forkinthread = _test_forkinthread
-
     def tearDown(self):
         try: