tests: Compatiblity shim for threading.Thread.is_alive()

On Python >= 3.8 thread.isAlive() is deprecated (removed in Python 3.9.
On Python <= 2.5 thread.is_alive() isn't present (added in Python 2.6).
pull/775/head
Alex Willmer 4 years ago
parent 522085ab35
commit 4b39013ef4

@ -103,6 +103,18 @@ if hasattr(subprocess.Popen, 'terminate'):
Popen__terminate = subprocess.Popen.terminate Popen__terminate = subprocess.Popen.terminate
def threading__thread_is_alive(thread):
"""Return whether the thread is alive (Python version compatibility shim).
On Python >= 3.8 thread.isAlive() is deprecated (removed in Python 3.9).
On Python <= 2.5 thread.is_alive() isn't present (added in Python 2.6).
"""
try:
return thread.is_alive()
except AttributeError:
return thread.isAlive()
def wait_for_port( def wait_for_port(
host, host,
port, port,
@ -334,7 +346,9 @@ class TestCase(unittest2.TestCase):
for thread in threading.enumerate(): for thread in threading.enumerate():
name = thread.getName() name = thread.getName()
# Python 2.4: enumerate() may return stopped threads. # Python 2.4: enumerate() may return stopped threads.
assert (not thread.isAlive()) or name in self.ALLOWED_THREADS, \ assert \
not threading__thread_is_alive(thread) \
or name in self.ALLOWED_THREADS, \
'Found thread %r still running after tests.' % (name,) 'Found thread %r still running after tests.' % (name,)
counts[name] = counts.get(name, 0) + 1 counts[name] = counts.get(name, 0) + 1

@ -31,14 +31,14 @@ class RunWithRouterTest(testlib.TestCase):
def test_run_with_broker(self): def test_run_with_broker(self):
router = mitogen.utils.run_with_router(func0) router = mitogen.utils.run_with_router(func0)
self.assertIsInstance(router, mitogen.master.Router) self.assertIsInstance(router, mitogen.master.Router)
self.assertFalse(router.broker._thread.isAlive()) self.assertFalse(testlib.threading__thread_is_alive(router.broker._thread))
class WithRouterTest(testlib.TestCase): class WithRouterTest(testlib.TestCase):
def test_with_broker(self): def test_with_broker(self):
router = func() router = func()
self.assertIsInstance(router, mitogen.master.Router) self.assertIsInstance(router, mitogen.master.Router)
self.assertFalse(router.broker._thread.isAlive()) self.assertFalse(testlib.threading__thread_is_alive(router.broker._thread))
class Dict(dict): pass class Dict(dict): pass

Loading…
Cancel
Save