python - 模拟 Tornado AsyncHTTPClient 请求/响应

Question

我有一个 Tornado Web 应用程序，它向外部服务发送请求，但似乎无法模拟这些请求的响应。

我尝试过 tornado-http-mock 和 mock 库，但没有成功。

应用程序起始点app.py:

import tornado.ioloop
import tornado.web

from handlers import HealthCheckHandler, MainHandler, LoginHandler, PaymentHandler, UserDetailsHandler
from settings import PORT, tornado_settings


def make_app(settings):
    handlers = [
        ('/static/(.*)', tornado.web.StaticFileHandler, {'path': './public'}),
        ('/', MainHandler),
        ('/health', HealthCheckHandler),
        ('/login', LoginHandler),
        ('/user', UserDetailsHandler),
        ('/payment', PaymentHandler),
    ]
    return tornado.web.Application(handlers=handlers, **settings)


if __name__ == '__main__':
    app = make_app(tornado_settings)
    app.listen(PORT)
    tornado.ioloop.IOLoop.current().start()

我正在尝试测试登录功能(它是一个 OAuth2 服务器)，当没有传递 code GET 参数(如果用户尚未登录)时，该功能会重定向用户，或者它尝试用访问 token 交换代码。这是登录处理程序。

import base64
import urllib.parse
import json
import traceback
import tornado.web
import tornado.httpclient

from .base import BaseHandler
from settings import OID_AUTH_API, OID_REDIRECT_URI, OID_CLIENT_ID, OID_CLIENT_PASSWORD
from lib import logger


class LoginHandler(BaseHandler):
    _redirect_uri = urllib.parse.quote(OID_REDIRECT_URI, safe='')
    _scope = 'openid+profile+email'
    _response_type = 'code'
    _http_client = tornado.httpclient.AsyncHTTPClient()

    async def get(self):
        try:
            code = self.get_argument('code', None)

            if (code is None):
                self.redirect('%s/authorization?client_id=%s&scope=%s&response_type=%s&redirect_uri=%s' % (
                    OID_AUTH_API, OID_CLIENT_ID, self._scope, self._response_type, self._redirect_uri), self.request.uri)
                return

            # exchange the authorization code with the access token
            grant_type = 'authorization_code'
            redirect_uri = self._redirect_uri
            authorization_header = '%s:%s' % (
                OID_CLIENT_ID, OID_CLIENT_PASSWORD)
            authorization_header_encoded = base64.b64encode(
                authorization_header.encode('UTF-8')).decode('UTF-8')
            url = '%s/token?grant_type=%s&code=%s&redirect_uri=%s' % (
                OID_AUTH_API, grant_type, code, redirect_uri)
            token_exchange_response = await self._http_client.fetch(
                url,
                method='POST',
                headers={
                    'Content-Type': 'application/x-www-form-urlencoded',
                    'Authorization': 'Basic %s' % authorization_header_encoded,
                    'Accept': 'application/json'
                },
                body='')

            token_exchange_response_body_dict = json.loads(
                token_exchange_response.body)

            access_token = token_exchange_response_body_dict.get('access_token')

            self.send_response({
                'access_token': access_token
            })
        except Exception as error:
            logger.log_error_with_traceback(error)
            self.send_response({
                'success': False,
                'message': 'Internal server error. Please try again later.'
            }, 500)

我有两个问题: 1. 如果没有提供授权码，如何测试重定向功能？ 2. 这种情况下如何模拟对OAuth2服务器的请求？

我尝试使用tornado-http-mock，但出现错误。

import app
import json

from tornado.httpclient import HTTPClient, HTTPResponse, HTTPRequest, HTTPError
from tornado.testing import AsyncHTTPTestCase, AsyncTestCase, AsyncHTTPClient

from tornado_mock.httpclient import get_response_stub, patch_http_client, set_stub

from .base import TestHandlerBase
from settings import OID_AUTH_API


class TestLoginHandler(AsyncHTTPTestCase):
    def get_app(self):
        test_app = app.make_app({})
        self.app_http_client = test_app.http_client = AsyncHTTPClient(force_instance=True)
        return test_app

    def test_token_code_exchange(self):        
        patch_http_client(self.app_http_client)

        set_stub(self.app_http_client, '%s/token' % (OID_AUTH_API), request_method='POST', response_body='oauth_server_token')

        response = self.fetch('/login?code=123')

        self.assertEqual(response.code, 200)
        print(response.body)

我收到以下错误，表明不支持 POST 方法(看起来请求实际上正在发送到服务器而不是被模拟)。

Login Error code: 405 | Response body: 
[E 190626 13:29:33 web:2246] 500 GET /login?code=123 (127.0.0.1) 238.10ms
======================================================================
FAIL: test_token_code_exchange (tests.handlers.login.TestLoginHandler)
----------------------------------------------------------------------

Traceback (most recent call last):
  File "/Users/.../venv/lib/python3.7/site-packages/tornado/testing.py", line 98, in __call__
    result = self.orig_method(*args, **kwargs)
  File "/Users/.../tests/handlers/login.py", line 60, in test_token_code_exchange
    self.assertEqual(response.code, 200)
AssertionError: 500 != 200

我期待得到 stub 回复，但显然，我没有得到。我在这里缺少什么？还有其他解决办法吗？

Answer 1

您可以使用mock.patch和gen.coroutine来模拟tornado中的外部请求。 你可以尝试这样的事情:

将外部请求提取到类似...的方法

async def new_fetch(self, url, authorization_header_encoded):
    return await self._http_client.fetch(
        url,
        method='POST',
        headers={
            'Content-Type': 'application/x-www-form-urlencoded',
            'Authorization': 'Basic %s' % authorization_header_encoded,
            'Accept': 'application/json'
        },
        body='')

更改您的 LoginHandler 以调用此新方法:

token_exchange_response =等待new_fetch(url，authorization_header_encoded)

在您的 TestLoginHandler 中创建一个新方法来模拟 http 响应，并使用 gen.coroutine 装饰器装饰此方法，并使用 mock.patch 装饰器在您的测试方法上模拟外部请求方法:

import app
import json

from tornado.httpclient import HTTPClient, HTTPResponse, HTTPRequest, HTTPError
from tornado.testing import AsyncHTTPTestCase, AsyncTestCase, AsyncHTTPClient
# from tornado_mock.httpclient import get_response_stub, patch_http_client, 
from tornado import gen 

from .base import TestHandlerBase
from settings import OID_AUTH_API


class TestLoginHandler(AsyncHTTPTestCase):
    @gen.coroutine
    def mock_fetch(self, url, authorization_header_encoded)
        request = HTTPRequest(
            headers=authorization_header_encoded,
            method='POST',
            body='',
            url=url)
        resp = HTTPResponse(request, HTTPStatus.OK, buffer=json.dumps({}))
        resp._body = json.dumps({"YOUR_RESPONSE_BODY":"YOUR_RESPONSE_BODY"})
        return resp

    def get_app(self):
        test_app = app.make_app({})
        # self.app_http_client = test_app.http_client = AsyncHTTPClient(force_instance=True)
        return test_app

    @mock.patch("full.path.to.LoginHandler.new_fetch")
    def test_token_code_exchange(self, mocked_method):
        mocked_method.return_value = self.mock_fetch('optional_url', 'optional_header') 

        # patch_http_client(self.app_http_client)

        # set_stub(self.app_http_client, '%s/token' % (OID_AUTH_API), request_method='POST', response_body='oauth_server_token')

        response = self.fetch('/login?code=123')

        self.assertEqual(response.code, 200)
        print(response.body)

我没有测试这段代码，我编写只是为了向您传递制作此模拟的想法，所以也许您需要调整这段代码中的一些内容