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

Question

I am trying to create a test suite for my torrent web socket server.

I use the client for this - connect to the server through the website, send a request and expect a specific response.

I use python unittest to run my tests, so I cannot (and do not want to really) use the sequence in which the tests are executed.

This is how my base test class works (after which all test cases are inherited). (Logging and some parts that are irrelevant here are devoid of).

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()

, , , , ( - , ). , .

.

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

, (, , ). websocket async, unittest , ( -) , .

, , . , ( start test_2 test_1_callback).

Tornado , , websockets ( tornado.ioloop , ).

python websocket, RFC 6455. Pypi websocket-client .

:

• - python, , ?

• , , ( )?

• , websocket, IOStreams , , ( ). ?

Answer 1

-, ? , :

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

gen_test , ioloop .