Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

non working example of using celery_worker pytest fixture (celery.exception.NotRegistered) #7750

Closed
4 tasks done
hq9000 opened this issue Sep 9, 2022 · 3 comments · Fixed by #7857
Closed
4 tasks done

Comments

@hq9000
Copy link

hq9000 commented Sep 9, 2022

Checklist

  • I have checked the issues list
    for similar or identical bug reports.
  • I have checked the pull requests list
    for existing proposed fixes.
  • I have checked the commit log
    to find out if the bug was already fixed in the master branch.
  • I have included all related issues and possible duplicate issues in this issue
    (If there are none, check this box anyway).

Related Issues and Possible Duplicates

Related Issues

  • None

Possible Duplicates

  • None

Description

the part of the documentation describing celery_app and celery_worker fixtures has a test example that doesn't pass:

https://github.com/celery/celery/blame/master/docs/userguide/testing.rst#L152

image

if we use this test as it is, it will fail because the task does not get registered in the worker:

image

celery.exceptions.NotRegistered: 'test_create_task.mul'

Suggestions

the problem is that executed this way, the tasks are not registered in the worker. The way I found working is to override the celery_app fixture:

from celery import Celery
from celery.contrib.testing.worker import TestWorkController
from celery.result import AsyncResult
import pytest

@pytest.fixture
def celery_app(celery_app: Celery):

    @celery_app.task(name='xyz')
    def mul(x, y):
        """register a task named 'test_create_task.mul'"""
        return x * y

    return celery_app



def test_create_task(celery_app: Celery, celery_worker: TestWorkController):    
    result: AsyncResult = celery_app.send_task('xyz', kwargs = {"x": 2, "y": 2})
    result.wait()
    assert result.get() == 4

image

@open-collective-bot
Copy link

Hey @hq9000 👋,
Thank you for opening an issue. We will get back to you as soon as we can.
Also, check out our Open Collective and consider backing us - every little helps!

We also offer priority support for our sponsors.
If you require immediate assistance please consider sponsoring us.

@hq9000 hq9000 changed the title non working example of using celery_worker pytest fixture non working example of using celery_worker pytest fixture (celery.exception.NotRegistered) Sep 9, 2022
@auvipy
Copy link
Member

auvipy commented Sep 9, 2022

would you mind sending a patch to improve this?

@hq9000
Copy link
Author

hq9000 commented Sep 9, 2022

would you mind sending a patch to improve this?

will do in a couple of days

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Projects
None yet
Development

Successfully merging a pull request may close this issue.

2 participants