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

Question

在我正在使用的源代码（source link here 和 WIP PR here）中，我试图通过在类的 __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。

我想知道我在这里做错了什么以及如何成功测试此类的__init__ 方法？

非常感谢您的宝贵时间！

Answer 1

这里有两个问题：

1. 您正在设置get_running_loop 的返回值，但异常不是返回值。如果你想让你的模拟代码引发异常，你需要配置一个side_effect。

2. 您的代码捕获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)

---

编辑：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 的返回值。我可能会摆脱嵌套的 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 语句来做同样的事情。