How To Use A Test Tornado Server Handler That Authenticates A User Via A Secure Cookie

How can I write a unit test for a tornado handler that authenticates a user via a secure cookie? Here is the code (and sudo code) for a dummy test that I'd like to make pass. I'm u

Solution 1:

Using mock:

import mock

...

classUserAPITest(AsyncHTTPTestCase):
    defget_app(self):
        self.app = Application([('/', MainHandler)],
                    cookie_secret='asdfasdf')
        return self.app

    deftest_user_profile_annoymous(self):
        with mock.patch.object(MainHandler, 'get_secure_cookie') as m:
            m.return_value = 'user_email'
            response = self.fetch('/', method='GET')
        self.assertEqual('sucess', to_unicode(response.body) )

Solution 2:

It seems you may try to use a create_signed_value function from tornado.web module:

from tornado.web import create_signed_value

classUserAPITest(AsyncHTTPTestCase):

    defget_app(self):
         self.app = Application([('/', MainHandler)],
                                cookie_secret='asdfasdf')
         return self.app

    deftest_user_profile_annoymous(self):
        cookie_name, cookie_value = 'Cookie', 'value'
        secure_cookie = create_signed_value(
            self.app.settings["cookie_secret"],
            cookie_name,
            cookie_value)
        headers = {'Cookie': '='.join((cookie_name, secure_cookie))}

        response = self.fetch('/', method='GET', headers=headers)
        self.assertEqual('success', response.body)

Solution 3:

If you want to import excel through tornado REST API or want to do Unit Test the handler this may be helpful to you.It's working perfectly form my application. Here I am passing excel through form-data header in REST.

If want to write test cases for other Rest API's using cookies then it may be helpful.

import json
import os.path
import requests
import sys
import time
import tornado.testing
import tornado.web
import unittest
import unittest.mock as mock
from http import cookies
from tornado.options import define, options, parse_command_line

from portal.handlers import ListHandler,excelimportHandler   # handlers created in tornado

APP_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))  # root location where your file is existing
sys.path.append(os.path.join(APP_ROOT, '..'))  # joining file with root pathfrom portal.library.helpers import get_env_conf, read_env_conf  # functionality to get and resd the the environment config file.
os.environ["ASYNC_TEST_TIMEOUT"] = str(60)  # It will wait for the API for response

LIST = '/data/list'
IMPORT_EXCEL = '/import/excel'classProxyServerAPITest(tornado.testing.AsyncHTTPTestCase):
    def__init__(self, *rest):
        self.cookie = cookies.SimpleCookie()
        define("env", default="localtest", help="Environment")
        define("env_conf_path", default="", help="Environment config file path")
        parse_command_line()
        '''
        reading the env config file for unit test cases
        '''if options.env_conf_path:
            env_conf_path = options.env_conf_path
            env_conf = read_env_conf(env_conf_path)
        else:
            env_conf = get_env_conf(options.env)
        self.api_engine_address = env_conf.get("api_engine_address")
        self.api_token = env_conf.get("api_token")
        self.api_username = env_conf.get("api_username")
        self.elasticsearch_index = env_conf.get("elasticsearch_index")

        tornado.testing.AsyncHTTPTestCase.__init__(self, *rest)

    defget_app(self):

        env = mock.MagicMock(return_value='xyz') # mocking config filedefget(key):
            """
            whenever in handlers, if these parameters will be called from there config files
            it will take the parameters
            of config file of unit test cases.
            """if key == "api_username":
                return self.api_username
            elif key == "api_engine_address":
                return self.api_engine_address
            elif key == 'api_token':
                return self.api_token
            elif key == 'elasticsearch_index':
                return self.elasticsearch_index
        env.get = get
        application = tornado.web.Application(
            [
             (LIST, ListHandler, dict(env_conf=env)),
             (IMPORT_EXCEL, excelimportHandler, dict(env_conf=env)),
             ],
            cookie_secret="cookie-secret")
        return application

    def_update_cookies(self, headers):
        """
        It will take the cookies that will be passed in whole application.
        """try:
            sc = headers['Set-Cookie']
            tempcookies = tornado.escape.native_str(sc)
            tempcookies_dict = tempcookies.split('/,')
            self.cookie.update(
                cookies.SimpleCookie(tempcookies_dict[0].split(';')[0] + ';' + tempcookies_dict[1].split(';')[0] + ';' +
                                     tempcookies_dict[2].split(';')[0]))
        except KeyError:
            returndeffetch(self, url, new_headers=None, *r, **kw):
        """
        This function will set the headers and cookies for the Tornado application
        """if'follow_redirects'notin kw:
            kw['follow_redirects'] = False

        header = {}
        hs = self.cookie.output()
        header['Cookie'] = hs
        if hs != '':
            hs = hs.split('\r\n')
            hs = hs[0].split('-Cookie:')[1] + ";" + hs[1].split('-Cookie:')[1] + ";" + hs[2].split('-Cookie:')[1]
            header['Cookie'] = hs
        if new_headers:
            """
            These headers will be used if we wiil deal with excel file.
            """
            header.update(new_headers)

        resp = tornado.testing.AsyncHTTPTestCase.fetch(self, url, headers=header, *r, **kw)
        self._update_cookies(resp.headers)
        return resp
    deffind_body_headers_of_excel(self, file_name):
        """
        This function will provide the encrypted form of the excel in body and proper headers
        that to be passed in API to import the excel using REST.
        """
        fpath = os.path.join(os.path.dirname(__file__), file_name)
        f = open(fpath, 'rb')
        files = {"file": f}
        data = {}

        a = requests.Request(method='POST', url="http://abc/img/test",
                             files=files, data=data)
        prepare = a.prepare()
        content_type = prepare.headers.get('Content-Type')
        body = prepare.body

        headers = {
            "Content-Type": content_type,
        }
        return headers, body
    deftest_proxy_api(self):

        headers, body = self.find_body_headers_of_excel('excel.xlsx')
        resp = self.fetch(IMPORT_EXCEL, method='POST', body=body, connect_timeout=1000, request_timeout=1000,
                          new_headers=headers)
        self.assertEqual(200, resp.code)
        resp = self.fetch(LIST, method='GET', connect_timeout=1000, request_timeout=1000)
        self.assertEqual(200, resp.code)

if __name__ == '__main__':
    unittest.main()