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

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

此测试也失败，因为从未调用monkey补丁：> assert mock_new_loop.call_count == 1 E assert 0 == 1

我想知道我在这里做错了什么，我怎样才能成功地测试这个类的__init__方法

非常感谢您抽出时间

Answer 1

您有两个问题，shere：

1. 您正在设置get_running_loop的返回值，但异常不是返回值。如果希望模拟代码引发异常，则需要配置side_effect

2. 您的代码捕获了RuntimeError并且没有重新引发is：您只需设置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：要检查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语句执行相同的操作