在功能測試中測試任務隊列和管線API

我可以使用此代碼很好地運行任務作爲我們的功能測試的一部分。在功能測試中測試任務隊列和管線API

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‘[一些ID]’在嘗試1的消息，‘ ’該不會是準備到：以後的某個時候]「

相關來源在這裏。

+++ 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(),

我以爲猴子在代碼修補_RETRY_WIGGLE_TIMEDELTA是帶來了在我的測試包的AppEngine上的API，但似乎並沒有是不夠的。

等待時間到了，我的代碼會讓我的測試套件運行速度非常慢，我不知所措。

任何想法？

來源

2012-04-06 Tom Willis

那麼我誤診了這個問題。一項任務失敗了，那個邏輯就是重新嘗試的邏輯。這在管道API中被強制執行有點奇怪，但我確信這有一個很好的理由。因此，在處理任務循環中的某處，我需要檢查重試計數並以某種方式將這些數據從隊列中取出，然後提取從第一次運行中捕獲的所有錯誤信息，以便可以認爲測試失敗。

來源

2012-04-06 16:00:23

在功能測試中測試任務隊列和管線API

回答

相關問題