使用龙卷风测试AsyncHTTPTestCase连接超时
我有单元测试龙卷风应用程序的问题,请帮助我。错误堆栈跟踪:使用龙卷风测试AsyncHTTPTestCase连接超时
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)
我注意到了一些事情。
- 我看到的主要问题是你在你的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
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]