为异步龙卷风Web套接字服务器编写同步测试套件

Question

我正在尝试为我的龙卷风Web套接字服务器设计一个测试套件.

我正在使用客户端执行此操作 - 通过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中的start test_2).

Tornado提供了一个测试库来帮助进行同步测试,但我似乎无法使用websockets(tornado.ioloop有自己的线程,你无法阻止).

我找不到一个python websocket同步客户端库,它可以使用龙卷风服务器并符合RFC 6455.Pypi的websocket-client无法满足第二个需求.

我的问题是:

1. 是否有可靠的python同步websocket客户端库满足上述要求？

2. 如果没有,那么组织这样的测试套件的最佳方法是什么(测试不能真正并行运行)？

3. 据我所知,由于我们正在使用一个websocket,因此测试用例的IOStream无法分离,因此无法确定响应的请求(我对同一类型的请求进行了多次测试)用不同的参数).我错了吗？

Answer 1

您是否看过龙卷风附带的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装饰允许您运行异步测试案例作为协同程序,其中,龙卷风ioloop内运行时,能有效地使他们的行为同步测试目的.