Python 任务队列 RQ 中的测试 ( Testing )

当下最流行的测试模式应该就是单元测试了，所以，对于 RQ，我们只讲讲如何单元测试

对工作进程进行单元测试

可能，你希望将 RQ 任务 ( task ) 也纳入单元测试，但，很抱歉，许多框架 ( 例如 Django ) 使用的一般都是内存数据库。这些数据库与 RQ 的默认的 fork() 行为可能不能很好的兼容

当然了，也不是没有解决办法，我们可以使用 SimpleWorker 类来避免 fork() 操作

from redis import Redis
from rq import SimpleWorker, Queue

queue = Queue(connection=Redis())
queue.enqueue(my_long_running_job)
worker = SimpleWorker([queue], connection=queue.connection)
worker.work(burst=True)  # Runs enqueued job
# Check for result...

在单元测试中执行作业任务

除了使用 SimpleWorker 外，另一个可用于测试目的的解决方案是使用 is_async = False 队列参数

该参数指示它在同一个线程中立即执行作业，而不是将其分派给 worker，也就是说不再需要工作进程

另外，我们可以使用 fakeredis 来模拟 Redis 实例，也就是说我们不必单独运行 Redis 服务器

伪 Redis 服务器的实例可以直接作为连接参数传递给队列

rom fakeredis import FakeStrictRedis
from rq import Queue

queue = Queue(is_async=False, connection=FakeStrictRedis())
job = queue.enqueue(my_long_running_job)
assert job.is_finished