Testing RQ with Django and fakeredis

2 Aug. 2023, Python, 2231 views,   Django, RQ и FakeRedis

I usually use RQ with Django because RQ is one of the most popular and straightforward solutions among task queues in the Python ecosystem. A few months ago I stumbled upon a situation where I needed to test a code that used Django Signals. The scenario was simple. When the signal is emitted (the Django model object has been deleted) receiver listening to that signal invokes an RQ job using delay. The problem is that the corresponding object is created using pytest fixture and is deleted when a test finishes. The first straightforward solution was to patch an RQ job, but if we have more receivers in the future we should not forget to patch them all (which affects readability and code clarity). I decided to apply another solution that replaces a connection class based on a condition. The condition is having FAKE_REDIS equals to True inside settings.py.

Take a look at the code:

from rq.decorators import job as rq_job_decorator

def async_job(queue_name: str, *args: t.Any, **kwargs: t.Any) -> t.Any:
    """
    The same as RQ's job decorator, but it automatically replaces the
    ``connection`` argument with a fake one if ``settings.FAKE_REDIS`` is set to ``True``.
    """

    class LazyAsyncJob:
        def __init__(self, f: t.Callable[..., t.Any]) -> None:
            self.f = f
            self.job: t.Optional[t.Callable[..., t.Any]] = None

        def setup_connection(self) -> t.Callable[..., t.Any]:
            if self.job:
                return self.job
            if settings.FAKE_REDIS:
                from fakeredis import FakeRedis

                queue = get_queue(queue_name, connection=FakeRedis())  # type: ignore
            else:
                queue = get_queue(queue_name)

            RQ = getattr(settings, 'RQ', {})
            default_result_ttl = RQ.get('DEFAULT_RESULT_TTL')
            if default_result_ttl is not None:
                kwargs.setdefault('result_ttl', default_result_ttl)

            return rq_job_decorator(queue, *args, **kwargs)(self.f)

        def delay(self, *args: t.Any, **kwargs: t.Any) -> t.Any:
            self.job = self.setup_connection()
            return self.job.delay(*args, **kwargs)  # type: ignore

        def __call__(self, *args: t.Any, **kwargs: t.Any) -> t.Any:
            self.job = self.setup_connection()
            return self.job(*args, **kwargs)

    return LazyAsyncJob

In order to use this code you have to decorate all your jobs using async_job decorator:

@async_job('default')
def process_image():
    pass

To apply FAKE_REDIS setting for all tests use the following fixture:

@pytest.fixture(autouse=True)
def fake_redis(settings: t.Any) -> None:
    settings.FAKE_REDIS = True