Merge branch 'master' into maintenance/improve-error-handling

Author: matusdrobuliak66

packages/models-library/src/models_library/api_schemas_dynamic_sidecar/ports.py

@@ -0,0 +1,35 @@
+from enum import auto
+
+from models_library.projects import ProjectID
+from models_library.projects_nodes_io import NodeID
+from models_library.services_types import ServicePortKey
+from models_library.utils.enums import StrAutoEnum
+from pydantic import BaseModel
+
+
+class OutputStatus(StrAutoEnum):
+    UPLOAD_STARTED = auto()
+    UPLOAD_WAS_ABORTED = auto()
+    UPLOAD_FINISHED_SUCCESSFULLY = auto()
+    UPLOAD_FINISHED_WITH_ERRROR = auto()
+
+
+class InputStatus(StrAutoEnum):
+    DOWNLOAD_STARTED = auto()
+    DOWNLOAD_WAS_ABORTED = auto()
+    DOWNLOAD_FINISHED_SUCCESSFULLY = auto()
+    DOWNLOAD_FINISHED_WITH_ERRROR = auto()
+
+
+class _PortStatusCommon(BaseModel):
+    project_id: ProjectID
+    node_id: NodeID
+    port_key: ServicePortKey
+
+
+class OutputPortStatus(_PortStatusCommon):
+    status: OutputStatus
+
+
+class InputPortSatus(_PortStatusCommon):
+    status: InputStatus

packages/models-library/src/models_library/api_schemas_dynamic_sidecar/socketio.py

@@ -1,3 +1,5 @@
 from typing import Final
 
 SOCKET_IO_SERVICE_DISK_USAGE_EVENT: Final[str] = "serviceDiskUsage"
+SOCKET_IO_STATE_OUTPUT_PORTS_EVENT: Final[str] = "stateOutputPorts"
+SOCKET_IO_STATE_INPUT_PORTS_EVENT: Final[str] = "stateInputPorts"

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

@@ -0,0 +1,34 @@
+from pydantic import SecretStr
+
+
+def _mask(value):
+    """
+    Mask the password, showing only the first and last characters
+    or *** if very short passwords
+    """
+    if len(value) > 2:
+        masked_value = value[0] + "*" * (len(value) - 2) + value[-1]
+    else:
+        # In case of very short passwords
+        masked_value = "*" * len(value)
+    return masked_value
+
+
+def _hash(value):
+    """Uses hash number to mask the password"""
+    return f"hash:{hash(value)}"
+
+
+class Secret4TestsStr(SecretStr):
+    """Prints a hint of the secret
+    TIP: Can be handy for testing
+    """
+
+    def _display(self) -> str | bytes:
+        # SEE overrides _SecretBase._display
+        value = self.get_secret_value()
+        return _mask(value) if value else ""
+
+
+assert str(Secret4TestsStr("123456890")) == "1*******0"
+assert "1*******0" in repr(Secret4TestsStr("123456890"))

packages/service-integration/requirements/_base.in

@@ -13,3 +13,4 @@ jsonschema # pytest-plugin
 pytest # pytest-plugin
 pyyaml
 typer[all]
+yarl

packages/service-integration/requirements/_base.txt

@@ -35,6 +35,7 @@ idna==3.7
     # via
     #   email-validator
     #   requests
+    #   yarl
 iniconfig==2.0.0
     # via pytest
 jinja2==3.1.4
@@ -57,6 +58,8 @@ markupsafe==2.1.5
     # via jinja2
 mdurl==0.1.2
     # via markdown-it-py
+multidict==6.1.0
+    # via yarl
 orjson==3.10.7
     # via
     #   -c requirements/../../../packages/models-library/requirements/../../../requirements/constraints.txt
@@ -121,3 +124,5 @@ urllib3==2.2.2
     #   -c requirements/../../../requirements/constraints.txt
     #   docker
     #   requests
+yarl==1.12.1
+    # via -r requirements/_base.in

packages/service-integration/src/service_integration/cli/_compose_spec.py

@@ -8,6 +8,7 @@
 import yaml
 from models_library.utils.labels_annotations import to_labels
 from rich.console import Console
+from yarl import URL
 
 from ..compose_spec_model import ComposeSpecification
 from ..errors import UndefinedOciImageSpecError
@@ -34,6 +35,13 @@ def _run_git(*args) -> str:
     ).stdout.strip()
 
 
+def _strip_credentials(url: str) -> str:
+    if (yarl_url := URL(url)) and yarl_url.is_absolute():
+        stripped_url = URL(url).with_user(None).with_password(None)
+        return f"{stripped_url}"
+    return url
+
+
 def _run_git_or_empty_string(*args) -> str:
     try:
         return _run_git(*args)
@@ -118,8 +126,8 @@ def create_docker_compose_image_spec(
     extra_labels[f"{LS_LABEL_PREFIX}.vcs-ref"] = _run_git_or_empty_string(
         "rev-parse", "HEAD"
     )
-    extra_labels[f"{LS_LABEL_PREFIX}.vcs-url"] = _run_git_or_empty_string(
-        "config", "--get", "remote.origin.url"
+    extra_labels[f"{LS_LABEL_PREFIX}.vcs-url"] = _strip_credentials(
+        _run_git_or_empty_string("config", "--get", "remote.origin.url")
     )
 
     return create_image_spec(

packages/service-integration/tests/test_cli__compose_spec.py

@@ -0,0 +1,28 @@
+import pytest
+from service_integration.cli._compose_spec import _strip_credentials
+
+
+@pytest.mark.parametrize(
+    "url, expected_url",
+    [
+        (
+            "schema.veshttps://user:[email protected]/some/repo.git",
+            "schema.veshttps://example.com/some/repo.git",
+        ),
+        (
+            "https://user:[email protected]/some/repo.git",
+            "https://example.com/some/repo.git",
+        ),
+        (
+            "ssh://user:[email protected]/some/repo.git",
+            "ssh://example.com/some/repo.git",
+        ),
+        (
+            "[email protected]:some/repo.git",
+            "[email protected]:some/repo.git",
+        ),
+        ("any_str", "any_str"),
+    ],
+)
+def test__strip_credentials(url: str, expected_url: str):
+    assert _strip_credentials(url) == expected_url

packages/simcore-sdk/src/simcore_sdk/node_ports_v2/nodeports_v2.py

@@ -1,4 +1,6 @@
 import logging
+from abc import ABC, abstractmethod
+from asyncio import CancelledError
 from collections.abc import Callable, Coroutine
 from pathlib import Path
 from typing import Any
@@ -27,6 +29,20 @@
 log = logging.getLogger(__name__)
 
 
+class OutputsCallbacks(ABC):
+    @abstractmethod
+    async def aborted(self, key: ServicePortKey) -> None:
+        pass
+
+    @abstractmethod
+    async def finished_succesfully(self, key: ServicePortKey) -> None:
+        pass
+
+    @abstractmethod
+    async def finished_with_error(self, key: ServicePortKey) -> None:
+        pass
+
+
 class Nodeports(BaseModel):
     """
     Represents a node in a project and all its input/output ports
@@ -148,6 +164,7 @@ async def set_multiple(
         ],
         *,
         progress_bar: ProgressBarData,
+        outputs_callbacks: OutputsCallbacks | None,
     ) -> None:
         """
         Sets the provided values to the respective input or output ports
@@ -156,26 +173,44 @@ async def set_multiple(
 
         raises ValidationError
         """
+
+        async def _set_with_notifications(
+            port_key: ServicePortKey,
+            value: ItemConcreteValue | None,
+            set_kwargs: SetKWargs | None,
+            sub_progress: ProgressBarData,
+        ) -> None:
+            try:
+                # pylint: disable=protected-access
+                await self.internal_outputs[port_key]._set(  # noqa: SLF001
+                    value, set_kwargs=set_kwargs, progress_bar=sub_progress
+                )
+                if outputs_callbacks:
+                    await outputs_callbacks.finished_succesfully(port_key)
+            except UnboundPortError:
+                # not available try inputs
+                # if this fails it will raise another exception
+                # pylint: disable=protected-access
+                await self.internal_inputs[port_key]._set(  # noqa: SLF001
+                    value, set_kwargs=set_kwargs, progress_bar=sub_progress
+                )
+            except CancelledError:
+                if outputs_callbacks:
+                    await outputs_callbacks.aborted(port_key)
+                raise
+            except Exception:
+                if outputs_callbacks:
+                    await outputs_callbacks.finished_with_error(port_key)
+                raise
+
         tasks = []
         async with progress_bar.sub_progress(
             steps=len(port_values.items()), description=IDStr("set multiple")
         ) as sub_progress:
             for port_key, (value, set_kwargs) in port_values.items():
-                # pylint: disable=protected-access
-                try:
-                    tasks.append(
-                        self.internal_outputs[port_key]._set(
-                            value, set_kwargs=set_kwargs, progress_bar=sub_progress
-                        )
-                    )
-                except UnboundPortError:
-                    # not available try inputs
-                    # if this fails it will raise another exception
-                    tasks.append(
-                        self.internal_inputs[port_key]._set(
-                            value, set_kwargs=set_kwargs, progress_bar=sub_progress
-                        )
-                    )
+                tasks.append(
+                    _set_with_notifications(port_key, value, set_kwargs, sub_progress)
+                )
 
             results = await logged_gather(*tasks)
             await self.save_to_db_cb(self)

packages/simcore-sdk/tests/integration/test_node_ports_v2_nodeports2.py

@@ -13,6 +13,7 @@
 from collections.abc import Awaitable, Callable, Iterable
 from pathlib import Path
 from typing import Any
+from unittest.mock import AsyncMock
 from uuid import uuid4
 
 import np_helpers
@@ -28,13 +29,14 @@
     SimcoreS3FileID,
 )
 from models_library.services_types import ServicePortKey
+from pytest_mock import MockerFixture
 from servicelib.progress_bar import ProgressBarData
 from settings_library.r_clone import RCloneSettings
 from simcore_sdk import node_ports_v2
 from simcore_sdk.node_ports_common.exceptions import UnboundPortError
 from simcore_sdk.node_ports_v2 import exceptions
 from simcore_sdk.node_ports_v2.links import ItemConcreteValue, PortLink
-from simcore_sdk.node_ports_v2.nodeports_v2 import Nodeports
+from simcore_sdk.node_ports_v2.nodeports_v2 import Nodeports, OutputsCallbacks
 from simcore_sdk.node_ports_v2.port import Port
 from utils_port_v2 import CONSTANT_UUID
 
@@ -749,6 +751,34 @@ async def _upload_create_task(item_key: str) -> None:
     )
 
 
+class _Callbacks(OutputsCallbacks):
+    async def aborted(self, key: ServicePortKey) -> None:
+        pass
+
+    async def finished_succesfully(self, key: ServicePortKey) -> None:
+        pass
+
+    async def finished_with_error(self, key: ServicePortKey) -> None:
+        pass
+
+
+@pytest.fixture
+async def output_callbacks() -> _Callbacks:
+    return _Callbacks()
+
+
+@pytest.fixture
+async def spy_outputs_callbaks(
+    mocker: MockerFixture, output_callbacks: _Callbacks
+) -> dict[str, AsyncMock]:
+    return {
+        "aborted": mocker.spy(output_callbacks, "aborted"),
+        "finished_succesfully": mocker.spy(output_callbacks, "finished_succesfully"),
+        "finished_with_error": mocker.spy(output_callbacks, "finished_with_error"),
+    }
+
+
+@pytest.mark.parametrize("use_output_callbacks", [True, False])
 async def test_batch_update_inputs_outputs(
     user_id: int,
     project_id: str,
@@ -757,7 +787,12 @@ async def test_batch_update_inputs_outputs(
     port_count: int,
     option_r_clone_settings: RCloneSettings | None,
     faker: Faker,
+    output_callbacks: _Callbacks,
+    spy_outputs_callbaks: dict[str, AsyncMock],
+    use_output_callbacks: bool,
 ) -> None:
+    callbacks = output_callbacks if use_output_callbacks else None
+
     outputs = [(f"value_out_{i}", "integer", None) for i in range(port_count)]
     inputs = [(f"value_in_{i}", "integer", None) for i in range(port_count)]
     config_dict, _, _ = create_special_configuration(inputs=inputs, outputs=outputs)
@@ -771,12 +806,14 @@ async def test_batch_update_inputs_outputs(
     await check_config_valid(PORTS, config_dict)
 
     async with ProgressBarData(num_steps=2, description=faker.pystr()) as progress_bar:
+        port_values = (await PORTS.outputs).values()
         await PORTS.set_multiple(
-            {
-                ServicePortKey(port.key): (k, None)
-                for k, port in enumerate((await PORTS.outputs).values())
-            },
+            {ServicePortKey(port.key): (k, None) for k, port in enumerate(port_values)},
             progress_bar=progress_bar,
+            outputs_callbacks=callbacks,
+        )
+        assert len(spy_outputs_callbaks["finished_succesfully"].call_args_list) == (
+            len(port_values) if use_output_callbacks else 0
         )
         # pylint: disable=protected-access
         assert progress_bar._current_steps == pytest.approx(1)  # noqa: SLF001
@@ -786,6 +823,11 @@ async def test_batch_update_inputs_outputs(
                 for k, port in enumerate((await PORTS.inputs).values(), start=1000)
             },
             progress_bar=progress_bar,
+            outputs_callbacks=callbacks,
+        )
+        # inputs do not trigger callbacks
+        assert len(spy_outputs_callbaks["finished_succesfully"].call_args_list) == (
+            len(port_values) if use_output_callbacks else 0
         )
         assert progress_bar._current_steps == pytest.approx(2)  # noqa: SLF001
 
@@ -807,4 +849,11 @@ async def test_batch_update_inputs_outputs(
             await PORTS.set_multiple(
                 {ServicePortKey("missing_key_in_both"): (123132, None)},
                 progress_bar=progress_bar,
+                outputs_callbacks=callbacks,
             )
+
+    assert len(spy_outputs_callbaks["finished_succesfully"].call_args_list) == (
+        len(port_values) if use_output_callbacks else 0
+    )
+    assert len(spy_outputs_callbaks["aborted"].call_args_list) == 0
+    assert len(spy_outputs_callbaks["finished_with_error"].call_args_list) == 0

packages/simcore-sdk/tests/unit/test_node_ports_v2_nodeports_v2.py

@@ -5,6 +5,7 @@
 
 from pathlib import Path
 from typing import Any, Callable
+from unittest.mock import AsyncMock
 
 import pytest
 from faker import Faker
@@ -138,6 +139,7 @@ async def mock_node_port_creator_cb(*args, **kwargs):
                 + list(original_outputs.values())
             },
             progress_bar=progress_bar,
+            outputs_callbacks=AsyncMock(),
         )
     assert progress_bar._current_steps == pytest.approx(1)  # noqa: SLF001