how to clean up between each test to prevent RuntimeError: Event loop is closed

[nbraun-wolf]

Hi, I have an issue, with some code. I am importing a client from a separate file since it's basically a global dependency.
I am already importing the code only in an app factory, but it doesn't help.

Between individual pytest functions, it seems to use the client from the previous function. I am not a python pro, but I assume the client sticks around in some kind of import the import table.

I have seen to some time ago with a different package as well actually. I am looking for a way to tear down everything and clear every module that may still stick around. You can see the issue in action in this example project. https://github.com/nbraun-wolf/async-mqtt-fastapi-pytest.

[agronholm]

You need to tie the client creation and shutdown to the application or test life cycle, just like you do with the HTTP client. Creating an async client as an import side effect is not going to work very well.

[nbraun-wolf]

I made one branch where I used this clousure type of thing to solve the problem. It works, but I am not sure if this is the pythonic way of doing it. https://github.com/nbraun-wolf/async-mqtt-fastapi-pytest/tree/closure

[agronholm]

While I'm not sure what the 100% clean solution here is, creating a new client in create_app() and setting it as a global variable would likely work for you.

[nbraun-wolf]

Sorry, I am not sure if I understand what you mean. How do I set it as a global variable, and how will it enable other files to import and use it?

I just did, something like this as a test. But I could not import client from main in any other file.

def create_app():
    from asyncio_mqtt import Client
    global mqtt
    mqtt = Client("localhost")
    ...

Declaring the client first, at the top, did also not work. I got errors when importing and using it that none type doesn't have this or that method.

mqtt = None
def create_app():
    from asyncio_mqtt import Client
    global mqtt
    mqtt = Client("localhost")
    ...

[agronholm]

Trying to directly import it won't work since it hasn't been defined yet because at import time create_app() has not been called yet. So you should just import the module and then use it like yourmodule.mqtt.... You can even use the FastAPI dependency system to make a function that returns you the client (using the method above). Like so:

from yourpackage import mainmodule

def mqtt_client():
    return mainmodule.mqtt

And in your endpoints:

@router.get('/whatever')
async def myendpoint(mqtt: AsyncMQTTClient = Depends(mqtt_client)):
    ...

[nbraun-wolf]

Oh, I see. This works, actually. Thanks a lot for looking into this and solving the issue!