Logo Questions Linux Laravel Mysql Ubuntu Git Menu
 

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

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.

Client class code:

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()

Unit test function is below:

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')

Error I am getting:

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)
like image 959
AnilJ Avatar asked Mar 11 '23 13:03

AnilJ


1 Answers

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.

like image 191
Shuo Feng Avatar answered Apr 25 '23 18:04

Shuo Feng