The usage of global asyncio event loop makes async unit tests to leak their exceptions

[kummahiih]

Currently the usage of the global asyncio event loop ( asyncio.get_event_loop()) makes writing unit tests pretty hard, because failures on one unit test usually leak to another test if they are executed in the same pytest instance. There might also be some security implications of leaking exceptions, but I can not currently figure out a concrete example for you.

I have used the following unit test base class on some of my async tests:

import asyncio
import unittest

class AsyncTest(unittest.TestCase):
    """
    Async testcases need exception handling too which is easy to forget.
    """
    def setUp(self):
        self._loop = asyncio.new_event_loop()
        asyncio.set_event_loop(None)
        
        def handleException(loop, context):
            self.loop.stop()
            self.fail("test failed to an exception on async callback")        
        self.loop.set_exception_handler(handleException)
        
    @property
    def loop(self):
        return self._loop

class MyIsolatedTest(AsyncTest):
    def test_something()
        async def test_a():
            await something_a()
        self.loop.run_until_complete(
                    asyncio.wait_for(
                        test_a(),
                        5, # timeout 5 s
                        loop=self.loop))

but with pysnmp this can not work, because you use the global async event loop which makes the library ununittestable.

Suggestions:

• (the good) provide a way to give the event loop as a parameter on initialization
• (the bad) help me to write async unit tests so that I could test my code
• (the ugly) create and use a global event loop for your library so that only the pysnmp related code leaks exceptions to wrong places

[etingof]

Thank you for raising this concern!

I'd happily proceed the good way indeed!

You can speed this up by proposing the changes to pysnmp you have in mind. Either a PR or just general hint how you think it would be best to pass event loop object to the initializer.

Thanks! ;-)

[kummahiih]

It's quite big change to add the loop into all classes and functions which use asyncio loop, but I try to find the time to make you some sort of proposal. First thoughts about implementation details:

For classes I would implement:

import abc
class IWithAsyncLoop(metaclass=abc.ABCMeta):
    @abc.abstractproperty
    def loop(self) -> asyncio.AbstractEventLoop:
        raise NotImplemented()

and for functions (including the constructor) at start something like this:

def example_function(loop: asyncio.AbstractEventLoop=None):
    loop = loop if loop is not None else asyncio.get_event_loop()

and when the loop is carried to all places, it could be refactored into

def example_function(loop: asyncio.AbstractEventLoop=None):
    if loop is None:
        raise ValueError("loop should not be None")

To make this refactoring smooth I would suggest to define a function:

def check_async_loop(loop: asyncio.AbstractEventLoop):
    return loop if loop is not None else asyncio.get_event_loop()

and also I would make some unit tests like the one above.

[kummahiih]

Your thoughts @etingof ?