Tornado Unittest 模拟 yield 语句

Tornado Unittest mock yield statements

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

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 self.save(data) 中删除 yield 语句时,测试工作正常。我需要从 self.save 模拟一个 Future 对象到 return,并从中得到实际结果。

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

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

成为

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

yield Future 意味着 Tornado 将等到给定的未来被解决,由 set_resultset_exception 解决。出于安全原因(和理智),在单元测试中,每个 test/suite 都有时间限制。由于您还没有设置结果,测试将永远持续,幸运的是它超时了。

您可以使用 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')
    # ...