insert fake run_function

Author: bisgaard-itis

services/api-server/tests/unit/celery/conftest.py

@@ -47,12 +47,19 @@ def _(celery_app: Celery) -> None: ...
     return _
 
 
+@pytest.fixture
+def add_worker_tasks() -> bool:
+    "override to not add default worker tasks"
+    return True
+
+
 @pytest.fixture
 async def with_storage_celery_worker(
     app_environment: EnvVarsDict,
     celery_app: Celery,
     monkeypatch: pytest.MonkeyPatch,
     register_celery_tasks: Callable[[Celery], None],
+    add_worker_tasks: bool,
 ) -> AsyncIterator[TestWorkController]:
     # Signals must be explicitily connected
     monkeypatch.setenv("API_SERVER_WORKER_MODE", "true")
@@ -69,7 +76,8 @@ def _on_worker_init_wrapper(sender: WorkController, **_kwargs):
     worker_init.connect(_on_worker_init_wrapper)
     worker_shutdown.connect(on_worker_shutdown)
 
-    setup_worker_tasks(celery_app)
+    if add_worker_tasks:
+        setup_worker_tasks(celery_app)
     register_celery_tasks(celery_app)
 
     with start_worker(

services/api-server/tests/unit/celery/test_functions.py

@@ -1,5 +1,68 @@
+from collections.abc import Callable
+
+import pytest
+from celery import Celery, Task
 from celery.contrib.testing.worker import TestWorkController
+from celery_library.task import register_task
+from faker import Faker
+from models_library.functions import (
+    FunctionClass,
+    FunctionID,
+    FunctionInputs,
+    FunctionJobID,
+    RegisteredFunction,
+    RegisteredProjectFunctionJob,
+)
+from models_library.projects import ProjectID
+from servicelib.celery.models import TaskID
+from simcore_service_api_server.api.dependencies.authentication import Identity
+from simcore_service_api_server.celery._worker_tasks._functions_tasks import (
+    run_function as run_function_task,
+)
+from simcore_service_api_server.models.api_resources import JobLinks
+from simcore_service_api_server.models.schemas.jobs import (
+    JobPricingSpecification,
+    NodeID,
+)
+
+_faker = Faker()
+
+
+def _register_fake_run_function_task() -> Callable[[Celery], None]:
+
+    async def run_function(
+        task: Task,
+        task_id: TaskID,
+        *,
+        user_identity: Identity,
+        function: RegisteredFunction,
+        function_inputs: FunctionInputs,
+        pricing_spec: JobPricingSpecification | None,
+        job_links: JobLinks,
+        x_simcore_parent_project_uuid: NodeID | None,
+        x_simcore_parent_node_id: NodeID | None,
+    ):
+        return RegisteredProjectFunctionJob(
+            title=_faker.sentence(),
+            description=_faker.paragraph(),
+            function_uid=FunctionID(_faker.uuid4()),
+            inputs=function_inputs,
+            outputs=None,
+            function_class=FunctionClass.PROJECT,
+            uid=FunctionJobID(_faker.uuid4()),
+            created_at=_faker.date_time(),
+            project_job_id=ProjectID(_faker.uuid4()),
+        )
+
+    assert run_function_task.__name__ == run_function.__name__
+
+    def _(celery_app: Celery) -> None:
+        register_task(celery_app, run_function)
+
+    return _
 
 
+@pytest.mark.parametrize("register_celery_tasks", [_register_fake_run_function_task])
+@pytest.mark.parametrize("add_worker_tasks", [False])
 async def test_with_fake_run_function(with_storage_celery_worker: TestWorkController):
     pass