Merge branch 'master' into fix_duplicate_map_inputs

Author: wvangeit

packages/models-library/tests/test__pydantic_models_and_enums.py

@@ -30,7 +30,7 @@ def test_equivalent_enums_are_not_strictly_equal():
 #
 # Here two equivalent enum BUT of type str-enum
 #
-# SEE from models_library.utils.enums.AutoStrEnum
+# SEE from models_library.utils.enums.StrAutoEnum
 # SEE https://docs.pydantic.dev/dev-v2/usage/types/enums/
 #
 

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

@@ -20,8 +20,6 @@
     TaskID,
 )
 from models_library.functions_errors import (
-    FunctionExecuteAccessDeniedError,
-    FunctionsExecuteApiAccessDeniedError,
     UnsupportedFunctionClassError,
     UnsupportedFunctionFunctionJobClassCombinationError,
 )
@@ -245,36 +243,7 @@ async def get_cached_function_job(
         function: RegisteredFunction,
         job_inputs: JobInputs,
     ) -> RegisteredFunctionJob:
-        """
-        N.B. this function checks access rights
-
-        raises FunctionsExecuteApiAccessDeniedError if user cannot execute functions
-        raises FunctionJobCacheNotFoundError if no cached job is found
-
-        """
-
-        user_api_access_rights = (
-            await self._web_rpc_client.get_functions_user_api_access_rights(
-                user_id=self.user_id, product_name=self.product_name
-            )
-        )
-        if not user_api_access_rights.execute_functions:
-            raise FunctionsExecuteApiAccessDeniedError(
-                user_id=self.user_id,
-                function_id=function.uid,
-            )
-
-        user_permissions = await self._web_rpc_client.get_function_user_permissions(
-            function_id=function.uid,
-            user_id=self.user_id,
-            product_name=self.product_name,
-        )
-        if not user_permissions.execute:
-            raise FunctionExecuteAccessDeniedError(
-                user_id=self.user_id,
-                function_id=function.uid,
-            )
-
+        """Raises FunctionJobCacheNotFoundError if no cached job is found"""
         if cached_function_jobs := await self._web_rpc_client.find_cached_function_jobs(
             function_id=function.uid,
             inputs=job_inputs.values,

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

@@ -5,7 +5,11 @@
 
 from common_library.exclude import as_dict_exclude_none
 from models_library.functions import FunctionClass, FunctionID, RegisteredFunction
-from models_library.functions_errors import UnsupportedFunctionClassError
+from models_library.functions_errors import (
+    FunctionExecuteAccessDeniedError,
+    FunctionsExecuteApiAccessDeniedError,
+    UnsupportedFunctionClassError,
+)
 from models_library.products import ProductName
 from models_library.rest_pagination import (
     MAXIMUM_NUMBER_OF_ITEMS_PER_PAGE,
@@ -76,3 +80,37 @@ async def get_function(self, function_id: FunctionID) -> RegisteredFunction:
             product_name=self.product_name,
             function_id=function_id,
         )
+
+    async def check_execute_function_permission(
+        self,
+        *,
+        function: RegisteredFunction,
+    ) -> None:
+        """
+        Check execute permissions for a user on a function
+
+        raises FunctionsExecuteApiAccessDeniedError if user cannot execute functions via the functions API
+        raises FunctionExecuteAccessDeniedError if user cannot execute this functions
+        """
+
+        user_api_access_rights = (
+            await self._web_rpc_client.get_functions_user_api_access_rights(
+                user_id=self.user_id, product_name=self.product_name
+            )
+        )
+        if not user_api_access_rights.execute_functions:
+            raise FunctionsExecuteApiAccessDeniedError(
+                user_id=self.user_id,
+                function_id=function.uid,
+            )
+
+        user_permissions = await self._web_rpc_client.get_function_user_permissions(
+            function_id=function.uid,
+            user_id=self.user_id,
+            product_name=self.product_name,
+        )
+        if not user_permissions.execute:
+            raise FunctionExecuteAccessDeniedError(
+                user_id=self.user_id,
+                function_id=function.uid,
+            )

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

@@ -343,6 +343,10 @@ async def run_function(
         else None
     )
     pricing_spec = JobPricingSpecification.create_from_headers(request.headers)
+
+    await function_service.check_execute_function_permission(
+        function=to_run_function,
+    )
     job_links = await function_service.get_function_job_links(to_run_function, url_for)
 
     return await function_job_task_client_service.create_function_job_creation_task(
@@ -418,6 +422,10 @@ async def map_function(
         else None
     )
     pricing_spec = JobPricingSpecification.create_from_headers(request.headers)
+
+    await function_service.check_execute_function_permission(
+        function=to_run_function,
+    )
     job_links = await function_service.get_function_job_links(to_run_function, url_for)
 
     async def _run_single_function(function_inputs: FunctionInputs) -> FunctionJobID:

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/services/generic_scheduler/__init__.py

@@ -10,6 +10,7 @@
     get_step_group_proxy,
     get_step_store_proxy,
 )
+from ._errors import NoDataFoundError
 from ._event_after_registration import (
     register_to_start_after_on_executed_completed,
     register_to_start_after_on_reverted_completed,
@@ -24,21 +25,28 @@
 )
 from ._operation import (
     BaseStep,
+    BaseStepGroup,
     Operation,
     OperationRegistry,
     ParallelStepGroup,
     SingleStepGroup,
 )
-from ._store import OperationContextProxy, StepGroupProxy, StepStoreProxy
+from ._store import (
+    OperationContextProxy,
+    StepGroupProxy,
+    StepStoreProxy,
+)
 
 __all__: tuple[str, ...] = (
     "BaseStep",
+    "BaseStepGroup",
     "cancel_operation",
     "generic_scheduler_lifespan",
     "get_operation_context_proxy",
     "get_operation_name_or_none",
     "get_step_group_proxy",
     "get_step_store_proxy",
+    "NoDataFoundError",
     "Operation",
     "OperationContextProxy",
     "OperationName",

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/services/generic_scheduler/_core.py

@@ -30,6 +30,7 @@
 from ._errors import (
     CannotCancelWhileWaitingForManualInterventionError,
     NoDataFoundError,
+    OperationInitialContextKeyNotFoundError,
     OperationNotCancellableError,
     StepNameNotInCurrentGroupError,
     StepNotInErrorStateError,
@@ -106,6 +107,12 @@ async def start_operation(
         # check if operation is registered
         operation = OperationRegistry.get_operation(operation_name)
 
+        for required_key in operation.initial_context_required_keys:
+            if required_key not in initial_operation_context:
+                raise OperationInitialContextKeyNotFoundError(
+                    operation_name=operation_name, required_key=required_key
+                )
+
         # NOTE: to ensure reproducibility of operations, the
         # operation steps cannot overwrite keys in the
         # initial context with their results
@@ -237,6 +244,8 @@ async def restart_operation_step_stuck_in_error(
         """
         Force a step stuck in an error state to retry.
         Will raise errors if step cannot be retried.
+
+        raises NoDataFoundError
         """
         schedule_data_proxy = ScheduleDataStoreProxy(
             store=self._store, schedule_id=schedule_id
@@ -742,6 +751,8 @@ async def cancel_operation(app: FastAPI, schedule_id: ScheduleId) -> None:
 
     `reverting` refers to the act of reverting the effects of a step
     that has already been completed (eg: remove a created network)
+
+    raises NoDataFoundError
     """
     await Core.get_from_app_state(app).cancel_operation(schedule_id)
 
@@ -763,6 +774,8 @@ async def restart_operation_step_stuck_in_manual_intervention_during_execute(
     `waiting for manual intervention` refers to a step that has failed and exhausted
     all retries and is now waiting for a human to fix the issue (eg: storage service
     is reachable once again)
+
+    raises NoDataFoundError
     """
     await Core.get_from_app_state(app).restart_operation_step_stuck_in_error(
         schedule_id, step_name, in_manual_intervention=True
@@ -778,6 +791,8 @@ async def restart_operation_step_stuck_during_revert(
     `stuck step` is a step that has failed and exhausted all retries
     `reverting` refers to the act of reverting the effects of a step
     that has already been completed (eg: remove a created network)
+
+    raises NoDataFoundError
     """
     await Core.get_from_app_state(app).restart_operation_step_stuck_in_error(
         schedule_id, step_name, in_manual_intervention=False

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/services/generic_scheduler/_errors.py

@@ -19,7 +19,13 @@ class OperationNotFoundError(BaseGenericSchedulerError):
     )
 
 
-class StepNotFoundInoperationError(BaseGenericSchedulerError):
+class OperationInitialContextKeyNotFoundError(BaseGenericSchedulerError):
+    msg_template: str = (
+        "Operation '{operation_name}' required_key='{required_key}' not in initial_operation_context"
+    )
+
+
+class StepNotFoundInOperationError(BaseGenericSchedulerError):
     msg_template: str = (
         "Step '{step_name}' not found steps_names='{steps_names}' for operation '{operation_name}'"
     )

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/services/generic_scheduler/_event_after.py

@@ -5,6 +5,7 @@
 from servicelib.logging_utils import log_context
 
 from ._core import start_operation
+from ._errors import OperationInitialContextKeyNotFoundError
 from ._models import (
     EventType,
     OperationContext,
@@ -50,7 +51,12 @@ async def register_to_start_after(
             return
 
         # ensure operation exists
-        OperationRegistry.get_operation(to_start.operation_name)
+        operation = OperationRegistry.get_operation(to_start.operation_name)
+        for required_key in operation.initial_context_required_keys:
+            if required_key not in to_start.initial_context:
+                raise OperationInitialContextKeyNotFoundError(
+                    operation_name=to_start.operation_name, required_key=required_key
+                )
 
         await events_proxy.create_or_update_multiple(
             {

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/services/generic_scheduler/_operation.py

@@ -9,7 +9,7 @@
 from ._errors import (
     OperationAlreadyRegisteredError,
     OperationNotFoundError,
-    StepNotFoundInoperationError,
+    StepNotFoundInOperationError,
 )
 from ._models import (
     ALL_RESERVED_CONTEXT_KEYS,
@@ -232,9 +232,17 @@ def get_step_subgroup_to_run(self) -> StepsSubGroup:
 
 class Operation:
     def __init__(
-        self, *step_groups: BaseStepGroup, is_cancellable: bool = True
+        self,
+        *step_groups: BaseStepGroup,
+        initial_context_required_keys: set[str] | None = None,
+        is_cancellable: bool = True,
     ) -> None:
         self.step_groups = list(step_groups)
+        self.initial_context_required_keys = (
+            set()
+            if initial_context_required_keys is None
+            else initial_context_required_keys
+        )
         self.is_cancellable = is_cancellable
 
     def __repr__(self) -> str:
@@ -382,7 +390,7 @@ def get_step(
 
         steps_names = set(cls._OPERATIONS[operation_name]["steps"].keys())
         if step_name not in steps_names:
-            raise StepNotFoundInoperationError(
+            raise StepNotFoundInOperationError(
                 step_name=step_name,
                 operation_name=operation_name,
                 steps_names=steps_names,

services/dynamic-scheduler/tests/unit/services/generic_scheduler/test__operation.py

@@ -5,7 +5,7 @@
 from simcore_service_dynamic_scheduler.services.generic_scheduler._errors import (
     OperationAlreadyRegisteredError,
     OperationNotFoundError,
-    StepNotFoundInoperationError,
+    StepNotFoundInOperationError,
 )
 from simcore_service_dynamic_scheduler.services.generic_scheduler._models import (
     ALL_RESERVED_CONTEXT_KEYS,
@@ -237,7 +237,7 @@ def test_operation_registry_raises_errors():
     with pytest.raises(OperationNotFoundError):
         OperationRegistry.get_step("non_existing", "BS1")
 
-    with pytest.raises(StepNotFoundInoperationError):
+    with pytest.raises(StepNotFoundInOperationError):
         OperationRegistry.get_step("op1", "non_existing")