test_threadpool.py 文件源码

def test_scaleup(self):
        """
        test that a thread pool will scale up properly
        :return:
        """
        pool = self.ThreadPool(TestThread, initial_size=1, keywordarg="keywordstring")

        try:
            self.assertEqual(threading.activeCount(), 2)
            pool.scale_up()
            self.assertEqual(threading.activeCount(), 3)
            pool.scale_up()
            self.assertEqual(threading.activeCount(), 4)
        finally:
            pool.safe_terminate()