如何测试Tornado处理程序正确调用另一个API?

2024-03-28 23:58:17 发布

您现在位置:Python中文网/ 问答频道 /正文

我的Tornado应用程序有一个处理程序,它使用AsyncHTTPClient调用另一个外部API,然后返回。在

如何编写一个单元测试来测试我的处理程序正确调用外部API?在

我觉得我应该在我的单元测试中运行第二个服务器,它模拟外部服务器,并提供与我的处理程序正在命中的相同的API。我可以提供要命中的URI作为应用程序的参数,因此它不是硬编码的。但是,我不确定如何在AsyncHTTPTestCase中(正确地)关闭2个服务器。在


Tags: 服务器api应用程序处理程序编码参数uri单元测试
3条回答

我也有同样的问题。在

这是我在python3.5中的解决方案。在

在处理程序.py在

class VerificationCodeHandler(BaseRequestHandler):

    @asynchronous
    @gen.coroutine
    def post(self):
        code_type = self.body_argument.get('code_type', 'register')
        phone_number = self.body_argument.get('phone_number')
        # 发送阿里云短信
        try:
            # This is another API, encapsulate into a coroutine
            send_result = yield ali_sms.ali_send_code_sms(phone_number, code_type, code)
            http_code, send_sms_result = send_result.code, json.loads(send_result.body.decode('utf-8'))
            if int(http_code) == 200:
                if send_sms_result.get('Code').upper() == 'OK':
                    self.success(self.prompt_msg.AUTH_SEND_MSG_SUCCESS)
                    # 缓存数据
                    self.redis.set(key_lasttime, 'send', ex=settings.SMS_INTERVAL)
                    self.redis.set(key_times,
                                   int(is_overrun) + 1 if is_overrun else 1,
                                   ex=settings.SMS_USER_LIMIT)
                    self.redis.set(key_code, code, ex=settings.SMS_EXPIRATION)
                else:
                    self.fail(send_sms_result.get('Message', self.prompt_msg.AUTH_SEND_MSG_ERROR))
            else:
                self.fail(self.prompt_msg.AUTH_SEND_MSG_ERROR_AGAIN)
        except:
            # TODO 系统异常,需要通知管理员
            logger.exception('发送短信失败')
            self.fail(self.prompt_msg.AUTH_SEND_MSG_ERROR_AGAIN)

阿里_短信.py在

^{pr2}$

在unittest.mock在

class VerificationCodeTestCast(BaseHandelrTestCase):

    @mock.patch('common.aliyun.sms.ali_send_code_sms')
    def test_send_sms_fail(self, sms):
        fetch_future = tornado.concurrent.Future()
        fetch_future._result = mock.Mock(body=json_encode({'Code': 'OK'}).encode('utf-8'), code=200)
        fetch_future._done = True
        sms.return_value = fetch_future
        resp = self.fetch('/get_ver_code/', method='POST',
                      body=json_encode({'code_type': 'register',
                                        'phone_number': '17980888160'})
                      )
        result = json_decode(resp.body)
        self.assertEqual(200, resp.code)
        self.assertEqual(400, result.get('code'))
        self.assertEqual(result['msg'], '今日发送短信超限,请明日再试')

注: 在处理程序中使用ali_send_code_sms时,不要from ali_sms import ali_send_code_smsimport ali_sms然后ali_sms.ali_send_code_sms

如何分拆妥善处理检测,众说纷纭。对于我正在测试必须与第三方对话的内容的场景,在您的示例中,我将它分成3个部分一个非常薄的部分,它只与API进行通信。然后我将编写一个集成测试来测试与API对话的部分。然后,我将制作一个mock或其他测试fixture来替换对API的调用并编写处理API调用的代码。它可以很容易地放在单元测试中,因为它通常类似于:

my_new_data = do_process(call_my_third_party_api())

只需传入第三方api调用返回的数据的模拟,就可以轻松地为do_进程编写测试。在

然后,当需要测试处理程序时,您只需使用返回期望从第三方获得的结果的东西来模拟api调用。在

现在有三个不同的测试来测试应用程序的每个区域。一个可以运行以确保代码正确访问API的测试。一个测试告诉您是否正确地处理了API的处理,第三个测试则告诉您是否正确地将该信息返回给最终用户。在

您可以使用模拟库(unittest.mock在python3.5中):

import unittest
from cStringIO import StringIO

from mock import patch

import tornado.web
import tornado.httpclient
from tornado.testing import AsyncHTTPTestCase


class TestHandler(tornado.web.RequestHandler):
    @tornado.web.asynchronous
    def get(self):
        def request_cb(response):
            self.finish(response.body)

        http_client = tornado.httpclient.AsyncHTTPClient()
        request = tornado.httpclient.HTTPRequest('http://example.com')
        http_client.fetch(request, request_cb)

class TestHttpService(AsyncHTTPTestCase):
    def setUp(self):
        super(TestHttpService, self).setUp()
        self.async_client = tornado.httpclient.AsyncHTTPClient(self.io_loop)
        self.async_patcher = patch('tornado.httpclient.AsyncHTTPClient')
        self.mocked_async_client = self.async_patcher.start()

    def get_app(self):
        return application

    def test_async_httpclient(self):
        request = tornado.httpclient.HTTPRequest('http://example.com')
        response = tornado.httpclient.HTTPResponse(request, 200,
                                buffer=StringIO('test'))
        self.mocked_async_client().fetch.side_effect = lambda x,y: y(response)
        self.async_client.fetch(self.get_url('/'), self.stop)
        resp = self.wait()
        self.assertEquals(resp.body, 'test')

application = tornado.web.Application([(r'/', TestHandler)])

相关问题 更多 >