import weakref
import test.support
import test.support.script_helper
+from test import support
# Skip tests if _multiprocessing wasn't built.
queue.join_thread()
+def join_process(process, timeout):
+ # Since multiprocessing.Process has the same API than threading.Thread
+ # (join() and is_alive(), the support function can be reused
+ support.join_thread(process, timeout)
+
+
#
# Constants
#
for p in procs:
p.start()
for p in procs:
- p.join(timeout=10)
+ join_process(p, timeout=10)
for p in procs:
self.assertEqual(p.exitcode, 0)
for p in procs:
p.terminate()
for p in procs:
- p.join(timeout=10)
+ join_process(p, timeout=10)
if os.name != 'nt':
for p in procs:
self.assertEqual(p.exitcode, -signal.SIGTERM)
p = self.Process(target=self._test_sys_exit, args=(reason, testfn))
p.daemon = True
p.start()
- p.join(5)
+ join_process(p, timeout=5)
self.assertEqual(p.exitcode, 1)
with open(testfn, 'r') as f:
p = self.Process(target=sys.exit, args=(reason,))
p.daemon = True
p.start()
- p.join(5)
+ join_process(p, timeout=5)
self.assertEqual(p.exitcode, reason)
#
state.value += 1
cond.notify()
- p.join(5)
- self.assertFalse(p.is_alive())
+ join_process(p, timeout=5)
self.assertEqual(p.exitcode, 0)
@classmethod
state.value += 1
cond.notify()
- p.join(5)
+ join_process(p, timeout=5)
self.assertTrue(success.value)
@classmethod
self.assertEqual(conn.recv(), 456)
conn.close()
l.close()
- p.join(10)
+ join_process(p, timeout=10)
finally:
socket.setdefaulttimeout(old_timeout)
p = multiprocessing.Process(target=cls.child, args=(n-1, conn))
p.start()
conn.close()
- p.join(timeout=5)
+ join_process(p, timeout=5)
else:
conn.send(len(util._afterfork_registry))
conn.close()
p.start()
w.close()
new_size = r.recv()
- p.join(timeout=5)
+ join_process(p, timeout=5)
self.assertLessEqual(new_size, old_size)
#
p.start()
writer.close()
e = reader.recv()
- p.join(timeout=5)
+ join_process(p, timeout=5)
finally:
self.close(fd)
writer.close()
gc_collect()
+def join_thread(thread, timeout=30.0):
+ """Join a thread. Raise an AssertionError if the thread is still alive
+ after timeout seconds.
+ """
+ thread.join(timeout)
+ if thread.is_alive():
+ msg = f"failed to join the thread in {timeout:.1f} seconds"
+ raise AssertionError(msg)
+
+
def reap_children():
"""Use this function at the end of test_main() whenever sub-processes
are started. This will help ensure that no extra children (zombies)
c.push(b"I'm not dead yet!" + term)
c.push(SERVER_QUIT)
asyncore.loop(use_poll=self.usepoll, count=300, timeout=.01)
- s.join(timeout=TIMEOUT)
- if s.is_alive():
- self.fail("join() timed out")
+ support.join_thread(s, timeout=TIMEOUT)
self.assertEqual(c.contents, [b"hello world", b"I'm not dead yet!"])
c.push(data)
c.push(SERVER_QUIT)
asyncore.loop(use_poll=self.usepoll, count=300, timeout=.01)
- s.join(timeout=TIMEOUT)
- if s.is_alive():
- self.fail("join() timed out")
+ support.join_thread(s, timeout=TIMEOUT)
self.assertEqual(c.contents, [data[:termlen]])
c.push(data)
c.push(SERVER_QUIT)
asyncore.loop(use_poll=self.usepoll, count=300, timeout=.01)
- s.join(timeout=TIMEOUT)
- if s.is_alive():
- self.fail("join() timed out")
+ support.join_thread(s, timeout=TIMEOUT)
self.assertEqual(c.contents, [])
self.assertEqual(c.buffer, data)
p = asynchat.simple_producer(data+SERVER_QUIT, buffer_size=8)
c.push_with_producer(p)
asyncore.loop(use_poll=self.usepoll, count=300, timeout=.01)
- s.join(timeout=TIMEOUT)
- if s.is_alive():
- self.fail("join() timed out")
+ support.join_thread(s, timeout=TIMEOUT)
self.assertEqual(c.contents, [b"hello world", b"I'm not dead yet!"])
data = b"hello world\nI'm not dead yet!\n"
c.push_with_producer(data+SERVER_QUIT)
asyncore.loop(use_poll=self.usepoll, count=300, timeout=.01)
- s.join(timeout=TIMEOUT)
- if s.is_alive():
- self.fail("join() timed out")
+ support.join_thread(s, timeout=TIMEOUT)
self.assertEqual(c.contents, [b"hello world", b"I'm not dead yet!"])
c.push(b"hello world\n\nI'm not dead yet!\n")
c.push(SERVER_QUIT)
asyncore.loop(use_poll=self.usepoll, count=300, timeout=.01)
- s.join(timeout=TIMEOUT)
- if s.is_alive():
- self.fail("join() timed out")
+ support.join_thread(s, timeout=TIMEOUT)
self.assertEqual(c.contents,
[b"hello world", b"", b"I'm not dead yet!"])
# where the server echoes all of its data before we can check that it
# got any down below.
s.start_resend_event.set()
- s.join(timeout=TIMEOUT)
- if s.is_alive():
- self.fail("join() timed out")
+ support.join_thread(s, timeout=TIMEOUT)
self.assertEqual(c.contents, [])
# the server might have been able to send a byte or two back, but this
self.assertRaises(TypeError, c.push, 'unicode')
c.push(SERVER_QUIT)
asyncore.loop(use_poll=self.usepoll, count=300, timeout=.01)
- s.join(timeout=TIMEOUT)
+ support.join_thread(s, timeout=TIMEOUT)
self.assertEqual(c.contents, [b'bytes', b'bytes', b'bytes'])
proto.transport.close()
lsock.close()
- thread.join(1)
+ support.join_thread(thread, timeout=1)
self.assertFalse(thread.is_alive())
self.assertEqual(proto.state, 'CLOSED')
self.assertEqual(proto.nbytes, len(message))
self.assertEqual(cap.getvalue(), data*2)
finally:
- t.join(timeout=TIMEOUT)
- if t.is_alive():
- self.fail("join() timed out")
+ support.join_thread(t, timeout=TIMEOUT)
@unittest.skipUnless(hasattr(asyncore, 'file_wrapper'),
except OSError:
pass
finally:
- t.join(timeout=TIMEOUT)
- if t.is_alive():
- self.fail("join() timed out")
+ support.join_thread(t, timeout=TIMEOUT)
class TestAPI_UseIPv4Sockets(BaseTestAPI):
family = socket.AF_INET
# cleanup the server
self.server.shutdown()
self.server.server_close()
- self.thread.join(3.0)
+ support.join_thread(self.thread, 3.0)
+ # Explicitly clear the attribute to prevent dangling thread
+ self.thread = None
def test_EOF_without_complete_welcome_message(self):
# http://bugs.python.org/issue5949
to terminate.
"""
self.close()
- self._thread.join(timeout)
+ support.join_thread(self._thread, timeout)
+ self._thread = None
asyncore.close_all(map=self._map, ignore_all=True)
- alive = self._thread.is_alive()
- self._thread = None
- if alive:
- self.fail("join() timed out")
class ControlMixin(object):
"""
"""
self.shutdown()
if self._thread is not None:
- self._thread.join(timeout)
- alive = self._thread.is_alive()
+ support.join_thread(self._thread, timeout)
self._thread = None
- if alive:
- self.fail("join() timed out")
self.server_close()
self.ready.clear()
finally:
t.ready.wait(2.0)
logging.config.stopListening()
- t.join(2.0)
- if t.is_alive():
- self.fail("join() timed out")
+ support.join_thread(t, 2.0)
def test_listen_config_10_ok(self):
with support.captured_stdout() as output:
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)
+ support.join_thread(thread, 10) # make sure the thread terminates
# Call this instead if block_func is supposed to raise an exception.
def do_exceptional_blocking_test(self,block_func, block_args, trigger_func,
self.fail("expected exception of kind %r" %
expected_exception_class)
finally:
- thread.join(10) # make sure the thread terminates
- if thread.is_alive():
- self.fail("trigger function '%r' appeared to not return" %
- trigger_func)
+ support.join_thread(thread, 10) # make sure the thread terminates
if not thread.startedEvent.is_set():
self.fail("trigger thread ended but event never set")
import threading
import time
import unittest
+from test import support
TIMEOUT = 10
self.assertEqual(q.get(timeout=TIMEOUT), 5)
self.assertTrue(q.empty())
timer.advance(1000)
- t.join(timeout=TIMEOUT)
- self.assertFalse(t.is_alive())
+ support.join_thread(t, timeout=TIMEOUT)
self.assertTrue(q.empty())
self.assertEqual(timer.time(), 5)
self.assertEqual(q.get(timeout=TIMEOUT), 4)
self.assertTrue(q.empty())
timer.advance(1000)
- t.join(timeout=TIMEOUT)
- self.assertFalse(t.is_alive())
+ support.join_thread(t, timeout=TIMEOUT)
self.assertTrue(q.empty())
self.assertEqual(timer.time(), 4)