Skip to main content
 首页 » 编程设计

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

2025年02月15日6cloudgamer

我已经拼接了一个 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) 

请您参考如下方法:

你有自己的setUp功能吗?

io_loop是在AsyncTestCase的setUp函数下创建的,我想你需要调用super的setUp函数。