使用 IOLoop 测试电机调用
Testing Motor calls with IOLoop
我在电机数据库调用的回调中进行了 运行 单元测试,我成功地捕获了 AssertionErrors 并在 运行 nosetests 时让它们浮出水面,但是 AssertionErrors 被错误地捕获了测试。回溯到不同的文件。
我的单元测试通常是这样的:
def test_create(self):
@self.callback
def create_callback(result, error):
self.assertIs(error, None)
self.assertIsNot(result, None)
question_db.create(QUESTION, create_callback)
self.wait()
我正在使用的 unittest.TestCase class 看起来像这样:
class MotorTest(unittest.TestCase):
bucket = Queue.Queue()
# Ensure IOLoop stops to prevent blocking tests
def callback(self, func):
def wrapper(*args, **kwargs):
try:
func(*args, **kwargs)
except Exception as e:
self.bucket.put(traceback.format_exc())
IOLoop.current().stop()
return wrapper
def wait(self):
IOLoop.current().start()
try:
raise AssertionError(self.bucket.get(block = False))
except Queue.Empty:
pass
我看到的错误:
======================================================================
FAIL: test_sync_user (app.tests.db.test_user_db.UserDBTest)
----------------------------------------------------------------------
Traceback (most recent call last):
File "/Users/----/Documents/app/app-Server/app/tests/db/test_user_db.py", line 39, in test_sync_user
self.wait()
File "/Users/----/Documents/app/app-Server/app/tests/testutils/mongo.py", line 25, in wait
raise AssertionError(self.bucket.get(block = False))
AssertionError: Traceback (most recent call last):
File "/Users/----/Documents/app/app-Server/app/tests/testutils/mongo.py", line 16, in wrapper
func(*args, **kwargs)
File "/Users/----/Documents/app/app-Server/app/tests/db/test_question_db.py", line 32, in update_callback
self.assertEqual(result["question"], "updated question?")
TypeError: 'NoneType' object has no attribute '__getitem__'
错误报告在 UsersDbTest 中,但显然在 test_questions_db.py(即 QuestionsDbTest)中
我在 nosetests 和异步测试方面遇到了一些问题,所以如果有人对此有任何建议,我将不胜感激。
如果没有 SSCCE,我无法完全理解您的代码,但我想说您在一般情况下采用了一种不明智的异步测试方法。
您面临的特定问题是您在离开测试函数之前没有等待测试完成(异步),因此当您在下一个测试中恢复循环时,IOLoop 中仍有待处理的工作。使用 Tornado 自己的 "testing" 模块——它提供了启动和停止循环的便捷方法,并在测试之间重新创建循环,因此您不会遇到像您所报告的那样的干扰。最后,它有非常方便的测试协程的方法。
例如:
import unittest
from tornado.testing import AsyncTestCase, gen_test
import motor
# AsyncTestCase creates a new loop for each test, avoiding interference
# between tests.
class Test(AsyncTestCase):
def callback(self, result, error):
# Translate from Motor callbacks' (result, error) convention to the
# single arg expected by "stop".
self.stop((result, error))
def test_with_a_callback(self):
client = motor.MotorClient()
collection = client.test.collection
collection.remove(callback=self.callback)
# AsyncTestCase starts the loop, runs until "remove" calls "stop".
self.wait()
collection.insert({'_id': 123}, callback=self.callback)
# Arguments passed to self.stop appear as return value of "self.wait".
_id, error = self.wait()
self.assertIsNone(error)
self.assertEqual(123, _id)
collection.count(callback=self.callback)
cnt, error = self.wait()
self.assertIsNone(error)
self.assertEqual(1, cnt)
@gen_test
def test_with_a_coroutine(self):
client = motor.MotorClient()
collection = client.test.collection
yield collection.remove()
_id = yield collection.insert({'_id': 123})
self.assertEqual(123, _id)
cnt = yield collection.count()
self.assertEqual(1, cnt)
if __name__ == '__main__':
unittest.main()
(在这个例子中,我为每个测试创建了一个新的 MotorClient,这在测试使用 Motor 的应用程序时是个好主意。您的实际应用程序 不得 创建一个新的 MotorClient每个操作。为了获得良好的性能,您必须在应用程序开始时创建 one MotorClient,并在整个过程的生命周期中使用同一个客户端。)
看看测试模块,尤其是 gen_test 装饰器:
http://tornado.readthedocs.org/en/latest/testing.html
这些测试便利处理了与单元测试 Tornado 应用程序相关的许多细节。
我做了一个演讲并写了一篇关于在 Tornado 中进行测试的文章,这里有更多信息:
我在电机数据库调用的回调中进行了 运行 单元测试,我成功地捕获了 AssertionErrors 并在 运行 nosetests 时让它们浮出水面,但是 AssertionErrors 被错误地捕获了测试。回溯到不同的文件。
我的单元测试通常是这样的:
def test_create(self):
@self.callback
def create_callback(result, error):
self.assertIs(error, None)
self.assertIsNot(result, None)
question_db.create(QUESTION, create_callback)
self.wait()
我正在使用的 unittest.TestCase class 看起来像这样:
class MotorTest(unittest.TestCase):
bucket = Queue.Queue()
# Ensure IOLoop stops to prevent blocking tests
def callback(self, func):
def wrapper(*args, **kwargs):
try:
func(*args, **kwargs)
except Exception as e:
self.bucket.put(traceback.format_exc())
IOLoop.current().stop()
return wrapper
def wait(self):
IOLoop.current().start()
try:
raise AssertionError(self.bucket.get(block = False))
except Queue.Empty:
pass
我看到的错误:
======================================================================
FAIL: test_sync_user (app.tests.db.test_user_db.UserDBTest)
----------------------------------------------------------------------
Traceback (most recent call last):
File "/Users/----/Documents/app/app-Server/app/tests/db/test_user_db.py", line 39, in test_sync_user
self.wait()
File "/Users/----/Documents/app/app-Server/app/tests/testutils/mongo.py", line 25, in wait
raise AssertionError(self.bucket.get(block = False))
AssertionError: Traceback (most recent call last):
File "/Users/----/Documents/app/app-Server/app/tests/testutils/mongo.py", line 16, in wrapper
func(*args, **kwargs)
File "/Users/----/Documents/app/app-Server/app/tests/db/test_question_db.py", line 32, in update_callback
self.assertEqual(result["question"], "updated question?")
TypeError: 'NoneType' object has no attribute '__getitem__'
错误报告在 UsersDbTest 中,但显然在 test_questions_db.py(即 QuestionsDbTest)中
我在 nosetests 和异步测试方面遇到了一些问题,所以如果有人对此有任何建议,我将不胜感激。
如果没有 SSCCE,我无法完全理解您的代码,但我想说您在一般情况下采用了一种不明智的异步测试方法。
您面临的特定问题是您在离开测试函数之前没有等待测试完成(异步),因此当您在下一个测试中恢复循环时,IOLoop 中仍有待处理的工作。使用 Tornado 自己的 "testing" 模块——它提供了启动和停止循环的便捷方法,并在测试之间重新创建循环,因此您不会遇到像您所报告的那样的干扰。最后,它有非常方便的测试协程的方法。
例如:
import unittest
from tornado.testing import AsyncTestCase, gen_test
import motor
# AsyncTestCase creates a new loop for each test, avoiding interference
# between tests.
class Test(AsyncTestCase):
def callback(self, result, error):
# Translate from Motor callbacks' (result, error) convention to the
# single arg expected by "stop".
self.stop((result, error))
def test_with_a_callback(self):
client = motor.MotorClient()
collection = client.test.collection
collection.remove(callback=self.callback)
# AsyncTestCase starts the loop, runs until "remove" calls "stop".
self.wait()
collection.insert({'_id': 123}, callback=self.callback)
# Arguments passed to self.stop appear as return value of "self.wait".
_id, error = self.wait()
self.assertIsNone(error)
self.assertEqual(123, _id)
collection.count(callback=self.callback)
cnt, error = self.wait()
self.assertIsNone(error)
self.assertEqual(1, cnt)
@gen_test
def test_with_a_coroutine(self):
client = motor.MotorClient()
collection = client.test.collection
yield collection.remove()
_id = yield collection.insert({'_id': 123})
self.assertEqual(123, _id)
cnt = yield collection.count()
self.assertEqual(1, cnt)
if __name__ == '__main__':
unittest.main()
(在这个例子中,我为每个测试创建了一个新的 MotorClient,这在测试使用 Motor 的应用程序时是个好主意。您的实际应用程序 不得 创建一个新的 MotorClient每个操作。为了获得良好的性能,您必须在应用程序开始时创建 one MotorClient,并在整个过程的生命周期中使用同一个客户端。)
看看测试模块,尤其是 gen_test 装饰器:
http://tornado.readthedocs.org/en/latest/testing.html
这些测试便利处理了与单元测试 Tornado 应用程序相关的许多细节。
我做了一个演讲并写了一篇关于在 Tornado 中进行测试的文章,这里有更多信息: