如何测试模型在传递到任务队列之前是否已保存?
How to test if a model was saved before passed on to task queue?
在以下方法中,正确的行为是保存 news
模型 在 将其 id 传递到异步任务队列以进行进一步处理之前。
否则任务队列有旧版本。
如您所见,我犯了一个错误,在将其 ID 发送到任务队列后保存模型。
def save_scraped_body_into_model(url_string):
news = ndb.Key(urlsafe=url_string).get()
...
news.body = result
news.stage = 1
taskqueue.Task(url='/api/v1.0/worker/bbc-stage-2', headers=header,
payload=json.dumps({'news_url_string': news.key.urlsafe()})).add(queue_name='newstasks')
news.put()
我该如何测试呢?
只要模型被保存,我下面的测试就会通过。但那是错误的,顺序很重要,而且这个测试没有捕捉到它!有没有办法用 mock 实现这个?
def test_news_instance_saved_before_next_stage(self, get_head):
BBCSpider.save_scraped_body_into_model(self.news.key.urlsafe())
context = ndb.get_context()
context.clear_cache()
news = ndb.Key(urlsafe=self.news.key.urlsafe()).get()
self.assertEqual(news.stage, 1)
我正在使用 unittest.mock
framework 中的 patch
进行此类测试。
当我需要做这样的测试时,我做的第一件事就是寻找一个可以放钩子的点。然后我修补钩子并使用 side_effect
回调来进行测试。
在你的情况下,钩子将是 BBCSpider.taskqueue.Task
并且它的 side_effect
是这样的:
lambda *args,**kwargs: self.assertEqual(1, ndb.Key(urlsafe=self.news.key.urlsafe()).get().stage)
所以你的测试变成:
@patch('BBCSpider.taskqueue.Task', autospec=True)
def test_news_instance_saved_before_next_stage(self, get_head, mock_task):
def check_if_model_saved(*args,**kwargs):
news = ndb.Key(urlsafe=self.news.key.urlsafe()).get()
self.assertEqual(news.stage,1)
mock_task.side_effect = check_if_model_saved
BBCSpider.save_scraped_body_into_model(self.news.key.urlsafe())
self.assertTrue(mock_task.called)
注意 autospec=True
不是强制性的,但我喜欢在每次打补丁时使用它以避免愚蠢的错误。
如果代码中有错误,我深表歉意(我无法在不付出大量努力的情况下对其进行测试),但我希望思路清晰。
在以下方法中,正确的行为是保存 news
模型 在 将其 id 传递到异步任务队列以进行进一步处理之前。
否则任务队列有旧版本。
如您所见,我犯了一个错误,在将其 ID 发送到任务队列后保存模型。
def save_scraped_body_into_model(url_string):
news = ndb.Key(urlsafe=url_string).get()
...
news.body = result
news.stage = 1
taskqueue.Task(url='/api/v1.0/worker/bbc-stage-2', headers=header,
payload=json.dumps({'news_url_string': news.key.urlsafe()})).add(queue_name='newstasks')
news.put()
我该如何测试呢? 只要模型被保存,我下面的测试就会通过。但那是错误的,顺序很重要,而且这个测试没有捕捉到它!有没有办法用 mock 实现这个?
def test_news_instance_saved_before_next_stage(self, get_head):
BBCSpider.save_scraped_body_into_model(self.news.key.urlsafe())
context = ndb.get_context()
context.clear_cache()
news = ndb.Key(urlsafe=self.news.key.urlsafe()).get()
self.assertEqual(news.stage, 1)
我正在使用 unittest.mock
framework 中的 patch
进行此类测试。
当我需要做这样的测试时,我做的第一件事就是寻找一个可以放钩子的点。然后我修补钩子并使用 side_effect
回调来进行测试。
在你的情况下,钩子将是 BBCSpider.taskqueue.Task
并且它的 side_effect
是这样的:
lambda *args,**kwargs: self.assertEqual(1, ndb.Key(urlsafe=self.news.key.urlsafe()).get().stage)
所以你的测试变成:
@patch('BBCSpider.taskqueue.Task', autospec=True)
def test_news_instance_saved_before_next_stage(self, get_head, mock_task):
def check_if_model_saved(*args,**kwargs):
news = ndb.Key(urlsafe=self.news.key.urlsafe()).get()
self.assertEqual(news.stage,1)
mock_task.side_effect = check_if_model_saved
BBCSpider.save_scraped_body_into_model(self.news.key.urlsafe())
self.assertTrue(mock_task.called)
注意 autospec=True
不是强制性的,但我喜欢在每次打补丁时使用它以避免愚蠢的错误。
如果代码中有错误,我深表歉意(我无法在不付出大量努力的情况下对其进行测试),但我希望思路清晰。