further test fixes

Author: bisgaard-itis

services/api-server/src/simcore_service_api_server/_service_function_jobs_task_client.py

@@ -1,7 +1,9 @@
+import contextlib
 from dataclasses import dataclass
 from typing import Final
 
 from common_library.exclude import as_dict_exclude_none
+from models_library.api_schemas_rpc_async_jobs.async_jobs import AsyncJobFilter
 from models_library.functions import (
     FunctionClass,
     FunctionID,
@@ -22,22 +24,27 @@
     UnsupportedFunctionFunctionJobClassCombinationError,
 )
 from models_library.products import ProductName
+from models_library.projects import ProjectID
+from models_library.projects_nodes_io import NodeID
 from models_library.projects_state import RunningState
 from models_library.rest_pagination import PageMetaInfoLimitOffset, PageOffsetInt
 from models_library.rpc_pagination import PageLimitInt
 from models_library.users import UserID
-from servicelib.celery.models import TaskUUID
+from servicelib.celery.models import TaskFilter, TaskMetadata, TasksQueue, TaskUUID
 from servicelib.celery.task_manager import TaskManager
 from sqlalchemy.ext.asyncio import AsyncEngine
 
 from ._service_function_jobs import FunctionJobService
 from ._service_functions import FunctionService
 from ._service_jobs import JobService
+from .api.dependencies.authentication import Identity
+from .api.dependencies.celery import ASYNC_JOB_CLIENT_NAME
 from .api.routes.tasks import _get_task_filter
 from .exceptions.function_errors import (
     FunctionJobCacheNotFoundError,
 )
-from .models.schemas.jobs import JobInputs
+from .models.api_resources import JobLinks
+from .models.schemas.jobs import JobInputs, JobPricingSpecification
 from .services_http.webserver import AuthSession
 from .services_rpc.storage import StorageService
 from .services_rpc.wb_api_server import WbApiRpcClient
@@ -319,3 +326,65 @@ async def function_job_outputs(
             outputs=new_outputs,
             check_write_permissions=False,
         )
+
+    async def create_function_job_creation_task(
+        self,
+        *,
+        function: RegisteredFunction,
+        function_inputs: FunctionInputs,
+        user_identity: Identity,
+        pricing_spec: JobPricingSpecification | None,
+        job_links: JobLinks,
+        parent_project_uuid: ProjectID | None = None,
+        parent_node_id: NodeID | None = None,
+    ) -> RegisteredFunctionJob:
+
+        job_inputs = await self._function_job_service.create_function_job_inputs(
+            function=function, function_inputs=function_inputs
+        )
+
+        # check if results are cached
+        with contextlib.suppress(FunctionJobCacheNotFoundError):
+            return await self.get_cached_function_job(
+                function=function,
+                job_inputs=job_inputs,
+            )
+
+        pre_registered_function_job_data = (
+            await self._function_job_service.pre_register_function_job(
+                function=function,
+                job_inputs=job_inputs,
+            )
+        )
+
+        # run function in celery task
+        job_filter = AsyncJobFilter(
+            user_id=user_identity.user_id,
+            product_name=user_identity.product_name,
+            client_name=ASYNC_JOB_CLIENT_NAME,
+        )
+        task_filter = TaskFilter.model_validate(job_filter.model_dump())
+
+        task_uuid = await self._celery_task_manager.submit_task(
+            TaskMetadata(
+                name="run_function",
+                ephemeral=True,
+                queue=TasksQueue.API_WORKER_QUEUE,
+            ),
+            task_filter=task_filter,
+            user_identity=user_identity,
+            function=function,
+            pre_registered_function_job_data=pre_registered_function_job_data,
+            pricing_spec=pricing_spec,
+            job_links=job_links,
+            x_simcore_parent_project_uuid=parent_project_uuid,
+            x_simcore_parent_node_id=parent_node_id,
+        )
+
+        return await self._function_job_service.patch_registered_function_job(
+            user_id=user_identity.user_id,
+            product_name=user_identity.product_name,
+            function_job_id=pre_registered_function_job_data.function_job_id,
+            function_class=function.function_class,
+            job_creation_task_id=TaskID(task_uuid),
+        )

services/api-server/src/simcore_service_api_server/api/routes/functions_routes.py

@@ -1,5 +1,3 @@
-import contextlib
-
 # pylint: disable=too-many-positional-arguments
 from collections.abc import Callable
 from typing import Annotated, Final, Literal
@@ -18,23 +16,17 @@
     RegisteredFunctionJob,
     RegisteredFunctionJobCollection,
 )
-from models_library.api_schemas_rpc_async_jobs.async_jobs import AsyncJobFilter
 from models_library.functions import FunctionJobCollection, FunctionJobID
 from models_library.products import ProductName
 from models_library.projects import ProjectID
 from models_library.projects_nodes_io import NodeID
 from models_library.users import UserID
-from servicelib.celery.models import TaskFilter, TaskID, TaskMetadata, TasksQueue
 from servicelib.fastapi.dependencies import get_reverse_url_mapper
 from servicelib.utils import limited_gather
 
 from ..._service_function_jobs import FunctionJobService
 from ..._service_function_jobs_task_client import FunctionJobTaskClientService
 from ..._service_functions import FunctionService
-from ...celery_worker.worker_tasks.functions_tasks import (
-    run_function as run_function_task,
-)
-from ...exceptions.function_errors import FunctionJobCacheNotFoundError
 from ...models.pagination import Page, PaginationParams
 from ...models.schemas.errors import ErrorGet
 from ...models.schemas.jobs import JobPricingSpecification
@@ -45,7 +37,6 @@
     get_current_user_id,
     get_product_name,
 )
-from ..dependencies.celery import ASYNC_JOB_CLIENT_NAME, get_task_manager
 from ..dependencies.services import (
     get_function_job_service,
     get_function_job_task_client_service,
@@ -344,7 +335,6 @@ async def run_function(
     x_simcore_parent_node_id: Annotated[NodeID | Literal["null"], Header()],
 ) -> RegisteredFunctionJob:
     # preprocess inputs
-    task_manager = get_task_manager(request.app)
     parent_project_uuid = (
         x_simcore_parent_project_uuid
         if isinstance(x_simcore_parent_project_uuid, ProjectID)
@@ -357,55 +347,15 @@ async def run_function(
     )
     pricing_spec = JobPricingSpecification.create_from_headers(request.headers)
     job_links = await function_service.get_function_job_links(to_run_function, url_for)
-    job_inputs = await function_job_service.create_function_job_inputs(
-        function=to_run_function, function_inputs=function_inputs
-    )
-
-    # check if results are cached
-    with contextlib.suppress(FunctionJobCacheNotFoundError):
-        return await function_job_task_client_service.get_cached_function_job(
-            function=to_run_function,
-            job_inputs=job_inputs,
-        )
-
-    pre_registered_function_job_data = (
-        await function_job_service.pre_register_function_job(
-            function=to_run_function,
-            job_inputs=job_inputs,
-        )
-    )
 
-    # run function in celery task
-    job_filter = AsyncJobFilter(
-        user_id=user_identity.user_id,
-        product_name=user_identity.product_name,
-        client_name=ASYNC_JOB_CLIENT_NAME,
-    )
-    task_filter = TaskFilter.model_validate(job_filter.model_dump())
-    task_name = run_function_task.__name__
-
-    task_uuid = await task_manager.submit_task(
-        TaskMetadata(
-            name=task_name,
-            ephemeral=True,
-            queue=TasksQueue.API_WORKER_QUEUE,
-        ),
-        task_filter=task_filter,
-        user_identity=user_identity,
+    return await function_job_task_client_service.create_function_job_creation_task(
         function=to_run_function,
-        pre_registered_function_job_data=pre_registered_function_job_data,
+        function_inputs=function_inputs,
+        user_identity=user_identity,
         pricing_spec=pricing_spec,
         job_links=job_links,
-        x_simcore_parent_project_uuid=parent_project_uuid,
-        x_simcore_parent_node_id=parent_node_id,
-    )
-
-    return await function_job_service.patch_registered_function_job(
-        user_id=user_identity.user_id,
-        product_name=user_identity.product_name,
-        function_job_id=pre_registered_function_job_data.function_job_id,
-        function_class=to_run_function.function_class,
-        job_creation_task_id=TaskID(task_uuid),
+        parent_project_uuid=parent_project_uuid,
+        parent_node_id=parent_node_id,
     )
 
 
@@ -463,19 +413,30 @@ async def map_function(
     x_simcore_parent_project_uuid: Annotated[ProjectID | Literal["null"], Header()],
     x_simcore_parent_node_id: Annotated[NodeID | Literal["null"], Header()],
 ) -> RegisteredFunctionJobCollection:
+    parent_project_uuid = (
+        x_simcore_parent_project_uuid
+        if isinstance(x_simcore_parent_project_uuid, ProjectID)
+        else None
+    )
+    parent_node_id = (
+        x_simcore_parent_node_id
+        if isinstance(x_simcore_parent_node_id, NodeID)
+        else None
+    )
+    pricing_spec = JobPricingSpecification.create_from_headers(request.headers)
+    job_links = await function_service.get_function_job_links(to_run_function, url_for)
 
     async def _run_single_function(function_inputs: FunctionInputs) -> FunctionJobID:
-        result = await run_function(
-            request=request,
-            user_identity=user_identity,
-            to_run_function=to_run_function,
-            url_for=url_for,
-            function_inputs=function_inputs,
-            function_service=function_service,
-            function_job_service=function_jobs_service,
-            function_job_task_client_service=function_job_task_client_service,
-            x_simcore_parent_project_uuid=x_simcore_parent_project_uuid,
-            x_simcore_parent_node_id=x_simcore_parent_node_id,
+        result = (
+            await function_job_task_client_service.create_function_job_creation_task(
+                function=to_run_function,
+                function_inputs=function_inputs,
+                user_identity=user_identity,
+                pricing_spec=pricing_spec,
+                job_links=job_links,
+                parent_project_uuid=parent_project_uuid,
+                parent_node_id=parent_node_id,
+            )
         )
         return result.uid
 

services/api-server/tests/unit/api_functions/test_api_routers_function_jobs.py

@@ -36,10 +36,11 @@
 from servicelib.celery.models import TaskFilter, TaskState, TaskStatus, TaskUUID
 from simcore_service_api_server._meta import API_VTAG
 from simcore_service_api_server._service_function_jobs_task_client import (
+    _JOB_CREATION_TASK_NOT_YET_SCHEDULED_STATUS,
     _JOB_CREATION_TASK_STATUS_PREFIX,
     FunctionJobTaskClientService,
 )
-from simcore_service_api_server.api.routes import function_jobs_routes
+from simcore_service_api_server.api.dependencies import services as service_dependencies
 from simcore_service_api_server.models.schemas.jobs import JobStatus
 
 _faker = Faker()
@@ -275,7 +276,9 @@ async def test_get_function_job_status(
     mocked_app_dependencies: None,
     client: AsyncClient,
     mocker: MockerFixture,
-    mock_handler_in_functions_rpc_interface: Callable[[str, Any], None],
+    mock_handler_in_functions_rpc_interface: Callable[
+        [str, Any, Exception | None, Callable | None], MockType
+    ],
     mock_registered_project_function_job: RegisteredProjectFunctionJob,
     mock_registered_project_function: RegisteredProjectFunction,
     mock_method_in_jobs_service: Callable[[str, Any], None],
@@ -286,12 +289,12 @@ async def test_get_function_job_status(
     celery_task_state: TaskState,
 ) -> None:
 
-    _expected_return_status = (
-        status.HTTP_500_INTERNAL_SERVER_ERROR
-        if job_status not in ("SUCCESS", "FAILED")
-        and (project_job_id is None and job_creation_task_id is None)
-        else status.HTTP_200_OK
-    )
+    _expected_return_status = status.HTTP_200_OK
+    #     status.HTTP_500_INTERNAL_SERVER_ERROR
+    #     if job_status not in ("SUCCESS", "FAILED")
+    #     and (project_job_id is None and job_creation_task_id is None)
+    #     else status.HTTP_200_OK
+    # )
 
     def _mock_task_manager(*args, **kwargs) -> CeleryTaskManager:
         async def _get_task_status(
@@ -318,7 +321,7 @@ async def _get_task_status(
         obj.get_task_status = _get_task_status
         return obj
 
-    mocker.patch.object(function_jobs_routes, "get_task_manager", _mock_task_manager)
+    mocker.patch.object(service_dependencies, "get_task_manager", _mock_task_manager)
 
     mock_handler_in_functions_rpc_interface(
         "get_function_job",
@@ -329,13 +332,14 @@ async def _get_task_status(
                 "job_creation_task_id": job_creation_task_id,
             }
         ),
+        None,
+        None,
     )
     mock_handler_in_functions_rpc_interface(
-        "get_function", mock_registered_project_function
+        "get_function", mock_registered_project_function, None, None
     )
     mock_handler_in_functions_rpc_interface(
-        "get_function_job_status",
-        FunctionJobStatus(status=job_status),
+        "get_function_job_status", FunctionJobStatus(status=job_status), None, None
     )
     mock_method_in_jobs_service(
         "inspect_study_job",
@@ -347,25 +351,35 @@ async def _get_task_status(
             state=RunningState(value=job_status),
         ),
     )
+
+    async def _update_function_job_status_side_effect(*args, **kwargs):
+        return kwargs["job_status"]
+
     mock_handler_in_functions_rpc_interface(
         "update_function_job_status",
-        FunctionJobStatus(status=job_status),
+        None,
+        None,
+        _update_function_job_status_side_effect,
     )
 
     response = await client.get(
         f"{API_VTAG}/function_jobs/{mock_registered_project_function_job.uid}/status",
         auth=auth,
     )
     assert response.status_code == _expected_return_status
-    if response.status_code == status.HTTP_200_OK:
-        data = response.json()
-        if project_job_id is not None or job_status in ("SUCCESS", "FAILED"):
-            assert data["status"] == job_status
-        else:
-            assert (
-                data["status"]
-                == f"{_JOB_CREATION_TASK_STATUS_PREFIX}{celery_task_state}"
-            )
+    data = response.json()
+    if (project_job_id is not None and job_creation_task_id is not None) or (
+        job_status in ("SUCCESS", "FAILED")
+    ):
+        assert data["status"] == job_status
+    elif project_job_id is None and job_creation_task_id is None:
+        assert data["status"] == _JOB_CREATION_TASK_NOT_YET_SCHEDULED_STATUS
+    elif project_job_id is None and job_creation_task_id is not None:
+        assert (
+            data["status"] == f"{_JOB_CREATION_TASK_STATUS_PREFIX}{celery_task_state}"
+        )
+    else:
+        pytest.fail("Unexpected combination of parameters")
 
 
 @pytest.mark.parametrize(

services/api-server/tests/unit/api_functions/test_api_routers_functions.py

@@ -344,7 +344,7 @@ async def test_run_map_function_not_allowed(
     """Test that running a function is not allowed."""
 
     mocker.patch(
-        "simcore_service_api_server.api.routes.functions_routes.get_task_manager",
+        "simcore_service_api_server.api.dependencies.services.get_task_manager",
         return_value=mocker.MagicMock(spec=CeleryTaskManager),
     )