为您的代码摘录“GAE上下文”。在生产中为测试提供了真正的“GAE实现”提供了一个模拟自己,将提高DeadlineExceededError。测试不应该取决于任何超时,应该很快。
样品抽象(只是胶):
class AbstractGAETaskContext(object):
def task_spired(): pass # this will throw exception in mock impl
# here you define any method that you call into GAE, to be mocked
def defered(...): pass
如果你不喜欢抽象,你可以做猴子补丁仅适用于测试,你也需要定义task_expired功能,成为您的测试挂钩。 task_expired应该在您的任务执行功能期间被调用。
* 修订 *此第三解决方案:
首先,我想提一提的是,Nick's sample implementation没有那么大,映射器类有许多responsabilities(推迟,查询数据,批量更新);这使得测试很难做出,很多模拟需要被定义。所以我在一个单独的课程中提取了推迟的责任。 你只想测试推迟机制,实际发生了什么(更新,查询等)应该在其他测试中处理。
下面是deffering类,也是这个没有更多的依赖于GAE:
class DeferredCall(object):
def __init__(self, deferred):
self.deferred = deferred
def run(self, long_execution_call, context, *args, **kwargs):
''' long_execution_call should return a tuple that tell us how was terminate operation, with timeout and the context where was abandoned '''
next_context, timeouted = long_execution_call(context, *args, **kwargs)
if timeouted:
self.deferred(self.run, next_context, *args, **kwargs)
下面是测试模块:
class Test(unittest.TestCase):
def test_defer(self):
calls = []
def mock_deferrer(callback, *args, **kwargs):
calls.append((callback, args, kwargs))
def interrupted(self, context):
return "new_context", True
d = DeferredCall()
d.run(interrupted, "init_context")
self.assertEquals(1, len(calls), 'a deferred call should be')
def test_no_defer(self):
calls = []
def mock_deferrer(callback, *args, **kwargs):
calls.append((callback, args, kwargs))
def completed(self, context):
return None, False
d = DeferredCall()
d.run(completed, "init_context")
self.assertEquals(0, len(calls), 'no deferred call should be')
怎么会看尼克的Mapper实现:
class Mapper:
...
def _continue(self, start_key, batch_size):
... # here is same code, nothing was changed
except DeadlineExceededError:
# Write any unfinished updates to the datastore.
self._batch_write()
# Queue a new task to pick up where we left off.
##deferred.defer(self._continue, start_key, batch_size)
return start_key, True ## make compatible with DeferredCall
self.finish()
return None, False ## make it comaptible with DeferredCall
runner = _continue
注册长时间运行任务的代码;这只依赖于GAE延迟库。
import DeferredCall
import PersonMapper # this inherits the Mapper
from google.appengine.ext import deferred
mapper = PersonMapper()
DeferredCall(deferred).run(mapper.run)
最终,你需要的是当你想要的时候会抛出异常。难道你不能只写一个简单的方法来抛出它,并使用模拟库来替换它需要的地方吗? –
@尼克,我从来没有做过模拟,所以我有点迷失在这里。你是否暗示我在单元测试中嘲笑自己的代码,嘲笑GAE延期库或其他东西?任何额外的指针将不胜感激。 –
模拟您想要抛出该异常的任何方法 - 可能是SDK函数之一。 –