Search code examples
pythonunit-testingwebsocketcode-coveragetornado

Tornado WebSocket connection is not closed inside a test suite


I'm trying to test this WebSocket handler built with Tornado which I'm calling main.py

import tornado.websocket

class SocketHandler(tornado.websocket.WebSocketHandler):

    def open(self):
        print(' [T] Websocket connection open')

    def on_message(self, message):
        print(' [T] Websocket message received: %s' % message)

    def on_close(self):
        print(' [T] Websocket connection closed')

using the builtin module tornado.testing with some adaptations from a simple AsyncHTTPTestCase example:

import tornado.testing
import tornado.web
import tornado.websocket

from main import SocketHandler

class TestWebSockets(tornado.testing.AsyncHTTPTestCase):

    def get_app(self):
        return tornado.web.Application([(r'/', SocketHandler)])

    @tornado.testing.gen_test
    async def test_async_client(self):

        url = "ws://localhost:" + str(self.get_http_port()) + "/"
        client = await tornado.websocket.websocket_connect(url)
        client.write_message('message')
        client.close()

This test passes. However, when I run pytest --cov-report term-missing --cov=. it points out that the last line of main.py was not reached, i.e. the server was not closed.

Anyone knows what's happening? I already tried adding a delay after closing but with no success.


Solution

  • If I add await tornado.gen.sleep(1) (an async sleep which allows the other async tasks to be processed) after client.close() it brings coverage up to 100%.