在Django中测试django-rq(python-rq)的最佳实践

Question

我将在我的项目中开始使用django-rq.

Django与基于Redis的Python排队库RQ集成.

测试使用RQ的django应用程序的最佳做法是什么？

例如,如果我想将我的应用程序测试为黑盒子,那么在用户执行某些操作后,我想执行当前队列中的所有作业,然后检查我的数据库中的所有结果.我怎么能在我的django测试中做到这一点？

Answer 1

我刚刚发现django-rq,它允许你在一个测试环境中启动一个worker来执行队列中的任何任务然后退出.

from django.test impor TestCase
from django_rq import get_worker

class MyTest(TestCase):
    def test_something_that_creates_jobs(self):
        ...                      # Stuff that init jobs.
        get_worker().work(burst=True)  # Processes all jobs then stop.
        ...                      # Asserts that the job stuff is done.

Answer 2

上面的答案都没有真正解决如何在没有安装 redis 和使用 django 设置的情况下进行测试。我发现在测试中包含以下代码不会影响项目本身，但会提供所需的一切。

该代码使用 fakeredis 假装 Redis 服务可用，在 RQ Django 读取设置之前设置连接。

默认情况下，fakeredis 连接不共享状态（服务器），因此连接必须相同。因此，复用它是一个单例对象。

from fakeredis import FakeStrictRedis, FakeRedis

class FakeRedisConn:
    """Singleton FakeRedis connection."""

    def __init__(self):
        self.conn = None

    def __call__(self, _, strict):
        if not self.conn:
            self.conn = FakeStrictRedis() if strict else FakeRedis()
        return self.conn


django_rq.queues.get_redis_connection = FakeRedisConn()

def test_case():
   ...

FakeRedis 可以选择直接使用以下方式支持它FakeRedisConnSingleton：

from fakeredis import FakeRedisConnSingleton

django_rq.queues.get_redis_connection = FakeRedisConnSingleton()

Answer 3

当队列中仍有作业时，您需要暂停测试。为此，您可以检查Queue.is_empty()，如果队列中仍有作业，则暂停执行：

import time
from django.utils.unittest import TestCase
import django_rq

class TestQueue(TestCase):

def test_something(self):
    # simulate some User actions which will queue up some tasks

    # Wait for the queued tasks to run
    queue = django_rq.get_queue('default')
    while not queue.is_empty():
        time.sleep(5) # adjust this depending on how long your tasks take to execute

    # queued tasks are done, check state of the DB
    self.assert(.....)