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

Question

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

客户class代码：

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函数。

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

Unit testing using Tornado Websocket - no attribute 'io_loop' error

python

unit-testing

tornado

python-unittest

客户class代码：

单元测试函数如下：

我得到的错误：