def test_timeout_thread(self):
        queue = Queue.Queue()
        killer_thread = threading.Thread(
            target=utils.timeout_thread,
            args=(queue,),
        )
        test_thread = threading.Thread(
            target=utils.interruptable_sleep,
            args=(30, 0.01),
        )
        test_thread.start()
        start_time = time.time()
        queue.put((test_thread.ident, start_time + 1))
        killer_thread.start()
        test_thread.join()
        end_time = time.time()
        queue.put((None, None))
        killer_thread.join()
        time_elapsed = end_time - start_time
        # NOTE(sskripnick): Killing thread with PyThreadState_SetAsyncExc
        # works with sinificant delay. Make sure this delay is less
        # than 10 seconds.
        self.assertTrue(time_elapsed < 11,
                        "Thread killed too late (%s seconds)" % time_elapsed)
Create and kill thread by timeout.
dataset/ETHPy150Open openstack/rally/tests/unit/common/test_utils.py/TimeoutThreadTestCase.test_timeout_thread