Testing Asyncio With Pytest: How To Test A Try-except Block By Mocking The Event Loop?

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

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

    withpatch("webrtc.logger") as mock_logger:
        withpatch("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)

    ┆   ifself.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.