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_result
或 set_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')
# ...
当 运行 对龙卷风应用程序进行单元测试时,我不断收到此错误:
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_result
或 set_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')
# ...