在 python 中使用 multiprocessing.Process() 对函数进行单元测试和覆盖

如何解决在 python 中使用 multiprocessing.Process() 对函数进行单元测试和覆盖

我正在尝试使用 multiprocessing.Process() 测试可能会崩溃的函数。如果函数在 10 分钟内没有理想地响应，下面编写的代码会重新启动调用，这完全正常，即使 self.assert... 测试工作正常。问题出在覆盖率上，出于某种原因，在我的文件上运行覆盖率表明 my_func 正在被调用，但在测试时它内部没有任何内容被覆盖，而测试（需要执行整个函数）没有崩溃）正在通过。

        while True:
           p = multiprocessing.Process(
               target=my_func,args=(return_dict) # return_dict to get the return values
           )
           p.start()
           p.join(600)

           if p.is_alive():
               print(
                   "Taking too long,"
                   + "KILLING IT and starting AGAIN."
               )
               p.kill()
               p.join()
           else:
               break

我做错了什么吗？如果信息似乎不完整，请告诉我，然后我会添加我的测试和功能。

解决方法

Reference link

为了覆盖在子进程中运行的代码，您需要实际指定它。

1. 在您的文件中添加 import coverage 导入语句

2. 在每个 coverage.process_startup() 语句的顶部添加 multiprocessing.Process()，在我的例子中是这样的 -

   coverage.process_startup()

   p = multiprocessing.Process( 目标=my_func,args=(return_dict) )

   .... 其余代码

3. 在您的根目录中创建一个 .coveragerc 文件并添加以下行以指定覆盖子进程和多进程。

   [运行]

   并发=多处理

4. 设置一个名为 COVERAGE_PROCESS_START 的路径变量来存储 .coveragerc 文件的位置

   设置 COVERAGE_PROCESS_START=%cd%\.coveragerc // Windows

   export COVERAGE_PROCESS_START=$PWD/.coveragerc // UNIX/LINUX

5. 一切就绪，使用 coverage run -m unittest 运行测试，注意现在生成了 1 个以上的覆盖率报告文件。要组合它们运行 coverage combine，最后运行 coverage report