Writing a synchronous test suite for an async tornado web socket server

Question

I am trying to design a test suite for my tornado web socket server.

I am using a client to do this - connect to a server through a websocket, send a request and expect a certain response.

I am using python's unittest to run my tests, so I cannot (and do not want to really) enforce the sequence in which the tests are running.

This is how my base test class (after which all test cases inherit) is organized. (The logging and certain parts, irrelevant here are stripped).

class BaseTest(tornado.testing.AsyncTestCase):
    ws_delay = .05

    @classmethod
    def setUpClass(cls):
        cls.setup_connection()
        return

    @classmethod
    def setup_connection(cls):
            # start websocket threads
            t1 = threading.Thread(target=cls.start_web_socket_handler)
            t1.start()

            # websocket opening delay
            time.sleep(cls.ws_delay)

            # this method initiates the tornado.ioloop, sets up the connection
            cls.websocket.connect('localhost', 3333)

        return

    @classmethod
    def start_web_socket_handler(cls):
            # starts tornado.websocket.WebSocketHandler
            cls.websocket = WebSocketHandler()
            cls.websocket.start()

The scheme I came up with is to have this base class which inits the connection once for all tests (although this does not have to be the case - I am happy to set up and tear down the connection for each test case if it solves my problems). What is important that I do not want to have multiple connections open at the same time.

The simple test case looks like that.

class ATest(BaseTest):
    @classmethod
    def setUpClass(cls):
        super(ATest, cls).setUpClass()

    @classmethod
    def tearDownClass(cls):
        super(ATest, cls).tearDownClass()

    def test_a(self):
        saved_stdout = sys.stdout
        try:
            out = StringIO()
            sys.stdout = out
            message_sent = self.websocket.write_message(
                str({'opcode': 'a_message'}})
            )

            output = out.getvalue().strip()
            # the code below is useless
            while (output is None or not len(output)):
                self.log.debug("%s waiting for response." % str(inspect.stack()[0][3]))
                output = out.getvalue().strip()

            self.assertIn(
                'a_response', output,
                "Server didn't send source not a_response. Instead sent: %s" % output
            )
        finally:
            sys.stdout = saved_stdout

It works fine most of the time, yet it is not fully deterministic (and therefore reliable). Since the websocket communication is performed async, and the unittest executes test synchronously, the server responses (which are received on the same websocket) get mixed up with the requests and the tests fail occasionally.

I know it should be callback based, but this won't solve the response mixing issue. Unless, all the tests are artifically sequenced in a series of callbacks (as in start test_2 inside a test_1_callback).

Tornado offers a testing library to help with synchronous testing, but I cannot seem to get it working with websockets (the tornado.ioloop has it's own thread which you cannot block).

I cannot find a python websocket synchronous client library which would work with tornado server and be RFC 6455 compliant. Pypi's websocket-client fails to meet the second demand.

My questions are:

1. Is there a reliable python synchronous websocket client library that meets the demands described above?

2. If not, what is the best way to organize a test suite like this (the tests cannot really be run in parallel)?

3. As far as I understand, since we're working with one websocket, the IOStreams for test cases cannot be separated, and therefore there is no way of determining to which request the response is coming (I have multiple tests for requests of the same type with different parameters). Am I wrong in this ?

Answer 1

Have you looked at the websocket unit tests included with tornado? They show you how you can do this:

from tornado.testing import AsyncHTTPTestCase, gen_test
from tornado.websocket import WebSocketHandler, websocket_connect

class MyHandler(WebSocketHandler):
    """ This is the server code you're testing."""

    def on_message(self, message):
        # Put whatever response you want in here.
        self.write_message("a_response\n")


class WebSocketTest(AsyncHTTPTestCase):
    def get_app(self):
        return Application([
            ('/', MyHandler, dict(close_future=self.close_future)),
        ])  

    @gen_test
    def test_a(self):
        ws = yield websocket_connect(
            'ws://localhost:%d/' % self.get_http_port(),
            io_loop=self.io_loop)
        ws.write_message(str({'opcode': 'a_message'}}))
        response = yield ws.read_message()
        self.assertIn(
                'a_response', response,
                "Server didn't send source not a_response. Instead sent: %s" % response
            )v

The gen_test decorator allows you to run asynchronous testcases as coroutines, which, when run inside tornado's ioloop, effectively makes them behave synchronously for testing purposes.