Django Q tasks are not executed with ./manage.py test

[epinna]

Hello,

My models have some django Q task starting on certain action such as overridden Model.save() . The async tasks work correctly during functional testing, but the tasks are submitted but not executed if I hit the same code with ./manage.py test.

I just see some message of my 13:07:53 [Q] INFO Enqueued 1, probably of my tasks being queued, but they're not executed.

Is this something expected?

[Eagllus]

Short answer is: Yes it is something expected.

By calling async you queue your task 13:07:53 [Q] INFO Enqueued 1.

There are a few ways you can processed.

1. Execute the queued task during your test run
   or
2. (what I have done in my projects) Is write tests for the tasks and test them.

[epinna]

Hey Eagllus,

Thanks for the clarification.

How to execute the queued tasks? Is something I should write at code-level within my tests or can I trigger that with some manage.py command?

Thanks

[epinna]

I looked at your testsuite, I think I'll write as the same as you do here:

https://github.com/Koed00/django-q/blob/master/django_q/tests/test_cluster.py#L111

Thanks

[epinna]

I wrote in my tests something like this, which seems working for a single task:

            Model.save() # It submits an async task

            broker.queue_size()
            task_queue = Queue()
            stop_event = Event()
            stop_event.set()
            # push the tasks
            for i in range(task_count):
                pusher(task_queue, stop_event, broker=broker)
            task_queue.put('STOP')

            # let a worker handle them
            result_queue = Queue()
            worker(task_queue, result_queue, Value('f', -1))
            result_queue.put('STOP')

The problem is that I'm submitting a Chain of tasks, and it seems handling only the first one :(

Could you suggest me how to handle chain of tasks manually in order to be executed within TestCases?

Thanks for your help

[Eagllus]

@epinna a very easy solution would be to change a setting during you tests.

In your tests.py settings files

Q_CLUSTER = {
    'sync': True
}

or at the beginning of your test (if you only have a single test that needs it)

from django.conf import settings
settings.Q_CLUSTER['sync'] = True

This will remove the need to copy any code

[epinna]

That's what I've done at the end :) thanks for your help.

[NickVeld]

In case someone hit this... If you see that async tasks are executed in some cases (functional testing like here or running docker compose up in my case or ...) and just [Q] INFO Enqueued 1 in some cases (unit testing like here or running docker compose run SERVICE_NAME SOME_CUSTOM_COMMAND or ...) , it means that your QCluster is not running. Most probably it is not running because the command starting it python manage.py qcluster intentionally or accidentally skipped in some environments.

[NickVeld]

P.S. About the same thing I wrote above: #203