Synchronous testing of asynchronous server

[kosciej16]

Hi!
I have an FastApi application which uses sio = socketio.AsyncServer(async_mode="asgi")

I am trying to test it with socketio client. I found solution here and tried to adapt it into my case (where my setup often needs to create many clients that emits many signals before I can test real case).

I ended with tons of await asyncio.sleep(some_value) in many places because of race conditions.

Could someone help me and share some hints how can I test my application synchronously? The best would be to not use pytest-asyncio at all and process next emit only after previous one completes.

[miguelgrinberg]

I'm not sure if this is exactly what you need, but replacing your emit() calls with call() in your tests might help.

The emit() call is asynchronous, meaning that the call just schedules the emit and returns. The actual data is sent in the background. If you wanted to know when the emit has reached the client you would have to use the callback option. The call() method combines the emit with the callback and returns only when the callback triggers. It also returns the data sent by the other side as callback arguments.

[kosciej16]

Thank you @miguelgrinberg, works great!

I managed to remove all sleeps instead of one - the one between sio.connect and first sio.emit/sio.call (and as I run sleep in every client creation, it consumes a lot time - I tried to do it once after every client connect, but couldn't make it work).

I found few people had this problem but couldn't find another solution that asyncio.sleep.

Have you any advice in that case too?

[miguelgrinberg]

The sleep between connect and the first emit is not easy to remove. This is something that I hope to address in a future release. There is a discussion of this issue here: #634.

[kosciej16]

I see. Thank you for information, I am looking forward and wait for this issue addressed.

In case I could help you somehow with gathering information/contribute, let me knnow.