Django并行测试时,重写unittest的startTest/stopTest失效问题
我在自定义测试运行器中重写了unittest的TextTestResult类的startTest和stopTest方法,单进程运行测试时一切正常,但使用--parallel并行测试参数时出现异常:计算出的startTest与stopTest之间的耗时极小(例如4.8000000000492093e-05),和实际测试用例的执行时间(约2秒)完全不符。想确认这两个钩子方法是否适用于并行测试场景,或者是否需要使用其他替代钩子?
复现步骤
假设Django项目名为project,应用名为app:
- 在项目目录创建
custom_test_runner.py,代码如下:
from time import perf_counter from unittest import TextTestResult, TextTestRunner from django.test.runner import DiscoverRunner class CustomTestRunnerTextTestResult(TextTestResult): def __init__(self, stream, descriptions, verbosity): super(CustomTestRunnerTextTestResult, self).__init__(stream, descriptions, verbosity) def startTest(self, test): self.start_time = perf_counter() super(CustomTestRunnerTextTestResult, self).startTest(test) def stopTest(self, test): super(CustomTestRunnerTextTestResult, self).stopTest(test) print(f"Time elapsed {str(test)} for {perf_counter() - self.start_time}") class CustomTestRunnerTextTestRunner(TextTestRunner): def __init__(self, **kwargs): super().__init__(**kwargs) resultclass = CustomTestRunnerTextTestResult class CustomTestRunnerTestRunner(DiscoverRunner): def __init__(self, **kwargs): super(CustomTestRunnerTestRunner, self).__init__(**kwargs) test_runner = CustomTestRunnerTextTestRunner
- 在tests文件夹创建
test_custom_test_runner.py,代码如下:
import time from django.test import TestCase class Test1(TestCase): def test_1(self): time.sleep(2) class Test2(TestCase): def test_2(self): time.sleep(2)
- 在项目的
settings.py中配置自定义测试运行器:
TEST_RUNNER = "project.custom_test_runner.CustomTestRunnerTestRunner"
- 运行测试命令:
python manage.py test app.tests.test_custom_test_runner --parallel
运行结果显示每个测试用例的耗时极小,但去掉--parallel参数后,耗时符合预期(约2秒/测试用例)。
Django版本:3.2.14
问题原因
Django的--parallel并行测试是基于多进程实现的:主进程负责分发测试任务,子进程实际执行测试用例。你重写的TextTestResult实例是在主进程中创建的,但startTest方法是主进程在调度测试任务时调用的(此时测试还未开始执行),而stopTest是子进程执行完测试后通知主进程调用的。这导致你计算的时间差只是主进程"调度任务"到"收到子进程完成通知"的耗时,而非测试用例实际在子进程中执行的时间,所以结果异常偏小。
此外,多进程环境下主进程和子进程的内存空间完全隔离,子进程无法访问主进程中self.start_time的正确值,这也会导致时间计算错误。
解决方案
要在并行测试中准确统计每个测试用例的执行时间,需要在**测试用例的执行上下文(子进程内)**中记录时间,推荐两种方案:
方案1:重写TestCase的setUp/tearDown方法
直接在测试用例的生命周期方法中记录时间,每个测试用例在子进程中独立执行,时间记录不会受多进程影响:
import time from time import perf_counter from django.test import TestCase class TimedTestCase(TestCase): def setUp(self): super().setUp() self.test_start_time = perf_counter() def tearDown(self): elapsed = perf_counter() - self.test_start_time print(f"Time elapsed {self._testMethodName} for {elapsed}") super().tearDown() class Test1(TimedTestCase): def test_1(self): time.sleep(2) class Test2(TimedTestCase): def test_2(self): time.sleep(2)
方案2:使用Django测试信号
利用Django提供的测试信号test_started和test_finished,这两个信号会在测试用例执行的进程内触发,能准确捕获执行时间:
from time import perf_counter from django.test.signals import test_started, test_finished test_timings = {} def record_test_start(sender, **kwargs): test = kwargs['test'] test_timings[id(test)] = perf_counter() def record_test_finish(sender, **kwargs): test = kwargs['test'] start_time = test_timings.pop(id(test), None) if start_time: elapsed = perf_counter() - start_time print(f"Time elapsed {str(test)} for {elapsed}") test_started.connect(record_test_start) test_finished.connect(record_test_finish)
将这段代码放在项目的custom_test_runner.py中,或者在测试模块的开头导入即可。
内容的提问来源于stack exchange,提问作者Abhinav Prakash

