[3.6] bpo-31234: Join threads in tests (#3589)

* bpo-31234: Join threads in tests (#3572)

Call thread.join() on threads to prevent the "dangling threads"
warning.

(cherry picked from commit 18e95b4)

* bpo-31234: Join threads in test_hashlib (#3573)

* bpo-31234: Join threads in test_hashlib

Use thread.join() to wait until the parallel hash tasks complete
rather than using events. Calling thread.join() prevent "dangling
thread" warnings.

* test_hashlib: minor PEP 8 coding style fixes

(cherry picked from commit 8dcf22f)

* bpo-31234: Join threads in test_threading (#3579)

Call thread.join() to prevent the "dangling thread" warning.

(cherry picked from commit b8c7be2)

* bpo-31234: Join threads in test_queue (#3586)

Call thread.join() to prevent the "dangling thread" warning.

(cherry picked from commit 167cbde)

* bpo-31234: Join timers in test_threading (#3598)

Call the .join() method of threading.Timer timers to prevent the
"threading_cleanup() failed to cleanup 1 threads" warning.

(cherry picked from commit da3e5cf)

Author: vstinner

Lib/test/test_concurrent_futures.py

@@ -776,6 +776,7 @@ def notification():
         t.start()
 
         self.assertEqual(f1.result(timeout=5), 42)
+        t.join()
 
     def test_result_with_cancel(self):
         # TODO([email protected]): This test is timing dependent.
@@ -789,6 +790,7 @@ def notification():
         t.start()
 
         self.assertRaises(futures.CancelledError, f1.result, timeout=5)
+        t.join()
 
     def test_exception_with_timeout(self):
         self.assertRaises(futures.TimeoutError,
@@ -817,6 +819,7 @@ def notification():
         t.start()
 
         self.assertTrue(isinstance(f1.exception(timeout=5), OSError))
+        t.join()
 
 @test.support.reap_threads
 def test_main():

Lib/test/test_decimal.py

@@ -1614,6 +1614,9 @@ def test_threading(self):
         for sig in Signals[self.decimal]:
             self.assertFalse(DefaultContext.flags[sig])
 
+        th1.join()
+        th2.join()
+
         DefaultContext.prec = save_prec
         DefaultContext.Emax = save_emax
         DefaultContext.Emin = save_emin

Lib/test/test_hashlib.py

@@ -750,28 +750,28 @@ def test_threaded_hashing(self):
         hasher = hashlib.sha1()
         num_threads = 5
         smallest_data = b'swineflu'
-        data = smallest_data*200000
+        data = smallest_data * 200000
         expected_hash = hashlib.sha1(data*num_threads).hexdigest()
 
-        def hash_in_chunks(chunk_size, event):
+        def hash_in_chunks(chunk_size):
             index = 0
             while index < len(data):
-                hasher.update(data[index:index+chunk_size])
+                hasher.update(data[index:index + chunk_size])
                 index += chunk_size
-            event.set()
 
-        events = []
+        threads = []
         for threadnum in range(num_threads):
-            chunk_size = len(data) // (10**threadnum)
+            chunk_size = len(data) // (10 ** threadnum)
             self.assertGreater(chunk_size, 0)
             self.assertEqual(chunk_size % len(smallest_data), 0)
-            event = threading.Event()
-            events.append(event)
-            threading.Thread(target=hash_in_chunks,
-                             args=(chunk_size, event)).start()
-
-        for event in events:
-            event.wait()
+            thread = threading.Thread(target=hash_in_chunks,
+                                      args=(chunk_size,))
+            threads.append(thread)
+
+        for thread in threads:
+            thread.start()
+        for thread in threads:
+            thread.join()
 
         self.assertEqual(expected_hash, hasher.hexdigest())
 

Lib/test/test_queue.py

@@ -46,28 +46,27 @@ def run(self):
 
 class BlockingTestMixin:
 
-    def tearDown(self):
-        self.t = None
-
     def do_blocking_test(self, block_func, block_args, trigger_func, trigger_args):
-        self.t = _TriggerThread(trigger_func, trigger_args)
-        self.t.start()
-        self.result = block_func(*block_args)
-        # If block_func returned before our thread made the call, we failed!
-        if not self.t.startedEvent.is_set():
-            self.fail("blocking function '%r' appeared not to block" %
-                      block_func)
-        self.t.join(10) # make sure the thread terminates
-        if self.t.is_alive():
-            self.fail("trigger function '%r' appeared to not return" %
-                      trigger_func)
-        return self.result
+        thread = _TriggerThread(trigger_func, trigger_args)
+        thread.start()
+        try:
+            self.result = block_func(*block_args)
+            # If block_func returned before our thread made the call, we failed!
+            if not thread.startedEvent.is_set():
+                self.fail("blocking function '%r' appeared not to block" %
+                          block_func)
+            return self.result
+        finally:
+            thread.join(10) # make sure the thread terminates
+            if thread.is_alive():
+                self.fail("trigger function '%r' appeared to not return" %
+                          trigger_func)
 
     # Call this instead if block_func is supposed to raise an exception.
     def do_exceptional_blocking_test(self,block_func, block_args, trigger_func,
                                    trigger_args, expected_exception_class):
-        self.t = _TriggerThread(trigger_func, trigger_args)
-        self.t.start()
+        thread = _TriggerThread(trigger_func, trigger_args)
+        thread.start()
         try:
             try:
                 block_func(*block_args)
@@ -77,11 +76,11 @@ def do_exceptional_blocking_test(self,block_func, block_args, trigger_func,
                 self.fail("expected exception of kind %r" %
                                  expected_exception_class)
         finally:
-            self.t.join(10) # make sure the thread terminates
-            if self.t.is_alive():
+            thread.join(10) # make sure the thread terminates
+            if thread.is_alive():
                 self.fail("trigger function '%r' appeared to not return" %
                                  trigger_func)
-            if not self.t.startedEvent.is_set():
+            if not thread.startedEvent.is_set():
                 self.fail("trigger thread ended but event never set")
 
 
@@ -159,8 +158,11 @@ def worker(self, q):
 
     def queue_join_test(self, q):
         self.cum = 0
+        threads = []
         for i in (0,1):
-            threading.Thread(target=self.worker, args=(q,)).start()
+            thread = threading.Thread(target=self.worker, args=(q,))
+            thread.start()
+            threads.append(thread)
         for i in range(100):
             q.put(i)
         q.join()
@@ -169,6 +171,8 @@ def queue_join_test(self, q):
         for i in (0,1):
             q.put(-1)         # instruct the threads to close
         q.join()                # verify that you can join twice
+        for thread in threads:
+            thread.join()
 
     def test_queue_task_done(self):
         # Test to make sure a queue task completed successfully.

Lib/test/test_smtplib.py

@@ -604,7 +604,9 @@ def setUp(self):
         self.sock.settimeout(15)
         self.port = support.bind_port(self.sock)
         servargs = (self.evt, self.respdata, self.sock)
-        threading.Thread(target=server, args=servargs).start()
+        thread = threading.Thread(target=server, args=servargs)
+        thread.start()
+        self.addCleanup(thread.join)
         self.evt.wait()
         self.evt.clear()
 

Lib/test/test_threading.py

@@ -575,6 +575,7 @@ def f():
         self.assertFalse(t.is_alive())
         # And verify the thread disposed of _tstate_lock.
         self.assertIsNone(t._tstate_lock)
+        t.join()
 
     def test_repr_stopped(self):
         # Verify that "stopped" shows up in repr(Thread) appropriately.
@@ -601,6 +602,7 @@ def f():
                 break
             time.sleep(0.01)
         self.assertIn(LOOKING_FOR, repr(t)) # we waited at least 5 seconds
+        t.join()
 
     def test_BoundedSemaphore_limit(self):
         # BoundedSemaphore should raise ValueError if released too often.
@@ -915,6 +917,7 @@ def test_start_thread_again(self):
         thread = threading.Thread()
         thread.start()
         self.assertRaises(RuntimeError, thread.start)
+        thread.join()
 
     def test_joining_current_thread(self):
         current_thread = threading.current_thread()
@@ -928,6 +931,7 @@ def test_daemonize_active_thread(self):
         thread = threading.Thread()
         thread.start()
         self.assertRaises(RuntimeError, setattr, thread, "daemon", True)
+        thread.join()
 
     def test_releasing_unacquired_lock(self):
         lock = threading.Lock()
@@ -1090,6 +1094,8 @@ def test_init_immutable_default_args(self):
         self.callback_event.wait()
         self.assertEqual(len(self.callback_args), 2)
         self.assertEqual(self.callback_args, [((), {}), ((), {})])
+        timer1.join()
+        timer2.join()
 
     def _callback_spy(self, *args, **kwargs):
         self.callback_args.append((args[:], kwargs.copy()))

Lib/test/test_xmlrpc.py

@@ -759,7 +759,9 @@ def setUp(self):
         self.evt = threading.Event()
         # start server thread to handle requests
         serv_args = (self.evt, self.request_count, self.requestHandler)
-        threading.Thread(target=self.threadFunc, args=serv_args).start()
+        thread = threading.Thread(target=self.threadFunc, args=serv_args)
+        thread.start()
+        self.addCleanup(thread.join)
 
         # wait for the server to be ready
         self.evt.wait()
@@ -1211,7 +1213,9 @@ def setUp(self):
         self.evt = threading.Event()
         # start server thread to handle requests
         serv_args = (self.evt, 1)
-        threading.Thread(target=http_server, args=serv_args).start()
+        thread = threading.Thread(target=http_server, args=serv_args)
+        thread.start()
+        self.addCleanup(thread.join)
 
         # wait for the server to be ready
         self.evt.wait()