是否可以使用 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
...
更多信息
- 我研究过添加一个装饰器,例如
@pytest.mark.timeout(5)
,但那没有用,它停止了整个功能,所以我从来没有到达 assert
s。 (按照建议 here)。
- 如果我找不到方法,我将只测试部分,但理想情况下我想找到一种通过打破循环来测试的方法。
- 我知道我可以重写我的代码以使其具有超时,但这意味着更改代码以使其可测试,我认为这不是一个好的设计。
- 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
我有一个 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
...
更多信息
- 我研究过添加一个装饰器,例如
@pytest.mark.timeout(5)
,但那没有用,它停止了整个功能,所以我从来没有到达assert
s。 (按照建议 here)。 - 如果我找不到方法,我将只测试部分,但理想情况下我想找到一种通过打破循环来测试的方法。
- 我知道我可以重写我的代码以使其具有超时,但这意味着更改代码以使其可测试,我认为这不是一个好的设计。
- 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