Add tests to status/output fetching function jobs

Author: wvangeit

packages/service-library/src/servicelib/rabbitmq/rpc_interfaces/webserver/functions/functions_rpc_interface.py

@@ -18,6 +18,7 @@
 )
 from models_library.functions import (
     FunctionJobStatus,
+    FunctionOutputs,
     FunctionUserAccessRights,
     FunctionUserApiAccessRights,
 )
@@ -319,6 +320,64 @@ async def get_function_job_status(
     return TypeAdapter(FunctionJobStatus).validate_python(result)
 
 
+@log_decorator(_logger, level=logging.DEBUG)
+async def get_function_job_outputs(
+    rabbitmq_rpc_client: RabbitMQRPCClient,
+    *,
+    user_id: UserID,
+    function_job_id: FunctionJobID,
+    product_name: ProductName,
+) -> FunctionOutputs:
+    result = await rabbitmq_rpc_client.request(
+        WEBSERVER_RPC_NAMESPACE,
+        TypeAdapter(RPCMethodName).validate_python("get_function_job_outputs"),
+        function_job_id=function_job_id,
+        user_id=user_id,
+        product_name=product_name,
+    )
+    return TypeAdapter(FunctionOutputs).validate_python(result)
+
+
+@log_decorator(_logger, level=logging.DEBUG)
+async def update_function_job_status(
+    rabbitmq_rpc_client: RabbitMQRPCClient,
+    *,
+    user_id: UserID,
+    product_name: ProductName,
+    function_job_id: FunctionJobID,
+    job_status: FunctionJobStatus,
+) -> FunctionJobStatus:
+    result = await rabbitmq_rpc_client.request(
+        WEBSERVER_RPC_NAMESPACE,
+        TypeAdapter(RPCMethodName).validate_python("update_function_job_status"),
+        function_job_id=function_job_id,
+        job_status=job_status,
+        user_id=user_id,
+        product_name=product_name,
+    )
+    return TypeAdapter(FunctionJobStatus).validate_python(result)
+
+
+@log_decorator(_logger, level=logging.DEBUG)
+async def update_function_job_outputs(
+    rabbitmq_rpc_client: RabbitMQRPCClient,
+    *,
+    user_id: UserID,
+    product_name: ProductName,
+    function_job_id: FunctionJobID,
+    outputs: FunctionOutputs,
+) -> FunctionOutputs:
+    result = await rabbitmq_rpc_client.request(
+        WEBSERVER_RPC_NAMESPACE,
+        TypeAdapter(RPCMethodName).validate_python("update_function_job_outputs"),
+        function_job_id=function_job_id,
+        outputs=outputs,
+        user_id=user_id,
+        product_name=product_name,
+    )
+    return TypeAdapter(FunctionOutputs).validate_python(result)
+
+
 @log_decorator(_logger, level=logging.DEBUG)
 async def delete_function_job(
     rabbitmq_rpc_client: RabbitMQRPCClient,

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

@@ -199,9 +199,7 @@ async def function_job_status(
             user_id=user_id,
             director2_api=director2_api,
         )
-        job_status = FunctionJobStatus(status=job_status.state)
-
-    if (function.function_class == FunctionClass.SOLVER) and (
+    elif (function.function_class == FunctionClass.SOLVER) and (
         function_job.function_class == FunctionClass.SOLVER
     ):
         job_status = await solvers_jobs.inspect_job(
@@ -211,13 +209,20 @@ async def function_job_status(
             user_id=user_id,
             director2_api=director2_api,
         )
-        job_status = FunctionJobStatus(status=job_status.state)
     else:
         raise UnsupportedFunctionFunctionJobClassCombinationError(
             function_class=function.function_class,
             function_job_class=function_job.function_class,
         )
-    return job_status
+
+    new_job_status = FunctionJobStatus(status=job_status.state)
+
+    return await wb_api_rpc.update_function_job_status(
+        function_job_id=function_job.uid,
+        user_id=user_id,
+        product_name=product_name,
+        job_status=new_job_status,
+    )
 
 
 async def get_function_from_functionjobid(
@@ -255,7 +260,7 @@ async def get_function_from_functionjobid(
         changelog=CHANGE_LOGS["function_job_outputs"],
     ),
 )
-async def function_job_outputs(
+async def get_function_job_outputs(
     function_job_id: FunctionJobID,
     webserver_api: Annotated[AuthSession, Depends(get_webserver_session)],
     user_id: Annotated[UserID, Depends(get_current_user_id)],
@@ -271,11 +276,18 @@ async def function_job_outputs(
         product_name=product_name,
     )
 
+    old_job_outputs = await wb_api_rpc.get_function_job_outputs(
+        function_job_id=function_job.uid, user_id=user_id, product_name=product_name
+    )
+
+    if old_job_outputs is not None:
+        return old_job_outputs
+
     if (
         function.function_class == FunctionClass.PROJECT
         and function_job.function_class == FunctionClass.PROJECT
     ):
-        return dict(
+        new_outputs = dict(
             (
                 await studies_jobs.get_study_job_outputs(
                     study_id=function.project_id,
@@ -286,12 +298,11 @@ async def function_job_outputs(
                 )
             ).results
         )
-
-    if (
+    elif (
         function.function_class == FunctionClass.SOLVER
         and function_job.function_class == FunctionClass.SOLVER
     ):
-        return dict(
+        new_outputs = dict(
             (
                 await solvers_jobs_read.get_job_outputs(
                     solver_key=function.solver_key,
@@ -304,4 +315,12 @@ async def function_job_outputs(
                 )
             ).results
         )
-    raise UnsupportedFunctionClassError(function_class=function.function_class)
+    else:
+        raise UnsupportedFunctionClassError(function_class=function.function_class)
+
+    return await wb_api_rpc.update_function_job_outputs(
+        function_job_id=function_job.uid,
+        user_id=user_id,
+        product_name=product_name,
+        outputs=new_outputs,
+    )

services/api-server/src/simcore_service_api_server/services_rpc/wb_api_server.py

@@ -25,6 +25,7 @@
 from models_library.api_schemas_webserver.licensed_items import LicensedItemRpcGetPage
 from models_library.functions import (
     FunctionJobStatus,
+    FunctionOutputs,
     FunctionUserAccessRights,
     FunctionUserApiAccessRights,
 )
@@ -487,6 +488,52 @@ async def get_function_job_status(
             function_job_id=function_job_id,
         )
 
+    async def get_function_job_outputs(
+        self,
+        *,
+        user_id: UserID,
+        product_name: ProductName,
+        function_job_id: FunctionJobID,
+    ) -> FunctionOutputs:
+        return await functions_rpc_interface.get_function_job_outputs(
+            self._client,
+            user_id=user_id,
+            product_name=product_name,
+            function_job_id=function_job_id,
+        )
+
+    async def update_function_job_status(
+        self,
+        *,
+        function_job_id: FunctionJobID,
+        user_id: UserID,
+        product_name: ProductName,
+        job_status: FunctionJobStatus,
+    ) -> FunctionJobStatus:
+        return await functions_rpc_interface.update_function_job_status(
+            self._client,
+            function_job_id=function_job_id,
+            user_id=user_id,
+            product_name=product_name,
+            job_status=job_status,
+        )
+
+    async def update_function_job_outputs(
+        self,
+        *,
+        function_job_id: FunctionJobID,
+        user_id: UserID,
+        product_name: ProductName,
+        outputs: FunctionOutputs,
+    ) -> FunctionOutputs:
+        return await functions_rpc_interface.update_function_job_outputs(
+            self._client,
+            function_job_id=function_job_id,
+            user_id=user_id,
+            product_name=product_name,
+            outputs=outputs,
+        )
+
     async def find_cached_function_jobs(
         self,
         *,

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

@@ -232,3 +232,24 @@ def _mock(
         )
 
     return _mock
+
+
+@pytest.fixture()
+def mock_handler_in_study_jobs_rest_interface(
+    mock_wb_api_server_rpc: MockerFixture,
+) -> Callable[[str, Any, Exception | None], None]:
+    def _mock(
+        handler_name: str = "",
+        return_value: Any = None,
+        exception: Exception | None = None,
+    ) -> None:
+        from simcore_service_api_server.api.routes.functions_routes import studies_jobs
+
+        mock_wb_api_server_rpc.patch.object(
+            studies_jobs,
+            handler_name,
+            return_value=return_value,
+            side_effect=exception,
+        )
+
+    return _mock

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

@@ -1,3 +1,5 @@
+# pylint: disable=unused-argument
+
 import uuid
 from collections.abc import Callable
 from datetime import datetime
@@ -108,12 +110,13 @@ async def test_list_function_jobs(
     )
 
 
-@pytest.mark.parametrize("job_status", ["SUCCESS", "FAILED", "RUNNING"])
+@pytest.mark.parametrize("job_status", ["SUCCESS", "FAILED", "STARTED"])
 async def test_get_function_job_status(
     client: AsyncClient,
     mock_handler_in_functions_rpc_interface: Callable[[str, Any], None],
     mock_registered_function_job: RegisteredProjectFunctionJob,
     mock_registered_project_function: RegisteredProjectFunction,
+    mock_handler_in_study_jobs_rest_interface: Callable[[str, Any], None],
     auth: httpx.BasicAuth,
     job_status: str,
 ) -> None:
@@ -128,16 +131,20 @@ async def test_get_function_job_status(
         "get_function_job_status",
         FunctionJobStatus(status=job_status),
     )
-    mock_handler_in_functions_rpc_interface(
+    mock_handler_in_study_jobs_rest_interface(
         "inspect_study_job",
         JobStatus(
             job_id=uuid.uuid4(),
             submitted_at=datetime.fromisoformat("2023-01-01T00:00:00"),
             started_at=datetime.fromisoformat("2023-01-01T01:00:00"),
             stopped_at=datetime.fromisoformat("2023-01-01T02:00:00"),
-            state=RunningState(job_status),
+            state=RunningState(value=job_status),
         ),
     )
+    mock_handler_in_functions_rpc_interface(
+        "update_function_job_status",
+        FunctionJobStatus(status=job_status),
+    )
 
     response = await client.get(
         f"{API_VTAG}/function_jobs/{mock_registered_function_job.uid}/status",
@@ -146,3 +153,30 @@ async def test_get_function_job_status(
     assert response.status_code == status.HTTP_200_OK
     data = response.json()
     assert data["status"] == job_status
+
+
+@pytest.mark.parametrize("job_outputs", [{"X+Y": 42, "X-Y": 10}])
+async def test_get_function_job_outputs(
+    client: AsyncClient,
+    mock_handler_in_functions_rpc_interface: Callable[[str, Any], None],
+    mock_registered_function_job: RegisteredProjectFunctionJob,
+    mock_registered_project_function: RegisteredProjectFunction,
+    auth: httpx.BasicAuth,
+    job_outputs: dict[str, Any],
+) -> None:
+
+    mock_handler_in_functions_rpc_interface(
+        "get_function_job", mock_registered_function_job
+    )
+    mock_handler_in_functions_rpc_interface(
+        "get_function", mock_registered_project_function
+    )
+    mock_handler_in_functions_rpc_interface("get_function_job_outputs", job_outputs)
+
+    response = await client.get(
+        f"{API_VTAG}/function_jobs/{mock_registered_function_job.uid}/outputs",
+        auth=auth,
+    )
+    assert response.status_code == status.HTTP_200_OK
+    data = response.json()
+    assert data == job_outputs