封装一个unittest多线程执行测试用例怎么在最后执行run方法的时候添加线程组
可以通过继承unittest.TestRunner类并重写run()方法来实现添加线程组。具体实现如下:
import threading
import unittest
class MultiThreadTestRunner(unittest.TextTestRunner):
def __init__(self, thread_count=5, *args, **kwargs):
super().__init__(*args, **kwargs)
self.thread_count = thread_count
self.threads = []
def run(self, test):
result = self._makeResult()
threads = []
for i in range(self.thread_count):
t = threading.Thread(target=self._runThread, args=(test, result))
threads.append(t)
t.start()
for t in threads:
t.join()
self.stream.writeln()
self.stream.writeln("Ran %d tests in %.3fs" % (result.testsRun, result.elapsed_time))
self.stream.writeln()
self._printErrors(result)
return result
def _runThread(self, test, result):
thread_result = self._makeResult()
thread_result.failfast = result.failfast
thread_result.buffer = result.buffer
test(result=thread_result)
with result.lock:
result.testsRun += thread_result.testsRun
result.failures.extend(thread_result.failures)
result.errors.extend(thread_result.errors)
result.skipped.extend(thread_result.skipped)
result.expectedFailures.extend(thread_result.expectedFailures)
result.unexpectedSuccesses.extend(thread_result.unexpectedSuccesses)
result.elapsed_time += thread_result.elapsed_time
if __name__ == '__main__':
suite = unittest.TestLoader().discover('.')
runner = MultiThreadTestRunner(thread_count=5)
runner.run(suite)
在上述代码中,我们继承了unittest.TextTestRunner类,并在构造函数中添加了一个thread_count参数,表示要使用的线程数。在run()方法中,我们创建了一个线程数组threads,并循环创建thread_count个线程,并将每个线程的目标函数设置为_runThread()方法。_runThread()方法用于执行测试用例,并将结果存储在thread_result中。最后,在主线程中等待所有线程执行完毕后,将所有线程的结果合并到result中,并返回result。
在主程序中,我们使用unittest.TestLoader().discover()方法来自动发现所有测试用例,并将其添加到测试套件中。然后,我们创建了一个MultiThreadTestRunner实例,并将其传递给测试套件的run()方法来执行测试用例。在执行run()方法时,我们将线程数设置为5。
原文地址: https://www.cveoy.top/t/topic/bJdx 著作权归作者所有。请勿转载和采集!