python之GAE : Unit testing DeadlineExceededError

我一直在使用 testbed、webtest 和 nose 来测试我的 Python GAE 应用程序，这是一个很棒的设置。我现在正在实现类似于 Nick's great example of using the deferred library 的东西，但我想不出一个好的方法来测试由 DeadlineExceededError 触发的代码部分。

由于这是在任务队列的上下文中，构建一个运行时间超过 10 分钟的测试会很痛苦。有没有办法临时把taskqueue的时限设置成几秒来测试一下？或者也许可以通过其他方式优雅地测试 except DeadlineExceededError block 中代码的执行情况？

请您参考如下方法:

为您的代码抽象“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 函数作为你测试的钩子(Hook)。 task_expired 应该在你的任务实现函数中被调用。

*已更新*这是第三个解决方案:

首先我想提一下 Nick's sample implementation不是很好，Mapper 类有很多职责(延迟、查询数据、批量更新)；这使得测试很难进行，需要定义很多模拟。所以我将延迟责任提取到一个单独的类中。 您只想测试延迟机制，实际发生的事情(更新、查询等)应该在其他测试中处理。

这是递延类，也不再依赖于 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') 

Nick 的 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)