python-asyncio - 使用流为 asyncio 编写 pytest 测试用例
问题描述
我正在尝试为 asyncio 函数编写一个 pytest 测试用例,该函数确实读取输出流(stderr/stdout)并修改行。我要测试的函数(再次被调用 inside asyncio.gather
)如下所示:
import asyncio
async def watch(stream):
while True:
lines = await stream.read(2**16)
if not lines or lines == "":
break
lines = lines.strip().split("\n")
for line in lines:
print(f'myPrefix-{line}')
我写的pytest测试用例如下:
import asyncio
from io import StringIO
import pytest
@pytest.fixture(autouse=True)
def event_loop():
loop = asyncio.get_event_loop()
yield loop
loop.close()
@pytest.mark.asyncio
async def test_watch(event_loop):
expected_outcome = "myPrefix-This is stdout"
def write_something():
print("This is stdout")
with patch("sys.stdout", new=StringIO()) as mock_stdout:
write_something()
output = await watch(mock_stdout.getvalue())
assert output.return_value == expected_outcome
但是,当我执行这个 pytest 时,我遇到了AttributeError: 'str' object has no attribute 'read'
. 如何在处理 stdout/stderr 流时测试 asyncio 协程?
解决方案
StringIO
没有协程方法read
,所以你不能模拟它并让它与你的监视协程函数一起工作(调用getvalue
实例StringIO
也只是传入写入标准输出的字符串,这解释了你得到的错误)。假设您的 watch 函数中的流是 的一个实例StreamReader
,您可以StreamReader
在测试中创建一个 asyncio 实例并使用该feed_data
方法向流中写入一些内容。然后你可以把它传递给watch
. 然后,您可以使用capsys
Pytest 附带的夹具来捕获watch
写入标准输出的内容。
以下是作为独立传递的代码的更新版本:
import asyncio
import pytest
async def watch(stream):
while True:
lines = await stream.read(2 ** 16)
if not lines or lines == "":
break
lines = lines.decode().strip().split("\n") #note the use of decode()
for line in lines:
print(f'myPrefix-{line}')
@pytest.fixture(autouse=True)
def event_loop():
loop = asyncio.get_event_loop()
yield loop
loop.close()
@pytest.mark.asyncio
async def test_watch(capsys):
expected_outcome = "myPrefix-This is stdout\n"
stream = asyncio.StreamReader()
stream.feed_data(b'This is stdout\n')
stream.feed_eof()
await watch(stream)
captured = capsys.readouterr()
assert captured.out == expected_outcome
推荐阅读
- javascript - useEffect 中的 setState 在酶中不起作用
- git - 如何解决:“对不起,不能拆分这个大块”?
- azure - 在 Azure DevOps 中运行 az cli 命令
- raspberry-pi - 在 Raspberry pi 3 上未以单声道接收 UDP
- regex - 如何使用 OR 比较 Google 表格中的多个 RIGHT 函数来缩短公式?
- mysql - 如何从 amfphp 数组在 AS3(ActionScript 3.0)中的文本输入中显示字符串值?
- javascript - How to export csv object in react
- postgresql - 如何在 Postgresql 中进行主动活动 HA 节点设置
- python - 将 selenium webdriver 字段写入 txt 文件
- java - 如果另一个地图的键不存在,如何将元素添加到地图