GAE：单元测试DeadlineExceededError

Question

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

由于这是在任务队列的上下文中，构建一个花费超过10分钟运行的测试将是很痛苦的。为了测试目的，是否有办法将taskqueue时间限制临时设置为几秒钟？或者也许有其他方法来优雅地测试except DeadlineExceededError块中代码的执行情况？

Answer 1

为您的代码摘录“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)