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

yufei       4 年, 2 月 前       877

当下最流行的测试模式应该就是单元测试了,所以,对于 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
目前尚无回复
简单教程 = 简单教程,简单编程
简单教程 是一个关于技术和学习的地方
现在注册
已注册用户请 登入
关于   |   FAQ   |   我们的愿景   |   广告投放   |  博客

  简单教程,简单编程 - IT 入门首选站

Copyright © 2013-2022 简单教程 twle.cn All Rights Reserved.