Whoosy's Blog

藏巧于拙 用晦而明 寓清于浊 以屈为伸

0%

celery(二):如何对celery进行单元测试

编码不易,转载请注意出处!

前言

在接口开发过程中,对于简单功能的接口我们可能并不会对它编写单元测试,因为我们很容易就可以调用它判断结果是否是我们预期的。
但是对于执行接口逻辑相同的celery而言,我们想测试它并没有那么容易,因为celery任务在设计上是异步的,我们要遵循 “development driven development” 要困难的多。
测试驱动开发(TDD)也只是保证脑内的想法转成代码的时候逻辑自洽而已,并不能保证代码的真实质量, 但是涉及到celery任务时,我们想要保持高效的开发此过程是必不可少的,所以希望您保持理智😺,平静的测试,在这种情况下将您的代码发布到生产环境中。

单元测试

一个简单的celery任务

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# -*- coding: utf-8 -*-

import requests
import os

from datetime import datetime
from worker import app

@app.task(bind=True, name='fetch_data')
def fetch_data(self, url):
response = requests.get(url)
path = './data'
if response.ok:
if not os.path.exists(path):
os.makedirs(path)
slug = datetime.utcnow().strftime('%Y%m%dT%H%M%S%f')
with open(os.path.join(path, slug), 'w') as f:
f.write(response.text)
else:
raise ValueError('Unexpected response')

此celery任务发送一个向url发送GET请求,并将响应正文保存到文件中。我们有几种策略可以测试此Celery任务。

Strategy 1: 同步调度

一些 作者 建议异步调用 Celery 任务,然后让代码等待,直到任务执行结束并断言是否执行成功。

1
2
3
4
5
def test_fetch_data(self):
task = fetch_data.s(url='...').delay()
result = task.get()
self.assertEqual(task.status, 'SUCCESS')
...
优点
  • 很好的模拟了调度celery任务的整个流程
  • 测试用例非常接近真实环境
缺点
  • 需要消息中间件
  • 需要启用celery worker
  • 需要同步等待执行结果

Strategy 2:调用任务函数(官方推荐)

celery官方文档 建议celery任务应该像其他Python函数一样进行测试。

1
2
3
4
def test_fetch_data(self):
fetch_data(url='...')
self.assertEqual(...)
...
优点
  • Very Simple (就像我英语及烂还会使用这俩个单词一样)
  • 不依赖消息中间件
  • 不需要开启任务celery worker
  • 隔离性强
缺点
  • 并没有测试celery调度流程,仅仅测试了相关函数而已
  • 测试案例与真实使用场景并不相同

Strategy 3:异步调用任务

此策略结合了前面两个方面的优势。我们可以与实际环境中几乎相同的方式调度celery任务,异步发送任务(无需等待)并且会主动通知执行结果(在当前进程或线程中)。

1
2
3
4
def test_fetch_data(self):
task = fetch_data.s(url='...').apply()
self.assertEqual(task.result, 'SUCCESS')
...
优点
  • Very Simple
  • 不依赖消息中间件(在内存中保存任务及结果)
  • 不需要开启任务celery worker
  • 隔离性强
  • 很好的测试了整个任务的调度流程
  • 测试用例非常接近真实环境
缺点
  • Not sure there are any(不确定🤦‍)