python - 使用 Tornado Websocket 进行单元测试 - 无属性 'io_loop' 错误

Question

我已经拼接了一个 tornado 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函数下创建的，我想你需要调用super的setUp函数。