使用龙卷风测试AsyncHTTPTestCase连接超时

Question

我有单元测试龙卷风应用程序的问题，请帮助我。错误堆栈跟踪：

Error Traceback (most recent call last): File "/Users/doc/python/lib/python3.5/site-packages/tornado/testing.py", line 432, in tearDown timeout=get_async_test_timeout()) File "/Users/doc/python-virt1/lib/python3.5/site-packages/tornado/ioloop.py", line 456, in run_sync raise TimeoutError('Operation timed out after %s seconds' % timeout) tornado.ioloop.TimeoutError: Operation timed out after 5 seconds

ERROR:tornado.application:Future exception was never retrieved: Traceback (most recent call last): File "/Users/doc/python/lib/python3.5/site-packages/tornado/gen.py", line 1021, in run yielded = self.gen.throw(*exc_info) File "/Library/Frameworks/Python.framework/Versions/3.5/lib/python3.5/types.py", line 179, in throw return self.__wrapped.throw(tp, *rest) File "/Users/doc/python-virt1/lib/python3.5/site-packages/tornado/gen.py", line 1015, in run value = future.result() File "/Users/doc/python-virt1/lib/python3.5/site-packages/tornado/concurrent.py", line 237, in result raise_exc_info(self._exc_info) File "", line 3, in raise_exc_info tornado.curl_httpclient.CurlError: HTTP 599: Empty reply from server Traceback (most recent call last):

test.py文件：

from tornado.testing import gen_test 
from tests.api_tests.base import AbstractApplicationTestBase 


class ApiRestTest(AbstractApplicationTestBase): 
    def setUp(self): 
     super(ApiRestTest, self).setUp() 
     self.prepareDatabase(self.config) 
     self.insert_user(config=self.config) 

api_test/base.py

import logging 
from api import server 
from commons.constants import config 
from tests.base import BaseTestClass 


class AbstractApplicationTestBase(BaseTestClass): 
    def get_app(self): 
     application = server.get_application(self.config) 

     application.settings[config.APPLICATION_DB] = self.db 
     application.settings[config.APPLICATION_CONFIG] = self.config 
     application.settings[config.APPLICATION_AES] = self.aes 
     application.settings[config.APPLICATION_FS] = self.fs 
     logging.info(self.config.DEPLOY_API) 

     return application 

测试/ base.py

import logging 
from datetime import datetime 
import motor.motor_tornado 
from motor import MotorGridFSBucket 
from pymongo import MongoClient 
from tornado import escape 
from tornado import gen 
from tornado.testing import AsyncHTTPTestCase 

class BaseTestClass(AsyncHTTPTestCase): 
    @classmethod 
    def setUpClass(self): 
     super(BaseTestClass, self).setUpClass() 
     self.config = Config(Environment.TESTS.value) 
     self.client = utils.http_client(self.config.PROXY_HOST, self.config.PROXY_PORT) 
     self.db = motor.motor_tornado.MotorClient(self.config.MONGODB_URI)[self.config.MONGODB_NAME] 
     self.fs = MotorGridFSBucket(self.db) 

Answer 1

我注意到了一些事情。

• 我看到的主要问题是你在你的setUp方法中通过电机做了一些IO，并且setUp不能是gen_test（AFAIK）。如果你需要这种类型的功能，你可能需要下载到pymongo并同步调用数据库来存储这些数据库设备。
• 你有意对真实的数据库运行吗？这些应该是与mongodb真正的集成测试吗？当我编写这些类型的测试时，我通常会使用Mock类并嘲笑我与MongoDb的所有交互。
• 另外，创建一个AsyncHttpClient对象没有任何代价，因此将它从settings/config对象传递到每个处理程序中可能不是最佳实践。

下面是我的一个项目为例处理机测试：

example_fixture = [{'foo': 'bar'}] 
URL = r'/list' 

    class BaseListHandlerTests(BaseHandlerTestCase): 
     """ 
     Test the abstract list handler 
     """ 
     def setUp(self): 
      self.mongo_client = Mock() 
      self.fixture = deepcopy(example_fixture) 
      # Must be run last 
      BaseHandlerTestCase.setUp(self) 

     def get_app(self): 
      return Application([ 
       (URL, BaseListHandler, 
       dict(mongo_client=self.mongo_client)) 
      ], **settings) 

     def test_get_list_of_objects_returns_200_with_results(self): 
      self.mongo_client.find.return_value = self.get_future(example_fixture) 
      response = self.fetch('{}'.format(URL)) 
      response_json = self.to_json(response) 
      self.assertListEqual(response_json.get('results'), example_fixture) 
      self.assertEqual(response.code, 200) 

     def test_get_list_of_objects_returns_200_with_no_results(self): 
      self.mongo_client.find.return_value = self.get_future([]) 
      response = self.fetch('{}'.format(URL)) 
      self.assertEqual(response.code, 200) 

     def test_get_list_of_objects_returns_500_with_exception(self): 
      self.mongo_client.find.return_value = self.get_future_with_exception(Exception('FAILED!')) 
      response = self.fetch('{}'.format(URL)) 
      self.assertEqual(response.code, 500) 

以使这项工作的关键是，我的mongo_client传递到路由对象本身。所以我的处理程序初始化需要一个mongo_client kwarg。

class BaseListHandler(BaseHandler): 
    """ 
    Base list handler 
    """ 
    mongo_client = None 

    def initialize(self, mongo_client=None): 
     """ 
     Rest Client Initialize 
     Args: 
      mongo_client: The client used to access documents for this handler 

     Returns: 

     """ 
     BaseHandler.initialize(self) 
     self.mongo_client = mongo_client 

Answer 2

AsyncHTTPTestCase创建一个每次测试开始时新的IOLoop，并将其销毁t每次测试的结束。但是，您将在整个测试课程开始时创建MotorClient，并使用默认的全局IOLoop代替专门为每个测试创建的IOLoop。

我相信你只需要用setUp代替setUpClass。然后，您将在AsyncHTTPTestCase设置其IOLoop后创建您的MotorClient。为了清楚起见，请明确通过IOLoop：

client = MotorClient(io_loop=self.io_loop) 
self.db = client[self.config.MONGODB_NAME]