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