GAE:使用测试平台进行单元测试任务

Question

我正在使用testbed对我的谷歌应用引擎应用进行单元测试,而我的应用使用了一个任务队列.

当我在单元测试期间向任务队列提交任务时,似乎任务在队列中,但任务不会执行.

如何在单元测试期间执行任务？

Answer 1

使用Saxon的优秀答案,我能够使用testbed而不是gaetestbed做同样的事情.这就是我做的.

添加到我的setUp():

    self.taskqueue_stub = apiproxy_stub_map.apiproxy.GetStub('taskqueue')

然后,在我的测试中,我使用了以下内容:

    # Execute the task in the taskqueue
    tasks = self.taskqueue_stub.GetTasks("default")
    self.assertEqual(len(tasks), 1)
    task = tasks[0]
    params = base64.b64decode(task["body"])
    response = self.app.post(task["url"], params)

沿着这条线的某个地方,POST参数得到了base64编码,所以不得不撤消它以使其工作.

我比萨克森的答案更喜欢这个,因为我可以使用官方测试包,我可以在我自己的测试代码中完成所有操作.

编辑:我后来想用使用延迟库提交的任务做同样的事情,并且花了一些头脑来计算它,所以我在这里分享以减轻其他人的痛苦.

如果您的任务队列仅包含使用延迟提交的任务,那么这将运行所有任务以及由这些任务排队的任何任务:

def submit_deferred(taskq):
    tasks = taskq.GetTasks("default")
    taskq.FlushQueue("default")
    while tasks:
        for task in tasks:
            (func, args, opts) = pickle.loads(base64.b64decode(task["body"]))
            func(*args)
        tasks = taskq.GetTasks("default")
        taskq.FlushQueue("default")

Answer 2

实现此目的的另一个(更干净)选项是在测试平台中使用任务队列存根.要执行此操作,首先必须通过向setUp()方法添加以下内容来初始化任务队列存根:

self.testbed = init_testbed()
self.testbed.init_taskqueue_stub()

可以使用以下代码访问任务调度程序:

taskq = self.testbed.get_stub(testbed.TASKQUEUE_SERVICE_NAME)

使用队列存根的接口如下:

GetQueues() #returns a list of dictionaries with information about the available queues

#returns a list of dictionaries with information about the tasks in a given queue
GetTasks(queue_name)

DeleteTask(queue_name, task_name) #removes the task task_name from the given queue

FlushQueue(queue_name) #removes all the tasks from the queue

#returns tasks filtered by name & url pointed to by the task from the given queues
get_filtered_tasks(url, name, queue_names)

StartBackgroundExecution() #Executes the queued tasks

Shutdown() #Requests the task scheduler to shutdown.

此外,由于这使用App Engine SDK自己的设施 - 它与延迟库一起工作得很好.

Answer 3

dev app服务器是单线程的,因此当前台线程运行测试时,它无法在后台运行任务.

我在gaetestbed中的taskqueue.py中修改了TaskQueueTestCase以添加以下函数:

def execute_tasks(self, application):
    """
    Executes all currently queued tasks, and also removes them from the 
    queue.
    The tasks are execute against the provided web application.
    """

    # Set up the application for webtest to use (resetting _app in case a
    # different one has been used before). 
    self._app = None
    self.APPLICATION = application

    # Get all of the tasks, and then clear them.
    tasks = self.get_tasks()
    self.clear_task_queue()

    # Run each of the tasks, checking that they succeeded.
    for task in tasks:
        response = self.post(task['url'], task['params'])
        self.assertOK(response)

为此,我还必须将TaskQueueTestCase的基类从BaseTestCase更改为WebTestCase.

我的测试然后做这样的事情:

# Do something which enqueues a task.

# Check that a task was enqueued, then execute it.
self.assertTrue(len(self.get_tasks()), 1)
self.execute_tasks(some_module.application)

# Now test that the task did what was expected.

因此,这直接从前台单元测试执行任务.这与生产中的不完全相同(即,任务将在一段时间后在单独的请求中执行),但它对我来说效果很好.