为异步Tornado WebSocket服务器编写同步测试套件

Question

我正在为我的 Tornado WebSocket 服务器设计一个测试套件。

我使用一个客户端来完成这个工作——通过 WebSocket 连接到服务器，发送请求，并期待得到特定的响应。

我使用 Python 的 unittest 来运行我的测试，所以我不能（也不想）强制测试按特定顺序运行。

这是我的基础测试类的组织方式（所有测试用例都继承自这个类）。(这里省略了日志记录和某些无关的部分)。

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 通信是异步进行的，而 unittest 是同步执行测试的，所以服务器的响应（在同一个 WebSocket 上接收）会和请求混在一起，导致测试偶尔失败。

我知道应该使用回调，但这并不能解决响应混淆的问题。除非所有测试都被人为地按顺序放在一系列回调中（比如在 test_1_callback 中启动 test_2）。

Tornado 提供了一个 测试库 来帮助进行同步测试，但我似乎无法让它与 WebSocket 一起工作（tornado.ioloop 有自己的线程，不能被阻塞）。

我找不到一个与 Tornado 服务器兼容并符合 RFC 6455 的 Python WebSocket 同步客户端库。Pypi 的 websocket-client 不满足第二个要求。

我的问题是：

1. 有没有可靠的 Python 同步 WebSocket 客户端库，符合上述要求？

2. 如果没有，组织这样的测试套件的最佳方式是什么（测试不能真正并行运行）？

3. 据我理解，由于我们只使用一个 WebSocket，测试用例的 IOStreams 不能分开，因此无法确定响应来自哪个请求（我有多个相同类型请求但参数不同的测试）。我这样理解对吗？

Answer 1

你有没有看过tornado里包含的websocket单元测试? 这些测试展示了你可以怎么做：

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这个装饰器可以让你把异步的测试用例当成协程来运行。当这些测试在tornado的ioloop里运行时，它们实际上就像是同步的，这样方便测试。