使用 Pytest 测试 Asyncio：如何通过模拟事件循环来测试 try-except 块？

Question

在我正在使用的源代码 (source link here and WIP PR here) 中，我试图通过在 class' __init__ 中测试 try-except 块来提高测试覆盖率方法。

从源代码中剥离多余的代码，相关代码如下所示：

# webrtc.py

import asyncio
from loguru import logger
try:
    from asyncio import get_running_loop  # noqa Python >=3.7
except ImportError:  # pragma: no cover
    from asyncio.events import _get_running_loop as get_running_loop  # pragma: no cover

class WebRTCConnection:
    loop: Any

    def __init__(self) -> None:
        try:
            self.loop = get_running_loop()
        except RuntimeError as e:
            self.loop = None
            logger.error(e)
        
        if self.loop is None:
            self.loop = asyncio.new_event_loop()

在单独的测试文件中，我想模拟 RuntimeError 来测试 try except 块：

# webrtc_test.py

from unittest.mock import patch
from unittest.mock import Mock

import asyncio
import pytest
from webrtc import WebRTCConnection

@pytest.mark.asyncio
async def test_init_patch_runtime_error() -> None:
    nest_asyncio.apply()

    with patch("webrtc.get_running_loop", return_value=RuntimeError):
        with pytest.raises(RuntimeError):
            WebRTCConnection()

@pytest.mark.asyncio
async def test_init_mock_runtime_error() -> None:
    nest_asyncio.apply()

    mock_running_loop = Mock()
    mock_running_loop.side_effect = RuntimeError
    with patch("webrtc.get_running_loop", mock_running_loop):
        with pytest.raises(RuntimeError):
            domain = Domain(name="test")
            WebRTCConnection()

两个测试都不会通过，因为它们都不会引发 RuntimeError.

此外，我尝试用 monkeypatch:

模拟 asyncio.new_event_loop

# webrtc_test.py

from unittest.mock import patch
from unittest.mock import Mock

import asyncio
import pytest

from webrtc import WebRTCConnection

@pytest.mark.asyncio
async def test_init_new_event_loop(monkeypatch) -> None:
    nest_asyncio.apply()

    WebRTCConnection.loop = None
    mock_new_loop = Mock()
    monkeypatch.setattr(asyncio, "new_event_loop", mock_new_loop)
    WebRTCConnection()

    assert mock_new_loop.call_count == 1

这个测试也失败了，因为猴子补丁从未被调用：> assert mock_new_loop.call_count == 1 E assert 0 == 1.

我想知道我在这里做错了什么，我如何才能成功测试此 class 的 __init__ 方法？

非常感谢您的宝贵时间！

Answer 1

你这里有两个问题：

您正在设置 get_running_loop 的 return 值，但异常不是 return 值。如果你想让你的模拟代码引发异常，你需要配置一个 side_effect.
您的代码捕获 RuntimeError 并且不会重新引发：您只需设置 self.loop = None 并记录错误。这意味着即使您从 get_event_loop 成功引发 RuntimeError，该异常也永远不会对您的测试可见，因为它已被您的代码使用。

如果你要模拟你的 logger 对象，你可以检查 logger.error 是否被调用了异常。例如：

@pytest.mark.asyncio
async def test_init_patch_runtime_error() -> None:
    nest_asyncio.apply()

    with patch("webrtc.logger") as mock_logger:
        with patch("webrtc.get_running_loop", side_effect=RuntimeError()):
            WebRTCConnection()
            assert isinstance(mock_logger.error.call_args[0][0], RuntimeError)

Edit: W/r/t 检查 self.loop = None 部分，我可能会这样重写代码：

class WebRTCConnection:
    loop: Any = None

    def __init__(self) -> None:
    ┆   try:
    ┆   ┆   self.loop = get_running_loop()
    ┆   except RuntimeError as e:
    ┆   ┆   logger.error(e)

    ┆   if self.loop is None:
    ┆   ┆   self.loop = asyncio.new_event_loop()

然后在测试时，您需要为 new_event_loop 模拟一个 return 值。我可能会摆脱嵌套的 with 语句，而只是在函数上使用 patch 装饰器：

@pytest.mark.asyncio
@patch('webrtc.logger')
@patch('webrtc.get_running_loop', side_effect=RuntimeError())
@patch('webrtc.asyncio.new_event_loop', return_value='fake_loop')
async def test_init_patch_runtime_error(
    mock_new_event_loop,
    mock_get_running_loop,
    mock_logger
) -> None:
    nest_asyncio.apply()

    rtc = WebRTCConnection()
    assert isinstance(mock_logger.error.call_args[0][0], RuntimeError)
    assert rtc.loop == 'fake_loop'

...但显然您可以使用一系列嵌套的 with patch(...) 语句或单个长 with 语句来做同样的事情。

使用 Pytest 测试 Asyncio：如何通过模拟事件循环来测试 try-except 块？

Testing Asyncio with Pytest: How to test a try-except block by mocking the event loop?

python

monkeypatching

pytest

python-asyncio

pytest-asyncio