我们可以限制testtools.ConcurrentStreamTestSuite并行运行的测试数量吗?

Question

我正在从unittest转换到testtools以并行运行我的Python测试。

我想知道是否有一种方法可以限制使用testtools.ConcurrentStreamTestSuite时并发运行的测试数量?我很快就会有数百个测试，并且需要将执行次数限制在每次最多10个。当线程释放时，剩余的线程将排队等待执行。

如果是这样，你能分享一些代码片段，显示如何做到这一点吗?

Answer 1

我已经清理了逻辑，这里是该函数的新版本(工作得更好):

def split_suite_into_chunks(num_threads, suite):
    # Compute num_threads such that the number of threads does not exceed the value passed to the function
    # Keep num_threads to a reasonable number of threads
    if num_threads < 0: num_threads = 1
    if num_threads > 8: num_threads = 8
    num_tests = suite.countTestCases()
    s = []
    s_tmp = unittest.TestSuite()
    n = round(num_tests / num_threads)
    for case in suite:
        if n <= 0 and s_tmp.countTestCases() > 0:
            s.append([s_tmp, None])
            num_threads -= 1
            num_tests -= s_tmp.countTestCases()
            s_tmp = unittest.TestSuite()
            n = round(num_tests / num_threads)
        s_tmp.addTest(case)
        n -= 1
    if s_tmp.countTestCases() > 0:
        if s_tmp.countTestCases() > 0: s.append([s_tmp, None])
        num_tests -= s_tmp.countTestCases()
    if num_tests != 0: print("Error: num_tests should be 0 but is %s!" % num_tests)
    return s

Answer 2

我明白了。所需要做的就是拆分测试，并以某种方式将它们组织成套件。下面是一些示例代码(需要一些清理，但已经完成了工作):

""" Splits a Test Suite so that no more than 'n' threads will execute the tests """
def split_suite_into_chunks(n, suite):
    # Keep n to a reasonable number of threads
    if n < 0:
        n = 1
    if n > 8:
        n = 8
    # Compute n such that the number of threads does not exceed the value passed to the function
    n = math.ceil(suite.countTestCases() / n)
    s = []
    i = 0
    s_tmp = unittest.TestSuite()
    for case in suite:
        if i < n:
            s_tmp.addTest(case)
            i += 1
        if i == n:
            s.append([s_tmp, None])
            i = 0
            s_tmp = unittest.TestSuite()
    if (i > 0):
        s.append([s_tmp, None])
    return s
def add_test_case_to_suite(suite, tc_name):
    # Creates a Test Suite with each Test Case added n times
    n = 1
    for i in range(0, n):
        suite.addTest(tc_name)
def get_suite():
    suite = unittest.TestSuite()
    add_test_case_to_suite(suite, My_login_test('tc_login'))
    add_test_case_to_suite(suite, My_login_test('tc_logout'))
    return suite
if __name__ == "__main__":
    # This is what sets the number of threads
    num_threads = 4
    suite = get_suite()
    concurrent_suite = testtools.ConcurrentStreamTestSuite(lambda: (split_suite_into_chunks(num_threads, suite)))
    result.startTestRun()
    concurrent_suite.run(result)
    result.stopTestRun()

__main__中的

num_threads设置要使用的线程数。有趣的是，我意识到即使我将线程数设置为超过8个(比如10个)，我也无法一次运行超过8个线程。这不一定是一个问题(更多的是一个观察)，因为有那么多线程对我的性能没有帮助。