使用 Tornado Websocket 进行单元测试 - 没有属性“io_loop”错误

Question

我已经将一个龙卷风 websocket 客户端代码拼接在一起，并在我的 python 单元测试用例中使用它。这是我第一次使用 tornado websocket，对它的单元测试 API 不是很熟悉。寻找一些帮助来理解 tornado websocket 异步单元测试代码的使用和下面的案例工作。

客户端类代码：

import logging
import logging.config
import ssl
import time
import traceback

from tornado.concurrent import Future
from tornado import gen
from tornado.httpclient import HTTPError, HTTPRequest
from tornado.log import gen_log, app_log

from tornado.web import Application, RequestHandler

class TorWebSocketClient():

    def __init__(self, ip_addr, port, cookie):
        self.ip_addr = ip_addr
        self.port = port
        self.cookie = cookie
        self.sockConnected = False
        self.logger = logging.getLogger(__name__)

    def Connect(self):

        # Creating the websocket client for each test case.
        url = "ws://{0}:{1}/socket{2}".format(str(self.ip_addr), str(self.port), self.cookie)
        self.logger.debug('Websocket URL: ' + url)
        sslopt={"cert_reqs": ssl.CERT_NONE,
                "check_hostname": False,
                "ssl_version": ssl.PROTOCOL_TLSv1}

        self.logger.debug('New web socket connection is being establshed by the client')
        self.ws = websocket.websocket_connect(HTTPRequest(url, headers=headers, ssl_options=sslopt), io_loop=self.io_loop)

        # Start the websocket client thread. A wait is added till connection is established.
        self.sockConnected = True

    def send(self, data):
        # Wait till websocket is connected.
        if not self.ws.sock.connected:
            self.logger.debug('Send failed; Websocket connection is not yet established')
            return

        self.logger.info('Sending data to the server: ' + data)
        self.ws.write_message(data)

    def recv(self, expValues):    
        # Read data from the response message.
        resp = yield self.ws.read_message()
        print '>>>> Response: ', resp

    def stop(self):
        self.logger.debug('Client closing the websocket connection with the server')
        self.ws.close()

单元测试功能如下：

import functools
import json
import logging
import logging.config
import time

# These are couple of custom classes. 
import TorWebSocketClient
from infra.serverbase import Server

from tornado.testing import AsyncHTTPTestCase, gen_test, bind_unused_port, ExpectLog

class TornadoTest(AsyncHTTPTestCase):

def get_app(self):
    app = tornado.web.Application([('/', EchoWebSocketHandler)]) 
    return app

@gen_test
def testToradoWSConection(self):

    # Login to the server to get the cookie. 
    logger = logging.getLogger(__name__)
    server = Server(self.ipaddr, self.port, self.username, self.password)

    result = server.Login()
    self.assertEqual(result, True, 'login failed')

    webSocClient = yield TorWebSocketClient(self.ipaddr, self.port, server.GetCookie())
    result = webSocClient.Connect()
    self.assertEqual(result, True, 'Websocket connection failed')

我得到的错误：

Traceback (most recent call last):
  File "/users/usr1/pyvenv/venv/lib/python2.7/site-packages/tornado/testing.py", line 527, in post_coroutine
    return self.io_loop.run_sync(
AttributeError: TornadoTest instance has no attribute 'io_loop'
----------------------------------------------------------------------
Ran 1 tests in 0.002s

FAILED (errors=1)

Answer 1

你有自己的setUp功能吗？

该io_loop下AsyncTestCase的设置功能创建的，我想你需要调用超类的setUp功能。