是否可以使用 pytest 测试 while True 循环(我尝试超时)?

Is it possible to test a while True loop with pytest (I try with a timeout)?

我有一个 python 函数 foo,里面有一个 while True 循环。 对于背景:预计会从网络上流式传输信息,进行一些写作并 运行 无限期地进行。断言测试写入是否正确。

显然我需要它在某个时候停止,以便进行测试。

我所做的是通过 multirpocessing 到 运行 并在那里引入超时,但是当我看到测试覆盖率时,运行 通过多处理的功能未被标记如前所述。

问题 1:为什么 pytest 现在这样工作?

问题 2:我怎样才能完成这项工作?

我在想这可能是因为我在技术上退出了循环,所以也许 pytest 没有将其标记为已测试....

import time
import multiprocessing

def test_a_while_loop():
    # Start through multiprocessing in order to have a timeout.
    p = multiprocessing.Process(
        target=foo
        name="Foo",
    )
    try:
        p.start()
        # my timeout
        time.sleep(10)
        p.terminate()
    finally:
        # Cleanup.
        p.join()

    # Asserts below
    ...

更多信息

  1. 我研究过添加一个装饰器,例如 @pytest.mark.timeout(5),但那没有用,它停止了整个功能,所以我从来没有到达 asserts。 (按照建议 here)。
  2. 如果我找不到方法,我将只测试部分,但理想情况下我想找到一种通过打破循环来测试的方法。
  3. 我知道我可以重写我的代码以使其具有超时,但这意味着更改代码以使其可测试,我认为这不是一个好的设计。
    1. Mocks 我没有尝试过(建议 here),因为我不相信我可以模拟我所做的事情,因为它从网络上写入信息。我需要实际看到 "original" 工作。

将您要测试的功能分解为辅助方法。测试辅助方法。

def scrape_web_info(url):
  data = get_it(url)
  return data

# In production:

while True:
  scrape_web_info(...)

# During test:

def test_web_info():
  assert scrape_web_info(...) == ...

是的,这是可能的,上面的代码显示了一种方法(运行 通过带超时的多处理)。

由于断言 运行 没问题,我发现问题不是 pytest,而是覆盖率报告没有正确说明 multiprocessing

我描述了我如何解决这个(现在分开的)问题

实际上,我遇到了同样的问题,无休止的测试和覆盖任务。但是,在我的代码中,有一个 .run_forever() 方法在无限循环中运行 .run_once() 方法。因此,我可以为 .run_once() 方法编写单元测试来测试其功能。尽管如此,如果你想测试你的 forever 函数,尽管 Halting Problem 以获得更大范围的代码覆盖,我建议使用超时的以下方法,无论你提到的工具包括multiprocessing@pytest.mark.timeout(5) 对我也不起作用:

  • 首先,安装 interruptingcow PyPI 包以获得引发可选异常的良好超时:pip install interruptingcow
  • 然后:
import pytest
import asyncio
from interruptingcow import timeout
from <path-to-loop-the-module> import EventLoop

class TestCase:
    @pytest.mark.parametrize("test_case", ['none'])
    def test_events(self, test_case: list):
        assert EventLoop().run_once()  # It's usual

    @pytest.mark.parametrize("test_case", ['none'])
    def test_events2(self, test_case: list):
        try:
            with timeout(10, exception=asyncio.CancelledError):
                EventLoop().run_forever()
                assert False
        except asyncio.CancelledError:
            assert True