fix tests

Author: bisgaard-itis

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

@@ -1,5 +1,5 @@
 from dataclasses import dataclass
-from typing import Final, overload
+from typing import overload
 
 import jsonschema
 from common_library.exclude import as_dict_exclude_none
@@ -31,24 +31,16 @@
 from models_library.rpc_pagination import PageLimitInt
 from models_library.users import UserID
 from pydantic import ValidationError
-from servicelib.celery.models import TaskUUID
-from servicelib.celery.task_manager import TaskManager
 from simcore_service_api_server._service_functions import FunctionService
 from simcore_service_api_server.services_rpc.storage import StorageService
 
 from ._service_jobs import JobService
-from .api.routes.tasks import _get_task_filter
 from .models.api_resources import JobLinks
 from .models.domain.functions import PreRegisteredFunctionJobData
 from .models.schemas.jobs import JobInputs, JobPricingSpecification
 from .services_http.webserver import AuthSession
 from .services_rpc.wb_api_server import WbApiRpcClient
 
-_JOB_CREATION_TASK_STATUS_PREFIX: Final[str] = "JOB_CREATION_TASK_STATUS_"
-_JOB_CREATION_TASK_NOT_YET_SCHEDULED_STATUS: Final[str] = (
-    f"{_JOB_CREATION_TASK_STATUS_PREFIX}NOT_YET_SCHEDULED"
-)
-
 
 def join_inputs(
     default_inputs: FunctionInputs | None,
@@ -64,21 +56,6 @@ def join_inputs(
     return {**default_inputs, **function_inputs}
 
 
-async def _celery_task_status(
-    job_creation_task_id: TaskID | None,
-    task_manager: TaskManager,
-    user_id: UserID,
-    product_name: ProductName,
-) -> str:
-    if job_creation_task_id is None:
-        return _JOB_CREATION_TASK_NOT_YET_SCHEDULED_STATUS
-    task_filter = _get_task_filter(user_id, product_name)
-    task_status = await task_manager.get_task_status(
-        task_uuid=TaskUUID(job_creation_task_id), task_filter=task_filter
-    )
-    return f"{_JOB_CREATION_TASK_STATUS_PREFIX}{task_status.task_state}"
-
-
 @dataclass(frozen=True, kw_only=True)
 class FunctionJobService:
     user_id: UserID

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

@@ -93,7 +93,6 @@ class FunctionJobTaskClientService:
     async def list_function_jobs_with_status(
         self,
         *,
-        function: RegisteredFunction,
         filter_by_function_id: FunctionID | None = None,
         filter_by_function_job_ids: list[FunctionJobID] | None = None,
         filter_by_function_job_collection_id: FunctionJobCollectionID | None = None,
@@ -138,7 +137,9 @@ async def list_function_jobs_with_status(
                 if function_job_wso.status.status == RunningState.SUCCESS:
                     function_job_wso.outputs = await self.function_job_outputs(
                         function_job=function_job_wso,
-                        function=function,
+                        function=await self._function_service.get_function(
+                            function_id=function_job_wso.function_uid,
+                        ),
                         stored_job_outputs=None,
                     )
         return function_jobs_list_ws, meta

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

@@ -22,7 +22,6 @@
 from simcore_service_api_server._service_function_jobs_task_client import (
     FunctionJobTaskClientService,
 )
-from sqlalchemy.ext.asyncio import AsyncEngine
 
 from ..._service_function_jobs import FunctionJobService
 from ..._service_functions import FunctionService
@@ -33,8 +32,6 @@
 from ...models.schemas.functions_filters import FunctionJobsListFilters
 from ...services_rpc.wb_api_server import WbApiRpcClient
 from ..dependencies.authentication import get_current_user_id, get_product_name
-from ..dependencies.celery import get_task_manager
-from ..dependencies.database import get_db_asyncpg_engine
 from ..dependencies.functions import (
     get_function_from_functionjob,
     get_function_job_dependency,
@@ -127,15 +124,13 @@ async def list_function_jobs(
         FunctionJobService, Depends(get_function_job_service)
     ],
     filters: Annotated[FunctionJobsListFilters, Depends(get_function_jobs_filters)],
-    function: Annotated[RegisteredFunction, Depends(get_function_from_functionjob)],
     include_status: Annotated[  # noqa: FBT002
         bool, Query(description="Include job status in response")
     ] = False,
 ):
     if include_status:
         function_jobs_list_ws, meta = (
             await function_job_task_client_service.list_function_jobs_with_status(
-                function=function,
                 pagination_offset=page_params.offset,
                 pagination_limit=page_params.limit,
                 filter_by_function_job_ids=filters.function_job_ids,
@@ -240,13 +235,12 @@ async def function_job_status(
         RegisteredFunctionJob, Depends(get_function_job_dependency)
     ],
     function: Annotated[RegisteredFunction, Depends(get_function_from_functionjob)],
-    function_job_service: Annotated[
-        FunctionJobService, Depends(get_function_job_service)
+    function_job_task_client_service: Annotated[
+        FunctionJobTaskClientService, Depends(get_function_job_task_client_service)
     ],
 ) -> FunctionJobStatus:
-    task_manager = get_task_manager(app)
-    return await function_job_service.inspect_function_job(
-        function=function, function_job=function_job, task_manager=task_manager
+    return await function_job_task_client_service.inspect_function_job(
+        function=function, function_job=function_job
     )
 
 
@@ -285,22 +279,16 @@ async def function_job_outputs(
     function_job: Annotated[
         RegisteredFunctionJob, Depends(get_function_job_dependency)
     ],
-    function_job_service: Annotated[
-        FunctionJobService, Depends(get_function_job_service)
+    function_job_task_client_service: Annotated[
+        FunctionJobTaskClientService, Depends(get_function_job_task_client_service)
     ],
     function: Annotated[RegisteredFunction, Depends(get_function_from_functionjob)],
-    user_id: Annotated[UserID, Depends(get_current_user_id)],
-    product_name: Annotated[ProductName, Depends(get_product_name)],
     stored_job_outputs: Annotated[FunctionOutputs, Depends(get_stored_job_outputs)],
-    async_pg_engine: Annotated[AsyncEngine, Depends(get_db_asyncpg_engine)],
 ) -> FunctionOutputs:
-    return await function_job_service.function_job_outputs(
+    return await function_job_task_client_service.function_job_outputs(
         function_job=function_job,
-        user_id=user_id,
-        product_name=product_name,
         function=function,
         stored_job_outputs=stored_job_outputs,
-        async_pg_engine=async_pg_engine,
     )
 
 

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

@@ -29,6 +29,7 @@
 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,
@@ -45,7 +46,11 @@
     get_product_name,
 )
 from ..dependencies.celery import ASYNC_JOB_CLIENT_NAME, get_task_manager
-from ..dependencies.services import get_function_job_service, get_function_service
+from ..dependencies.services import (
+    get_function_job_service,
+    get_function_job_task_client_service,
+    get_function_service,
+)
 from ..dependencies.webserver_rpc import get_wb_api_rpc_client
 from ._constants import (
     FMSG_CHANGELOG_ADDED_IN_VERSION,
@@ -332,6 +337,9 @@ async def run_function(
     function_job_service: Annotated[
         FunctionJobService, Depends(get_function_job_service)
     ],
+    function_job_task_client_service: Annotated[
+        FunctionJobTaskClientService, Depends(get_function_job_task_client_service)
+    ],
     x_simcore_parent_project_uuid: Annotated[ProjectID | Literal["null"], Header()],
     x_simcore_parent_node_id: Annotated[NodeID | Literal["null"], Header()],
 ) -> RegisteredFunctionJob:
@@ -355,7 +363,7 @@ async def run_function(
 
     # check if results are cached
     with contextlib.suppress(FunctionJobCacheNotFoundError):
-        return await function_job_service.get_cached_function_job(
+        return await function_job_task_client_service.get_cached_function_job(
             function=to_run_function,
             job_inputs=job_inputs,
         )
@@ -447,6 +455,9 @@ async def map_function(
     function_jobs_service: Annotated[
         FunctionJobService, Depends(get_function_job_service)
     ],
+    function_job_task_client_service: Annotated[
+        FunctionJobTaskClientService, Depends(get_function_job_task_client_service)
+    ],
     function_service: Annotated[FunctionService, Depends(get_function_service)],
     web_api_rpc_client: Annotated[WbApiRpcClient, Depends(get_wb_api_rpc_client)],
     x_simcore_parent_project_uuid: Annotated[ProjectID | Literal["null"], Header()],
@@ -462,6 +473,7 @@ async def _run_single_function(function_inputs: FunctionInputs) -> FunctionJobID
             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,
         )

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

@@ -38,6 +38,7 @@
 from pytest_simcore.helpers.monkeypatch_envs import setenvs_from_dict
 from pytest_simcore.helpers.typing_env import EnvVarsDict
 from servicelib.rabbitmq._client_rpc import RabbitMQRPCClient
+from simcore_service_api_server.api.dependencies import services
 from simcore_service_api_server.api.dependencies.services import get_rabbitmq_rpc_client
 from simcore_service_api_server.api.routes.functions_routes import get_wb_api_rpc_client
 from simcore_service_api_server.services_rpc.wb_api_server import WbApiRpcClient
@@ -83,6 +84,14 @@ def _():
     return mocker
 
 
+@pytest.fixture
+async def mock_celery_task_manager(app: FastAPI, mocker: MockerFixture) -> MockType:
+    def _(app: FastAPI):
+        return None
+
+    return mocker.patch.object(services, services.get_task_manager.__name__, _)
+
+
 @pytest.fixture
 async def mock_wb_api_server_rpc(app: FastAPI, mocker: MockerFixture) -> MockerFixture:
 

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

@@ -34,11 +34,11 @@
 from pytest_mock import MockerFixture, MockType
 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 import FunctionJobService
-from simcore_service_api_server.api.routes import function_jobs_routes
-from simcore_service_api_server.api.routes.function_jobs_routes import (
+from simcore_service_api_server._service_function_jobs_task_client import (
     _JOB_CREATION_TASK_STATUS_PREFIX,
+    FunctionJobTaskClientService,
 )
+from simcore_service_api_server.api.routes import function_jobs_routes
 from simcore_service_api_server.models.schemas.jobs import JobStatus
 
 _faker = Faker()
@@ -112,6 +112,7 @@ async def test_get_function_job(
 async def test_list_function_jobs(
     client: AsyncClient,
     mock_rabbitmq_rpc_client: MockerFixture,
+    mock_celery_task_manager: MockType,
     mock_handler_in_functions_rpc_interface: Callable[[str, Any], None],
     mock_registered_project_function_job: RegisteredProjectFunctionJob,
     auth: httpx.BasicAuth,
@@ -138,6 +139,7 @@ async def test_list_function_jobs(
 async def test_list_function_jobs_with_status(
     client: AsyncClient,
     mock_rabbitmq_rpc_client: MockerFixture,
+    mock_celery_task_manager: MockType,
     mock_handler_in_functions_rpc_interface: Callable[[str, Any], None],
     mock_registered_project_function: RegisteredProjectFunction,
     mock_registered_project_function_job: RegisteredProjectFunctionJob,
@@ -167,7 +169,7 @@ async def test_list_function_jobs_with_status(
     )
 
     mock_function_job_outputs = mocker.patch.object(
-        FunctionJobService, "function_job_outputs", return_value=mock_outputs
+        FunctionJobTaskClientService, "function_job_outputs", return_value=mock_outputs
     )
     mock_handler_in_functions_rpc_interface("get_function_job_status", mock_status)
     response = await client.get(
@@ -191,6 +193,7 @@ async def test_list_function_jobs_with_status(
 
 async def test_list_function_jobs_with_job_id_filter(
     client: AsyncClient,
+    mock_celery_task_manager: MockType,
     mock_rabbitmq_rpc_client: MockerFixture,
     mock_handler_in_functions_rpc_interface: Callable[[str], MockType],
     mock_registered_project_function_job: RegisteredProjectFunctionJob,