Is there a test client?

[jacopofar]

Hello and thanks for the library!

I'm using it with Starlette and trying to implement some integration test. Is there a test client for socketio similar to the one they provide for the basic HTTP/websocket (here), or examples about how to implement such a test?

[miguelgrinberg]

There is currently no test client. I have one for the Flask integration with this package, it would be nice to have something similar that is generic.

The best approximation is to use a real server with a real client, both possibly running within the same process.

[jacopofar]

Thanks. I tried to implement the test in such a way, but cannot manage to have the ASGI server running in a separate thread or process while the test client connects to it.

My test looks like this:

@pytest.mark.asyncio
async def test_websocket():
    config = uvicorn.Config(app.create_app(), host='localhost', port=8000)
    server = uvicorn.Server(config)
    loop = asyncio.get_running_loop()

    serve_coroutine = await server.serve()
    executor = concurrent.futures.ProcessPoolExecutor(
        max_workers=5,
    )
    loop.run_in_executor(executor, serve_coroutine)

    # this line is reached only when I press Ctrl-C and kill the server

    async def connect_to_ws():
        sio = socketio.Client()
        sio.connect('http://localhost:8000')
        # here would go assertions on the socket responses

    k = await connect_to_ws()
    loop.run_in_executor(executor, k)

the app.create_app function is:

def create_app():

    app = Starlette(debug=True)
    app.mount('/', StaticFiles(directory='static'), name='static')

    sio = socketio.AsyncServer(async_mode='asgi')
    extended_app = socketio.ASGIApp(sio, app)

    # here define HTTP and Socketio handlers

    return extended_app

the basic idea is to start the complete server and just connect to it, I assumed I could run the server and the test client in the same event loop but apparently when I run the server (that is indeed started and I can reach with the browser) it blocks the test code. Only when I use Ctrl-C to stop it the server is killed and the rest of the test runs but of course it doesn't find the server.

Probably I'm missing something essential here, I expected the server and the test client to run concurrently on the same event loop without need for multithreading or multiprocessing.

[miguelgrinberg]

In your example you are using a process executor, so you are in fact using multiprocessing there. I think this can be done in a much simpler way. Here is a rough attempt that appears to be work well:

import asyncio
import socketio

sio = socketio.AsyncServer(async_mode='asgi', monitor_clients=False)
app = socketio.ASGIApp(sio)

def start_server():
    import asyncio
    from uvicorn import Config, Server
    config = Config(app, host='127.0.0.1', port=5000)
    server = Server(config=config)
    config.setup_event_loop()
    loop = asyncio.get_event_loop()
    server_task = server.serve()
    asyncio.ensure_future(server_task)
    return server_task

async def run_client():
    client = socketio.AsyncClient()
    await client.connect('http://localhost:5000')
    await asyncio.sleep(5)
    await client.disconnect()

start_server()
loop = asyncio.get_event_loop()
loop.run_until_complete(run_client())

Hopefully this will get you started.

[jacopofar]

Thanks a lot! Indeed from this example I was able to make it work :)

For whoever will encounter the problem in the future, in case someone in the future is interested this is my implementation:

The app:

def create_app():

    app = Starlette(debug=True)
    app.mount('/', StaticFiles(directory='static'), name='static')

    sio = socketio.AsyncServer(async_mode='asgi')
    extended_app = socketio.ASGIApp(sio, app)

    @sio.on('double')
    async def double(sid, data):
        logging.info(f"doubling for {sid}")
        return 'DOUBLED:' + data * 2
   # here add HTTP and WS handlers...
   return extended_app

The test, based on pytest-asyncio, uses an async ficture to start and stop the server has this structure:

import asyncio

import socketio
import uvicorn
import pytest

from myapp import app

def get_server():
    config = uvicorn.Config(app.create_app(), host='localhost', port=8000)
    server = uvicorn.Server(config=config)
    config.setup_event_loop()
    return server


@pytest.fixture
async def async_get_server():
    server = get_server()
    server_task = server.serve()
    asyncio.ensure_future(server_task)

@pytest.mark.asyncio
async def test_websocket(async_get_server):
    client = socketio.AsyncClient()
    await client.connect('http://localhost:8000')
    result = await client.call('double', 'hello')
    assert result == 'DOUBLED:hellohello'
    await client.disconnect()

This work although it produces a lot of warnings (I suspect some problem with the logs).

A weird thing I noticed is that to run this it is required to install aiohttp, if not I get timeout errors. Could it be worth to raise an explicit error in this case? I can try and do a PR if it's fine for you

[miguelgrinberg]

The aiohttp package provides the WebSocket client, without it the connection stays as long-polling. Not sure why the timeouts without it however, I'll have to test that.

[nbanmp]

I'm at a point where having a test client with socketio would be really helpful for me too. Is there any update on the progress of this?

[miguelgrinberg]

@nbanmp I am not currently working on a test client. The main reason is that there is a real Python client now. Is there anything that prevents you from using the real client against your real server for tests?

[nbanmp]

Thanks for the update.

Running the real client against the real server has some issues, the main one for me is that it is more difficult to run multiple tests asynchronously. But also important is that, I would like my unit tests to be as independent as possible, and I was expecting to run the actual server for integration testing.

[miguelgrinberg]

In any case, the test client that exists in the Flask-SocketIO package is very limited, if/when I get to do a generic test client it would be based on the real client talking to the real server. It would make it easier to start/stop the server for each test, but other than that I expect it will be the real thing, not a fake.

[databasedav]

There should to be a way to gracefully shutdown the server given the setup above. This would need to cancel the _service_task and disconnect the remaining connect clients. I can make this contribution but need some guidance on a few things, particularly how the _service_task is started from a socketio.AsyncServer; I can find the task being started in engineio.AsyncServer but not in the former.

[miguelgrinberg]

@databasedav The service task does not need to run when testing. A testing set up can be made by subclassing the Server and Client classes (and their asyncio counterparts) to re-implement the networking parts through direct calls.

[databasedav]

@miguelgrinberg I agree; I was just talking in the context of using a live server like discussed above