Merge branch 'master' into improve-celery-task-error-messages-and-improve-taskid-name

Author: bisgaard-itis

packages/pytest-simcore/src/pytest_simcore/helpers/docker.py

@@ -37,7 +37,7 @@ class ContainerStatus(str, Enum):
 
 _COLOR_ENCODING_RE = re.compile(r"\x1B\[([0-9]{1,2}(;[0-9]{1,2})?)?[mGK]")
 _MAX_PATH_CHAR_LEN_ALLOWED = 260
-_kFILENAME_TOO_LONG = 36
+_FILENAME_TOO_LONG = 36
 _NORMPATH_COUNT = 0
 
 
@@ -94,7 +94,7 @@ def get_service_published_port(
         )
 
         for target_port in ports_to_look_for:
-            target_port = int(target_port)
+            target_port = int(target_port)  # noqa: PLW2901
             for p in service_ports:
                 if p["TargetPort"] == target_port:
                     published_port = p["PublishedPort"]
@@ -158,7 +158,7 @@ def run_docker_compose_config(
     args = [f"{docker_compose_path}", *bash_options]
     print(" ".join(args))
 
-    process = subprocess.run(
+    process = subprocess.run(  # noqa: S603
         args,
         cwd=project_dir,
         capture_output=True,
@@ -189,7 +189,7 @@ def shorten_path(filename: str) -> Path:
     # This helper function tries to normalize the path
     # Another possibility would be that the path has some
     # problematic characters but so far we did not find any case ...
-    global _NORMPATH_COUNT  # pylint: disable=global-statement
+    global _NORMPATH_COUNT  # pylint: disable=global-statement  # noqa: PLW0603
 
     if len(filename) > _MAX_PATH_CHAR_LEN_ALLOWED:
         _NORMPATH_COUNT += 1
@@ -215,7 +215,7 @@ def safe_artifact_name(name: str) -> str:
     return BANNED_CHARS_FOR_ARTIFACTS.sub("_", name)
 
 
-def save_docker_infos(destination_dir: Path):
+def save_docker_infos(destination_dir: Path):  # noqa: C901
     client = docker.from_env()
 
     # Includes stop containers, which might be e.g. failing tasks
@@ -228,7 +228,7 @@ def save_docker_infos(destination_dir: Path):
             destination_dir.mkdir(parents=True, exist_ok=True)
 
         except OSError as err:
-            if err.errno == _kFILENAME_TOO_LONG:
+            if err.errno == _FILENAME_TOO_LONG:
                 destination_dir = shorten_path(err.filename)
                 destination_dir.mkdir(parents=True, exist_ok=True)
 
@@ -245,7 +245,7 @@ def save_docker_infos(destination_dir: Path):
                     )
 
                 except OSError as err:
-                    if err.errno == _kFILENAME_TOO_LONG:
+                    if err.errno == _FILENAME_TOO_LONG:
                         shorten_path(err.filename).write_text(
                             _COLOR_ENCODING_RE.sub("", logs)
                         )
@@ -256,12 +256,12 @@ def save_docker_infos(destination_dir: Path):
                         json.dumps(container.attrs, indent=2)
                     )
                 except OSError as err:
-                    if err.errno == _kFILENAME_TOO_LONG:
+                    if err.errno == _FILENAME_TOO_LONG:
                         shorten_path(err.filename).write_text(
                             json.dumps(container.attrs, indent=2)
                         )
 
-            except Exception as err:  # pylint: disable=broad-except  # noqa: PERF203
+            except Exception as err:  # pylint: disable=broad-except
                 if container.status != ContainerStatus.created:
                     print(
                         f"Error while dumping {container.name=}, {container.status=}.\n\t{err=}"

packages/pytest-simcore/src/pytest_simcore/helpers/paused_container.py

@@ -0,0 +1,64 @@
+from collections.abc import AsyncIterator, Callable
+from contextlib import AbstractAsyncContextManager, asynccontextmanager
+from typing import TYPE_CHECKING, Protocol
+
+from tenacity.asyncio import AsyncRetrying
+from tenacity.retry import retry_if_exception_type
+from tenacity.stop import stop_after_delay
+from tenacity.wait import wait_fixed
+
+if TYPE_CHECKING:
+    from servicelib.rabbitmq import RabbitMQClient
+    from servicelib.redis import RedisClientSDK
+
+
+class _ClientWithPingProtocol(Protocol):
+    async def ping(self) -> bool: ...
+
+
+@asynccontextmanager
+async def _paused_container(
+    paused_container: Callable[[str], AbstractAsyncContextManager[None]],
+    container_name: str,
+    client: _ClientWithPingProtocol,
+) -> AsyncIterator[None]:
+    async with paused_container(container_name):
+        yield
+
+    async for attempt in AsyncRetrying(
+        wait=wait_fixed(0.1),
+        stop=stop_after_delay(10),
+        reraise=True,
+        retry=retry_if_exception_type(AssertionError),
+    ):
+        with attempt:
+            assert await client.ping() is True
+
+
+@asynccontextmanager
+async def pause_rabbit(
+    paused_container: Callable[[str], AbstractAsyncContextManager[None]],
+    rabbit_client: "RabbitMQClient",
+) -> AsyncIterator[None]:
+    """
+    Pause RabbitMQ container during the context block,
+    ensuring it's fully down before and back up after.
+    """
+    async with _paused_container(paused_container, "rabbit", rabbit_client):
+        yield
+
+
+@asynccontextmanager
+async def pause_redis(
+    paused_container: Callable[[str], AbstractAsyncContextManager[None]],
+    redis_client: "RedisClientSDK",
+) -> AsyncIterator[None]:
+    """
+    Pause Redis container during the context block,
+    saving a DB snapshot first for a clean restore point.
+    Ensures Redis is down before yielding, and back up after.
+    """
+    await redis_client.redis.save()
+
+    async with _paused_container(paused_container, "redis", redis_client):
+        yield

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

@@ -1,5 +1,5 @@
 import logging
-from typing import Literal
+from typing import Annotated, Literal
 
 from models_library.api_schemas_webserver import WEBSERVER_RPC_NAMESPACE
 from models_library.api_schemas_webserver.functions import (
@@ -32,13 +32,15 @@
 from models_library.rest_ordering import OrderBy
 from models_library.rest_pagination import PageMetaInfoLimitOffset
 from models_library.users import UserID
-from pydantic import TypeAdapter
+from pydantic import PositiveInt, TypeAdapter
 
 from .....logging_utils import log_decorator
 from .... import RabbitMQRPCClient
 
 _logger = logging.getLogger(__name__)
 
+_FUNCTION_RPC_TIMEOUT_SEC: Annotated[int, PositiveInt] = 30
+
 
 @log_decorator(_logger, level=logging.DEBUG)
 async def register_function(
@@ -54,6 +56,7 @@ async def register_function(
         function=function,
         user_id=user_id,
         product_name=product_name,
+        timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
     )
     return TypeAdapter(RegisteredFunction).validate_python(
         result
@@ -74,6 +77,7 @@ async def get_function(
         function_id=function_id,
         user_id=user_id,
         product_name=product_name,
+        timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
     )
     return TypeAdapter(RegisteredFunction).validate_python(result)
 
@@ -92,6 +96,7 @@ async def get_function_input_schema(
         function_id=function_id,
         user_id=user_id,
         product_name=product_name,
+        timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
     )
     return TypeAdapter(FunctionInputSchema).validate_python(result)
 
@@ -110,6 +115,7 @@ async def get_function_output_schema(
         function_id=function_id,
         user_id=user_id,
         product_name=product_name,
+        timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
     )
     return TypeAdapter(FunctionOutputSchema).validate_python(result)
 
@@ -128,6 +134,7 @@ async def delete_function(
         function_id=function_id,
         user_id=user_id,
         product_name=product_name,
+        timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
     )
     assert result is None  # nosec
     return result
@@ -158,6 +165,7 @@ async def list_functions(
             filter_by_function_class=filter_by_function_class,
             search_by_function_title=search_by_function_title,
             search_by_multi_columns=search_by_multi_columns,
+            timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
         )
     )
     return TypeAdapter(
@@ -188,6 +196,7 @@ async def list_function_jobs(
             filter_by_function_id=filter_by_function_id,
             filter_by_function_job_ids=filter_by_function_job_ids,
             filter_by_function_job_collection_id=filter_by_function_job_collection_id,
+            timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
         )
     )
     return TypeAdapter(
@@ -220,6 +229,7 @@ async def list_function_jobs_with_status(
         filter_by_function_id=filter_by_function_id,
         filter_by_function_job_ids=filter_by_function_job_ids,
         filter_by_function_job_collection_id=filter_by_function_job_collection_id,
+        timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
     )
     return TypeAdapter(
         tuple[
@@ -247,6 +257,7 @@ async def list_function_job_collections(
         filters=filters,
         user_id=user_id,
         product_name=product_name,
+        timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
     )
     return TypeAdapter(
         tuple[list[RegisteredFunctionJobCollection], PageMetaInfoLimitOffset]
@@ -269,6 +280,7 @@ async def update_function_title(
         title=title,
         user_id=user_id,
         product_name=product_name,
+        timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
     )
     return TypeAdapter(RegisteredFunction).validate_python(result)
 
@@ -289,6 +301,7 @@ async def update_function_description(
         description=description,
         user_id=user_id,
         product_name=product_name,
+        timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
     )
     return TypeAdapter(RegisteredFunction).validate_python(result)
 
@@ -309,6 +322,7 @@ async def run_function(
         inputs=inputs,
         user_id=user_id,
         product_name=product_name,
+        timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
     )
     return TypeAdapter(RegisteredFunctionJob).validate_python(
         result
@@ -329,6 +343,7 @@ async def register_function_job(
         function_job=function_job,
         user_id=user_id,
         product_name=product_name,
+        timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
     )
     return TypeAdapter(RegisteredFunctionJob).validate_python(
         result
@@ -351,6 +366,7 @@ async def patch_registered_function_job(
         product_name=product_name,
         function_job_uuid=function_job_uuid,
         registered_function_job_patch=registered_function_job_patch,
+        timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
     )
     return TypeAdapter(RegisteredFunctionJob).validate_python(
         result
@@ -371,6 +387,7 @@ async def get_function_job(
         function_job_id=function_job_id,
         user_id=user_id,
         product_name=product_name,
+        timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
     )
 
     return TypeAdapter(RegisteredFunctionJob).validate_python(result)
@@ -390,6 +407,7 @@ async def get_function_job_status(
         function_job_id=function_job_id,
         user_id=user_id,
         product_name=product_name,
+        timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
     )
     return TypeAdapter(FunctionJobStatus).validate_python(result)
 
@@ -408,6 +426,7 @@ async def get_function_job_outputs(
         function_job_id=function_job_id,
         user_id=user_id,
         product_name=product_name,
+        timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
     )
     return TypeAdapter(FunctionOutputs).validate_python(result)
 
@@ -430,6 +449,7 @@ async def update_function_job_status(
         user_id=user_id,
         product_name=product_name,
         check_write_permissions=check_write_permissions,
+        timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
     )
     return TypeAdapter(FunctionJobStatus).validate_python(result)
 
@@ -452,6 +472,7 @@ async def update_function_job_outputs(
         user_id=user_id,
         product_name=product_name,
         check_write_permissions=check_write_permissions,
+        timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
     )
     return TypeAdapter(FunctionOutputs).validate_python(result)
 
@@ -470,6 +491,7 @@ async def delete_function_job(
         function_job_id=function_job_id,
         user_id=user_id,
         product_name=product_name,
+        timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
     )
     assert result is None  # nosec
 
@@ -490,6 +512,7 @@ async def find_cached_function_jobs(
         inputs=inputs,
         user_id=user_id,
         product_name=product_name,
+        timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
     )
     if result is None:
         return None
@@ -510,6 +533,7 @@ async def register_function_job_collection(
         function_job_collection=function_job_collection,
         user_id=user_id,
         product_name=product_name,
+        timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
     )
     return TypeAdapter(RegisteredFunctionJobCollection).validate_python(result)
 
@@ -528,6 +552,7 @@ async def get_function_job_collection(
         function_job_collection_id=function_job_collection_id,
         user_id=user_id,
         product_name=product_name,
+        timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
     )
     return TypeAdapter(RegisteredFunctionJobCollection).validate_python(result)
 
@@ -546,6 +571,7 @@ async def delete_function_job_collection(
         function_job_collection_id=function_job_collection_id,
         user_id=user_id,
         product_name=product_name,
+        timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
     )
     assert result is None  # nosec
 
@@ -564,6 +590,7 @@ async def get_function_user_permissions(
         function_id=function_id,
         user_id=user_id,
         product_name=product_name,
+        timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
     )
     return TypeAdapter(FunctionUserAccessRights).validate_python(result)
 
@@ -582,6 +609,7 @@ async def get_functions_user_api_access_rights(
         ),
         user_id=user_id,
         product_name=product_name,
+        timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
     )
     return TypeAdapter(FunctionUserApiAccessRights).validate_python(result)
 
@@ -614,6 +642,7 @@ async def set_group_permissions(
         read=read,
         write=write,
         execute=execute,
+        timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
     )
     return TypeAdapter(
         list[tuple[FunctionID | FunctionJobID, FunctionGroupAccessRights]]