Ability to register tasks individually, not via modules

[shtlrs]

Checklist

• I have checked the issues list
  for similar or identical feature requests.
• I have checked the pull requests list
  for existing proposed implementations of this feature.
• I have checked the commit log
  to find out if the same feature was already implemented in the
  main 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

Brief Summary

I am currently trying to build an abstraction over celery that would allow registering and running tasks.

However, my tests do not work because the worker doesn't find the tasks, and even using the default_worker_tasks doesn't work because the decorator used to define the tasks isn't the celery.Celery(...).task one, but a custom decorator that returns a custom Task class which completely abstracts celery away.

It would be very helpful to have a way to register tasks in the worker individually, by passing an instance of the celery.Task class to some fixture

Design

Have a fixture called something like register_task that would look like this

@pytest.fixture
def register_task():
    def _register(task: celery.Task):
         ...  # do something to register
     
     return _register

# in the test

@pytest.fixture
def register_test_tasks(default_worker_app, register_task):
     task = default_worker_app.task(some_callable, name="task_name"))
     register_task(task)
 

Proposed Behavior

The behaviour is simply being able to register tasks individually in the worker, by passing an instance of celery.Task the the fixture defined above.

Proposed UI/UX

Diagrams

N/A

Alternatives

None

[Nusnus]

Adding single tasks is not possible and not planned for any future release at the moment.
However, you can mount an entire project into the worker to get everything running smoothly.

Check Celery’s smoke tests for a real-world example, where we create a celery worker that contains the entire celery repo at the local commit hash, to test it using pytest-celery.

https://github.com/celery/celery/blob/main/t/smoke/workers/dev.py#L57

default_worker_container = container(
    image="{celery_dev_worker_image.id}",
    ports=fxtr("default_worker_ports"),
    environment=fxtr("default_worker_env"),
    network="{default_pytest_celery_network.name}",
    volumes={
        # Volume: Worker /app
        "{default_worker_volume.name}": defaults.DEFAULT_WORKER_VOLUME,
        # Mount: Celery source
        os.path.abspath(os.getcwd()): {
            "bind": "/celery",
            "mode": "rw",
        },
    },
    wrapper_class=SmokeWorkerContainer,
    timeout=defaults.DEFAULT_WORKER_CONTAINER_TIMEOUT,
    command=fxtr("default_worker_command"),
)

Note that you still need to add the tasks: https://github.com/celery/celery/blob/main/t/smoke/conftest.py#L52

@pytest.fixture
def default_worker_tasks(default_worker_tasks: set) -> set:
    """Use all of the integration and smoke suites tasks in the smoke tests workers."""
    from t.integration import tasks as integration_tests_tasks
    from t.smoke import tasks as smoke_tests_tasks

    default_worker_tasks.add(integration_tests_tasks)
    default_worker_tasks.add(smoke_tests_tasks)
    return default_worker_tasks