add test for checking incompatible patch error raised

Author: bisgaard-itis

packages/models-library/src/models_library/functions_errors.py

@@ -158,3 +158,8 @@ class FunctionJobCollectionsExecuteApiAccessDeniedError(FunctionBaseError):
         "User {user_id} does not have the permission to execute function job collections"
     )
     status_code: int = 403  # Forbidden
+
+
+class FunctionJobPatchModelIncompatibleError(FunctionBaseError):
+    msg_template = "Incompatible patch model for Function '{function_id}' in product '{product_name}'."
+    status_code: int = 422

services/web/server/src/simcore_service_webserver/functions/_controller/_functions_rpc.py

@@ -30,6 +30,7 @@
     FunctionJobCollectionsWriteApiAccessDeniedError,
     FunctionJobCollectionWriteAccessDeniedError,
     FunctionJobIDNotFoundError,
+    FunctionJobPatchModelIncompatibleError,
     FunctionJobReadAccessDeniedError,
     FunctionJobsReadApiAccessDeniedError,
     FunctionJobsWriteApiAccessDeniedError,
@@ -93,6 +94,7 @@ async def register_function_job(
     reraise_if_error_type=(
         UnsupportedFunctionJobClassError,
         FunctionJobsWriteApiAccessDeniedError,
+        FunctionJobPatchModelIncompatibleError,
     )
 )
 async def patch_registered_function_job(

services/web/server/src/simcore_service_webserver/functions/_functions_exceptions.py

@@ -7,9 +7,3 @@ class FunctionGroupAccessRightsNotFoundError(WebServerBaseError, RuntimeError):
     msg_template = user_message(
         "Group access rights could not be found for Function '{function_id}' in product '{product_name}'."
     )
-
-
-class IncompatiblePatchModelError(WebServerBaseError, ValueError):
-    msg_template = user_message(
-        "Incompatible patch model for Function '{function_id}' in product '{product_name}'."
-    )

services/web/server/src/simcore_service_webserver/functions/_functions_service.py

@@ -34,6 +34,7 @@
     RegisteredSolverFunctionJob,
 )
 from models_library.functions_errors import (
+    FunctionJobPatchModelIncompatibleError,
     UnsupportedFunctionClassError,
     UnsupportedFunctionJobClassError,
 )
@@ -47,7 +48,6 @@
 from . import _functions_repository
 from ._functions_exceptions import (
     FunctionGroupAccessRightsNotFoundError,
-    IncompatiblePatchModelError,
 )
 
 router = RPCRouter()
@@ -114,7 +114,7 @@ async def patch_registered_function_job(
         function_job_id=function_job_uuid,
     )
     if job.function_class != registered_function_job_patch.function_class:
-        raise IncompatiblePatchModelError(
+        raise FunctionJobPatchModelIncompatibleError(
             function_id=job.function_uuid,
             product_name=product_name,
         )

services/web/server/tests/unit/with_dbs/04/functions/test_function_jobs_controller_rpc.py

@@ -25,6 +25,7 @@
 )
 from models_library.functions_errors import (
     FunctionJobIDNotFoundError,
+    FunctionJobPatchModelIncompatibleError,
     FunctionJobReadAccessDeniedError,
     FunctionJobsReadApiAccessDeniedError,
     FunctionJobWriteAccessDeniedError,
@@ -540,6 +541,71 @@ async def test_patch_registered_function_jobs(
         assert registered_job.solver_job_id == patch.solver_job_id
 
 
+@pytest.mark.parametrize(
+    "user_role",
+    [UserRole.USER],
+)
+@pytest.mark.parametrize(
+    "function_job, patch",
+    [
+        (
+            ProjectFunctionJob(
+                function_uid=_faker.uuid4(),
+                title="Test Function Job",
+                description="A test function job",
+                project_job_id=None,
+                inputs=None,
+                outputs=None,
+                job_creation_task_id=None,
+            ),
+            RegisteredSolverFunctionJobPatch(
+                title=_faker.word(),
+                description=_faker.sentence(),
+                job_creation_task_id=TaskID(_faker.uuid4()),
+                inputs={"input1": _faker.pyint(min_value=0, max_value=1000)},
+                outputs={"output1": _faker.word()},
+                solver_job_id=_faker.uuid4(),
+            ),
+        ),
+    ],
+)
+async def test_incompatible_patch_model_error(
+    client: TestClient,
+    rpc_client: RabbitMQRPCClient,
+    add_user_function_api_access_rights: None,
+    logged_user: UserInfoDict,
+    other_logged_user: UserInfoDict,
+    osparc_product_name: ProductName,
+    mock_function_factory: Callable[[FunctionClass], Function],
+    clean_functions: None,
+    function_job: RegisteredFunctionJob,
+    patch: RegisteredFunctionJobPatch,
+):
+    function = mock_function_factory(function_job.function_class)
+
+    registered_function = await functions_rpc.register_function(
+        rabbitmq_rpc_client=rpc_client,
+        function=function,
+        user_id=logged_user["id"],
+        product_name=osparc_product_name,
+    )
+    function_job.function_uid = registered_function.uid
+    registered_job = await functions_rpc.register_function_job(
+        rabbitmq_rpc_client=rpc_client,
+        function_job=function_job,
+        user_id=logged_user["id"],
+        product_name=osparc_product_name,
+    )
+    with pytest.raises(FunctionJobPatchModelIncompatibleError):
+        registered_job = await functions_rpc.patch_registered_function_job(
+            rabbitmq_rpc_client=rpc_client,
+            user_id=logged_user["id"],
+            function_job_uuid=registered_job.uid,
+            product_name=osparc_product_name,
+            registered_function_job_patch=patch,
+        )
+
+
 @pytest.mark.parametrize(
     "user_role",
     [UserRole.USER],