I have stitched together a tornado websocket client code and using it in my python unit test case. This is my first time use of tornado websocket and not very familiar with its unit test API. Looking for some help to understand the use of tornado websocket asynchronous unit test code and the below case working.
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)
Did you have your own setUp
function?
The io_loop
is created under AsyncTestCase's setUp function, I think you need to call super's setUp
function.
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With