在功能测试期间测试任务队列和管道api

Question

我可以运行任务作为我们的功能测试的一部分只是罚款使用此代码。你知道吗

def run_tasks():
    taskq = apiproxy_stub_map.apiproxy.GetStub('taskqueue')
    tasks = taskq.GetTasks("default")
    taskq.FlushQueue("default")
    try:
        while tasks:
            for task in tasks:
                url, body, headers, method = task["url"], \
                                             base64.b64decode(task["body"]), \
                                             task["headers"], \
                                             task["method"]
                try:
                    res = Request.blank(url,
                                        body=body,
                                        headers=headers,
                                        method=method).get_response(wsgi_app)
                    if res.status_int != 200:
                        log.error("%s\n%s" % (url, str(res)))
                except Exception:
                    log.error(url, exc_info=True)

            tasks = taskq.GetTasks("default")
            taskq.FlushQueue("default")
    finally:
        pass

然而，管道任务在这里爆炸了

“尝试对某个任务执行评估，'ID] “要到[以后某个时候]才能准备好”

相关消息来源在这里。你知道吗

+++ b/src/pipeline/pipeline/pipeline.py Wed Apr 04 20:01:12 2012 -0400
@@ -1946,6 +1952,7 @@
     if pipeline_record.next_retry_time is not None:
       retry_time = pipeline_record.next_retry_time - _RETRY_WIGGLE_TIMEDELTA
       if self._gettime() <= retry_time:
+        # error under unit tests
         detail_message = (
             'Received evaluation task for pipeline ID "%s" on attempt %d, '
             'which will not be ready until: %s' % (pipeline_key.name(),

我想到了在代码中使用monkey patching\u RETRY\u WIGGLE\u TIMEDELTA 在我的testsuite中出现appengineapi，但似乎不是这样 够了。你知道吗

在我的代码中，没有等到时间，这将使我的 测试套件运行得太慢了，我不知所措。你知道吗

有什么想法吗？你知道吗

Answer 1

我诊断错了。一个任务失败了，这个逻辑就是重新尝试的逻辑。在管道api中强制执行这一点有点奇怪，但我相信这是有充分理由的。因此，在我处理任务的循环中的某个地方，我需要检查重试次数，并以某种方式将这些重试次数从队列中取出，然后引发从第一次运行中捕获的所有错误信息，以便可以认为测试失败。你知道吗