GAE：单元测试 DeadlineExceededError

Question

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

由于这是在任务队列的上下文中，构建一个运行时间超过 10 分钟的测试会很痛苦。有没有办法临时将任务队列时间限制设置为几秒钟以进行测试？或者也许是其他方式来优雅地测试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 不是很好，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)