Merge branch 'master' into pr-osparc-long-running-refactor-2

Author: GitHK

.github/workflows/ci-testing-deploy.yml

@@ -335,7 +335,7 @@ jobs:
   unit-test-webserver-02:
     needs: changes
     if: ${{ needs.changes.outputs.webserver == 'true' || github.event_name == 'push' || github.event.inputs.force_all_builds == 'true' }}
-    timeout-minutes: 25 # if this timeout gets too small, then split the tests
+    timeout-minutes: 35 # if this timeout gets too small, then split the tests
     name: "[unit] webserver 02"
     runs-on: ${{ matrix.os }}
     strategy:

.vscode/settings.template.json

@@ -1,6 +1,7 @@
 // This is a template. Clone and replace extension ".template.json" by ".json"
 {
   "autoDocstring.docstringFormat": "pep257",
+
   "editor.tabSize": 2,
   "editor.insertSpaces": true,
   "editor.detectIndentation": false,
@@ -33,8 +34,6 @@
   "python.analysis.typeCheckingMode": "basic",
   "python.analysis.extraPaths": [
     "./packages/aws-library/src",
-    "./packages/common-library/src",
-    "./packages/dask-task-models-library/src",
     "./packages/models-library/src",
     "./packages/postgres-database/src",
     "./packages/postgres-database/tests",

packages/service-library/src/servicelib/logging_errors.py

@@ -27,12 +27,22 @@ def create_troubleshotting_log_message(
         error_context -- Additional context surrounding the exception, such as environment variables or function-specific data. This can be derived from exc.error_context() (relevant when using the OsparcErrorMixin)
         tip -- Helpful suggestions or possible solutions explaining why the error may have occurred and how it could potentially be resolved
     """
+
+    def _collect_causes(exc: BaseException) -> str:
+        causes = []
+        current = exc.__cause__
+        while current is not None:
+            causes.append(f"[{type(current).__name__}]'{current}'")
+            current = getattr(current, "__cause__", None)
+        return " <- ".join(causes)
+
     debug_data = json_dumps(
         {
             "exception_type": f"{type(error)}",
             "exception_details": f"{error}",
             "error_code": error_code,
             "context": error_context,
+            "causes": _collect_causes(error),
             "tip": tip,
         },
         default=representation_encoder,

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

@@ -16,6 +16,7 @@
     RegisteredFunctionJob,
     RegisteredFunctionJobCollection,
 )
+from models_library.functions import FunctionUserAccessRights
 from models_library.products import ProductName
 from models_library.rabbitmq_basic_types import RPCMethodName
 from models_library.rest_pagination import PageMetaInfoLimitOffset
@@ -388,3 +389,21 @@ async def delete_function_job_collection(
         product_name=product_name,
     )
     assert result is None  # nosec
+
+
+@log_decorator(_logger, level=logging.DEBUG)
+async def get_function_user_permissions(
+    rabbitmq_rpc_client: RabbitMQRPCClient,
+    *,
+    user_id: UserID,
+    product_name: ProductName,
+    function_id: FunctionID,
+) -> FunctionUserAccessRights:
+    result = await rabbitmq_rpc_client.request(
+        WEBSERVER_RPC_NAMESPACE,
+        TypeAdapter(RPCMethodName).validate_python("get_function_user_permissions"),
+        function_id=function_id,
+        user_id=user_id,
+        product_name=product_name,
+    )
+    return TypeAdapter(FunctionUserAccessRights).validate_python(result)

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

@@ -21,8 +21,11 @@
     RegisteredFunctionJobCollection,
     SolverFunctionJob,
 )
+from models_library.functions import FunctionUserAccessRights
 from models_library.functions_errors import (
+    FunctionExecuteAccessDeniedError,
     FunctionInputsValidationError,
+    FunctionReadAccessDeniedError,
     UnsupportedFunctionClassError,
 )
 from models_library.products import ProductName
@@ -371,6 +374,22 @@ async def run_function(  # noqa: PLR0913
     job_service: Annotated[JobService, Depends(get_job_service)],
 ) -> RegisteredFunctionJob:
 
+    user_permissions: FunctionUserAccessRights = (
+        await wb_api_rpc.get_function_user_permissions(
+            function_id=function_id, user_id=user_id, product_name=product_name
+        )
+    )
+    if not user_permissions.read:
+        raise FunctionReadAccessDeniedError(
+            user_id=user_id,
+            function_id=function_id,
+        )
+    if not user_permissions.execute:
+        raise FunctionExecuteAccessDeniedError(
+            user_id=user_id,
+            function_id=function_id,
+        )
+
     from .function_jobs_routes import function_job_status
 
     to_run_function = await wb_api_rpc.get_function(

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

@@ -23,6 +23,7 @@
     RegisteredFunctionJobCollection,
 )
 from models_library.api_schemas_webserver.licensed_items import LicensedItemRpcGetPage
+from models_library.functions import FunctionUserAccessRights
 from models_library.licenses import LicensedItemID
 from models_library.products import ProductName
 from models_library.projects import ProjectID
@@ -526,6 +527,20 @@ async def delete_function_job_collection(
             function_job_collection_id=function_job_collection_id,
         )
 
+    async def get_function_user_permissions(
+        self,
+        *,
+        user_id: UserID,
+        product_name: ProductName,
+        function_id: FunctionID,
+    ) -> FunctionUserAccessRights:
+        return await functions_rpc_interface.get_function_user_permissions(
+            self._client,
+            user_id=user_id,
+            product_name=product_name,
+            function_id=function_id,
+        )
+
 
 def setup(app: FastAPI, rabbitmq_rmp_client: RabbitMQRPCClient):
     wb_api_rpc_client = WbApiRpcClient(_client=rabbitmq_rmp_client)

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

@@ -7,6 +7,7 @@
 from uuid import uuid4
 
 import httpx
+import respx
 from httpx import AsyncClient
 from models_library.api_schemas_webserver.functions import (
     FunctionJobCollection,
@@ -16,11 +17,14 @@
     RegisteredProjectFunction,
     RegisteredProjectFunctionJob,
 )
+from models_library.functions import FunctionUserAccessRights
 from models_library.functions_errors import (
     FunctionIDNotFoundError,
     FunctionReadAccessDeniedError,
 )
 from models_library.rest_pagination import PageMetaInfoLimitOffset
+from models_library.users import UserID
+from pytest_mock import MockType
 from servicelib.aiohttp import status
 from simcore_service_api_server._meta import API_VTAG
 
@@ -580,3 +584,34 @@ async def test_list_function_job_collections_with_function_filter(
         RegisteredFunctionJobCollection.model_validate(data["items"][0])
         == mock_registered_function_job_collection
     )
+
+
+async def test_run_function_not_allowed(
+    client: AsyncClient,
+    mock_handler_in_functions_rpc_interface: Callable[[str, Any], None],
+    mock_registered_function: RegisteredProjectFunction,
+    auth: httpx.BasicAuth,
+    user_id: UserID,
+    mocked_webserver_rest_api_base: respx.MockRouter,
+    mocked_webserver_rpc_api: dict[str, MockType],
+) -> None:
+    """Test that running a function is not allowed."""
+    mock_handler_in_functions_rpc_interface(
+        "get_function_user_permissions",
+        FunctionUserAccessRights(
+            user_id=user_id,
+            execute=False,
+            read=True,
+            write=True,
+        ),
+    )
+
+    response = await client.post(
+        f"{API_VTAG}/functions/{mock_registered_function.uid}:run",
+        json={},
+        auth=auth,
+    )
+    assert response.status_code == status.HTTP_403_FORBIDDEN
+    assert response.json()["errors"][0] == (
+        f"Function {mock_registered_function.uid} execute access denied for user {user_id}"
+    )

services/dask-sidecar/src/simcore_service_dask_sidecar/computational_sidecar/core.py

@@ -7,7 +7,7 @@
 from pathlib import Path
 from pprint import pformat
 from types import TracebackType
-from typing import cast
+from typing import Final, cast
 from uuid import uuid4
 
 from aiodocker import Docker
@@ -32,7 +32,6 @@
 from ..utils.dask import TaskPublisher
 from ..utils.files import (
     check_need_unzipping,
-    log_partial_file_content,
     pull_file_from_remote,
     push_file_to_remote,
 )
@@ -49,8 +48,8 @@
 from .task_shared_volume import TaskSharedVolumes
 
 _logger = logging.getLogger(__name__)
-_CONTAINER_WAIT_TIME_SECS = 2
-_MAX_LOGGED_FILE_CHARS = 40
+CONTAINER_WAIT_TIME_SECS = 2
+_TASK_PROCESSING_PROGRESS_WEIGHT: Final[float] = 0.99
 
 
 @dataclass(kw_only=True, frozen=True, slots=True)
@@ -148,17 +147,11 @@ async def _retrieve_output_data(
             upload_tasks = []
             for output_params in output_data.values():
                 if isinstance(output_params, FileUrl):
-                    assert (
+                    assert (  # nosec
                         output_params.file_mapping
                     ), f"{output_params.model_dump_json(indent=1)} expected resolved in TaskOutputData.from_task_output"
 
                     src_path = task_volumes.outputs_folder / output_params.file_mapping
-                    await log_partial_file_content(
-                        src_path,
-                        logger=_logger,
-                        log_level=logging.DEBUG,
-                        max_chars=_MAX_LOGGED_FILE_CHARS,
-                    )
                     upload_tasks.append(
                         push_file_to_remote(
                             src_path,
@@ -274,7 +267,7 @@ async def run(self, command: list[str]) -> TaskOutputData:
                 )
                 # wait until the container finished, either success or fail or timeout
                 while (container_data := await container.show())["State"]["Running"]:
-                    await asyncio.sleep(_CONTAINER_WAIT_TIME_SECS)
+                    await asyncio.sleep(CONTAINER_WAIT_TIME_SECS)
                 if container_data["State"]["ExitCode"] > os.EX_OK:
                     raise ServiceRuntimeError(
                         service_key=self.task_parameters.image,

services/dask-sidecar/src/simcore_service_dask_sidecar/utils/files.py

@@ -20,7 +20,7 @@
 from settings_library.s3 import S3Settings
 from yarl import URL
 
-_logger = logging.getLogger(__name__)
+logger = logging.getLogger(__name__)
 
 HTTP_FILE_SYSTEM_SCHEMES: Final = ["http", "https"]
 S3_FILE_SYSTEM_SCHEMES: Final = ["s3", "s3a"]
@@ -208,7 +208,7 @@ async def pull_file_from_remote(
             await log_publishing_cb(
                 f"Uncompressing '{download_dst_path.name}'...", logging.INFO
             )
-            _logger.debug(
+            logger.debug(
                 "%s is a zip file and will be now uncompressed", download_dst_path
             )
             with repro_zipfile.ReproducibleZipFile(download_dst_path, "r") as zip_obj:
@@ -258,7 +258,7 @@ async def _push_file_to_remote(
     log_publishing_cb: LogPublishingCB,
     s3_settings: S3Settings | None,
 ) -> None:
-    _logger.debug("Uploading %s to %s...", file_to_upload, dst_url)
+    logger.debug("Uploading %s to %s...", file_to_upload, dst_url)
     assert dst_url.path  # nosec
 
     storage_kwargs: S3FsSettingsDict | dict[str, Any] = {}
@@ -306,7 +306,7 @@ async def push_file_to_remote(
                 await asyncio.get_event_loop().run_in_executor(
                     None, zfp.write, src_path, src_path.name
                 )
-            _logger.debug("%s created.", archive_file_path)
+            logger.debug("%s created.", archive_file_path)
             assert archive_file_path.exists()  # nosec
             file_to_upload = archive_file_path
             await log_publishing_cb(
@@ -319,7 +319,7 @@ async def push_file_to_remote(
         )
 
         if dst_url.scheme in HTTP_FILE_SYSTEM_SCHEMES:
-            _logger.debug("destination is a http presigned link")
+            logger.debug("destination is a http presigned link")
             await _push_file_to_http_link(file_to_upload, dst_url, log_publishing_cb)
         else:
             await _push_file_to_remote(
@@ -330,22 +330,3 @@ async def push_file_to_remote(
         f"Upload of '{src_path.name}' to '{dst_url.path.strip('/')}' complete",
         logging.INFO,
     )
-
-
-async def log_partial_file_content(
-    src_path: Path, *, logger: logging.Logger, log_level: int, max_chars: int
-) -> None:
-    if max_chars < 0:
-        msg = "max_chars must be non-negative"
-        raise ValueError(msg)
-    if max_chars == 0:
-        return
-    if not src_path.exists():
-        logger.log(log_level, "file does not exist: %s", src_path)
-        return
-    async with aiofiles.open(src_path, encoding="utf-8") as f:
-        content = await f.read(max_chars + 1)
-    if len(content) > max_chars:
-        logger.log(log_level, "file content (truncated): %s...", content[:max_chars])
-    else:
-        logger.log(log_level, "file content: %s", content)

services/dask-sidecar/tests/unit/test_utils_files.py

@@ -4,7 +4,6 @@
 
 import asyncio
 import hashlib
-import logging
 import mimetypes
 import zipfile
 from collections.abc import AsyncIterable
@@ -22,7 +21,6 @@
 from settings_library.s3 import S3Settings
 from simcore_service_dask_sidecar.utils.files import (
     _s3fs_settings_from_s3_settings,
-    log_partial_file_content,
     pull_file_from_remote,
     push_file_to_remote,
 )
@@ -513,48 +511,3 @@ async def test_push_file_to_remote_creates_reproducible_zip_archive(
     assert dst_path2.exists()
 
     assert _compute_hash(dst_path1) == _compute_hash(dst_path2)
-
-
-async def test_log_partial_file_content(
-    tmp_path: Path, caplog: pytest.LogCaptureFixture
-):
-    # Create a file with known content
-    file_content = "abcdefghijklmnopqrstuvwxyz0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ"
-    file_path = tmp_path / "testfile.txt"
-    file_path.write_text(file_content)
-    logger = logging.getLogger("pytest.utils.files")
-
-    # Case 1: file longer than max_chars
-    with caplog.at_level(logging.DEBUG, logger=logger.name):
-        await log_partial_file_content(
-            file_path, logger=logger, log_level=logging.DEBUG, max_chars=10
-        )
-    assert any(
-        "file content (truncated): abcdefghij..." in record.getMessage()
-        for record in caplog.records
-    )
-
-    # Case 2: file shorter than max_chars
-    caplog.clear()
-    short_content = "short"
-    short_file = tmp_path / "short.txt"
-    short_file.write_text(short_content)
-    with caplog.at_level(logging.DEBUG, logger=logger.name):
-        await log_partial_file_content(
-            short_file, logger=logger, log_level=logging.DEBUG, max_chars=10
-        )
-    assert any(
-        "file content: short" in record.getMessage() for record in caplog.records
-    )
-
-    # Case 3: file does not exist
-    caplog.clear()
-    non_existent = tmp_path / "doesnotexist.txt"
-    with caplog.at_level(logging.DEBUG, logger=logger.name):
-        await log_partial_file_content(
-            non_existent, logger=logger, log_level=logging.DEBUG, max_chars=10
-        )
-    assert any(
-        f"file does not exist: {non_existent}" in record.getMessage()
-        for record in caplog.records
-    )