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.
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%.