changeset:   76637:1255cac63dfc
user:        Victor Stinner <victor.stinner@gmail.com>
date:        Sun Apr 29 19:46:19 2012 +0200
files:       Lib/test/test_time.py
description:
Issue #14428: Rewrite test_process_time_threads() test

time.process_time() is CPU time, not a number of seconds.


diff -r cee5cb877631 -r 1255cac63dfc Lib/test/test_time.py
--- a/Lib/test/test_time.py	Sun Apr 29 18:35:56 2012 +0100
+++ b/Lib/test/test_time.py	Sun Apr 29 19:46:19 2012 +0200
@@ -392,20 +392,53 @@
     @unittest.skipUnless(threading,
                          'need threading')
     def test_process_time_threads(self):
-        class BusyThread(threading.Thread):
+        def factorial(n):
+            if n >= 2:
+                return n * factorial(n-1)
+            else:
+                return 1
+
+        def use_cpu(n, loops):
+            for loop in range(loops):
+                factorial(n)
+
+        class FactorialThread(threading.Thread):
+            def __init__(self, n, loops):
+                threading.Thread.__init__(self)
+                self.n = n
+                self.loops = loops
+
             def run(self):
-                while not self.stop:
-                    pass
+                use_cpu(self.n, self.loops)
 
-        thread = BusyThread()
-        thread.stop = False
+        # Calibrate use_cpu() to use at least 1 ms of system time
+        n = 50
+        loops = 1
+        resolution = time.get_clock_info('process_time').resolution
+        min_rdt = max(resolution, 0.001)
+        while 1:
+            rt1 = time.time()
+            t1 = time.process_time()
+            use_cpu(n, loops)
+            t2 = time.process_time()
+            rt2 = time.time()
+            rdt = rt2 - rt1
+            if rdt >= min_rdt:
+                break
+            loops *= 2
+        busy = t2 - t1
+
+        # Ensure that time.process_time() includes the CPU time of all threads
+        thread = FactorialThread(n, loops)
         t1 = time.process_time()
         thread.start()
-        time.sleep(0.2)
+        # Use sleep() instead of thread.join() because thread.join() time may
+        # be included in time.process_time() depending on its implementation
+        time.sleep(rdt * 2)
         t2 = time.process_time()
         thread.stop = True
         thread.join()
-        self.assertGreater(t2 - t1, 0.1)
+        self.assertGreaterEqual(t2 - t1, busy)
 
     @unittest.skipUnless(hasattr(time, 'monotonic'),
                          'need time.monotonic')