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

Question

In a source code (source link here and WIP PR here) I am working with, I am trying to improve test coverage by testing a try-except block in a class' __init__ method.

Stripping off the extra code from the source, the relevant code looks like:

# 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()

In a separate test file, I'd like to mock RuntimeError to test the try except block:

# 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()

Neither test would pass because neither would raise RuntimeError.

Additionally, I tried to mock asyncio.new_event_loop with monkeypatch:

# 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

This test also fails because the monkey patch is never called: > assert mock_new_loop.call_count == 1 E assert 0 == 1.

I'm wondering what I did wrong here and how could I successfully test the __init__ method for this class?

Thanks a lot for your time!

Erminia

Answer 1

You have two problem shere:

1. You're setting the return value of get_running_loop, but an exception isn't a return value. If you want your mocked code to raise an exception, you need to configure a side_effect.

2. Your code catches the RuntimeError and doesn't re-raise is: you simply set self.loop = None and log an error. This means that even when you successfully raise a RuntimeError from get_event_loop, that exception will never be visible to your tests because it is consumed by your code.

If you were to mock your logger object, you can check that logger.error was called with the exception. E.g.:

@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 to checking the self.loop = None part, I would probably rewrite the code like this:

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()

And then when testing, you would need to mock a return value for new_event_loop. I would probably get rid of the nested with statements and just use the patch decorator on the function instead:

@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'

...but obviously you could do the same thing with either a series of nested with patch(...) statements or a single long with statement.

Cube

I've updated the answer to address your comment. – Cube

Thanks. I deleted my original comment as the question wasn't really valid (I was not the author of the source code and did not fully digest the code). I could also put monkeypatch directly in the first code you suggested with mock_new_loop = Mock() and monkeypatch.setattr(asyncio, "new_event_loop", mock_new_loop) and assert mock_new_loop.call_count == 1. That seems to pass as well. But your suggestion below is even better :) – Erminia