1

为龙卷风应用程序运行单元测试时,我不断收到此错误:

tornado.ioloop.TimeoutError: Operation timed out after 5 seconds

这是测试代码:

class TestMongo(testing.AsyncTestCase):

def setUp(self):
    super().setUp()
@patch('stashboard.checkers.events')
@testing.gen_test()
def test_check(self, event_mock,): # tests for Ok connection status
    event_mock.STATUS_OK = events.STATUS_OK
    event_mock.STATUS_FAIL = events.STATUS_FAIL
    event_mock.save.return_value = Future()
    d = {'path': '/test'}
    test = MongoChecker(d, 1, None)
    yield test.check()
    event_mock.save.assert_called_with({'path': '/test'},
                                 {'status': events.STATUS_OK,
                             'address': '#address#'})

这是正在测试的代码:

class MongoChecker(Checker):
# pings Mongo servers noted in configuration to make sure
# they are still running

def __init__(self, event, frequency, params):
    super().__init__(event, frequency, params)
    self.clients = []
    for server in configuration["mongodb"]:
        host = server["host"]
        port = server["port"]
        address = 'mongodb://{}:{}'.format(host, port)
        client = motor.motor_tornado.MotorClient(
            address)
        # creates client to test connection
        client.address = address
        self.clients.append(client)

@gen.coroutine
def check(self):
    for client in self.clients:
        try:
            yield client.admin.command('ping')
            data= {'address': client.address, 'status': events.STATUS_OK}
        except ConnectionError:
            data = {'address': client.address, 'status': events.STATUS_FAIL}
        yield self.save(data)

这是为断言实际调用的方法:

@gen.coroutine
def save(self, data):
    yield events.save(self.event, data)

当我从测试中删除 yield 语句时yield self.save(data)工作正常。我需要模拟一个Future要从中返回的对象,self.save并从中获得实际结果。

4

1 回答 1

2

您已修补stashboard.checkers.events并设置Future为返回值,因此代码:

@gen.coroutine
def save(self, data):
    yield events.save(self.event, data)

变得

@gen.coroutine
def save(self, data):
    yield Future()

yield Future意味着 Tornado 将等到给定的未来将被解决,无论是 byset_result还是 by set_exception。在出于安全原因(和理智)的单元测试中,每个测试/套件都有时间限制。由于您尚未设置结果,因此测试将永远持续,幸好它已超时。

您可以使用tornado.gen.maybe_future来设置协程的结果。所以模拟设置可能如下所示:

@patch('stashboard.checkers.events')
@testing.gen_test()
def test_check(self, event_mock,): # tests for Ok connection status
    event_mock.STATUS_OK = events.STATUS_OK
    event_mock.STATUS_FAIL = events.STATUS_FAIL
    event_mock.save.return_value = tornado.gen.maybe_future('some dummy ret')
    # ...
于 2016-06-29T08:46:26.520 回答