Merge remote-tracking branch 'upstream/master' into pr-osparc-add-dy-sidecar-lrt-rabbit

Author: Andrei Neagu

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

@@ -3,7 +3,7 @@
 """
 
 from enum import Enum, unique
-from typing import Annotated, Self, TypeAlias
+from typing import Annotated, Final, Self, TypeAlias
 
 from pydantic import (
     BaseModel,
@@ -65,6 +65,13 @@ def is_running(self) -> bool:
         return self in self.list_running_states()
 
 
+RUNNING_STATE_COMPLETED_STATES: Final[tuple[RunningState, ...]] = (
+    RunningState.ABORTED,
+    RunningState.FAILED,
+    RunningState.SUCCESS,
+)
+
+
 @unique
 class DataState(str, Enum):
     UP_TO_DATE = "UPTODATE"

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

@@ -325,4 +325,4 @@ class ComputationalPipelineStatusMessage(RabbitMessageBase, ProjectMessageBase):
     run_result: RunningState
 
     def routing_key(self) -> str | None:
-        return f"{self.project_id}"
+        return f"{self.project_id}.all_nodes"

packages/pytest-simcore/src/pytest_simcore/docker_registry.py

@@ -15,6 +15,7 @@
 import docker
 import jsonschema
 import pytest
+import pytest_asyncio
 import tenacity
 from pytest_simcore.helpers.logging_tools import log_context
 from pytest_simcore.helpers.typing_env import EnvVarsDict
@@ -146,7 +147,7 @@ def wait_till_registry_is_responsive(url: str) -> bool:
 # ********************************************************* Services ***************************************
 
 
-def _pull_push_service(
+async def _pull_push_service(
     pull_key: str,
     tag: str,
     new_registry: str,
@@ -213,11 +214,13 @@ def _pull_push_service(
     assert image.tag(new_image_tag)
 
     # push the image to the new location
-    with log_context(
-        logging.INFO,
-        msg=f"Pushing {pull_key}:{tag}  -> {new_image_tag} ...",
-    ):
-        client.images.push(new_image_tag)
+    async with aiodocker.Docker() as client:
+        await client.images.push(new_image_tag)
+    # with log_context(
+    #     logging.INFO,
+    #     msg=f"Pushing {pull_key}:{tag}  -> {new_image_tag} ...",
+    # ):
+    #     client.images.push(new_image_tag)
 
     # return image io.simcore.* labels
     image_labels = dict(image.labels)
@@ -230,10 +233,10 @@ def _pull_push_service(
     }
 
 
-@pytest.fixture(scope="session")
+@pytest_asyncio.fixture(scope="session", loop_scope="session")
 def docker_registry_image_injector(
     docker_registry: str, node_meta_schema: dict
-) -> Callable[..., dict[str, Any]]:
+) -> Callable[[str, str, str | None], Awaitable[dict[str, Any]]]:
     def inject_image(
         source_image_repo: str, source_image_tag: str, owner_email: str | None = None
     ):
@@ -249,82 +252,86 @@ def inject_image(
 
 
 @pytest.fixture
-def osparc_service(
+async def osparc_service(
     docker_registry: str, node_meta_schema: dict, service_repo: str, service_tag: str
 ) -> dict[str, Any]:
     """pulls the service from service_repo:service_tag and pushes to docker_registry using the oSparc node meta schema
     NOTE: 'service_repo' and 'service_tag' defined as parametrization
     """
-    return _pull_push_service(
+    return await _pull_push_service(
         service_repo, service_tag, docker_registry, node_meta_schema
     )
 
 
-@pytest.fixture(scope="session")
-def sleeper_service(docker_registry: str, node_meta_schema: dict) -> dict[str, Any]:
+@pytest_asyncio.fixture(scope="session", loop_scope="session")
+async def sleeper_service(
+    docker_registry: str, node_meta_schema: dict
+) -> dict[str, Any]:
     """Adds a itisfoundation/sleeper in docker registry"""
-    return _pull_push_service(
+    return await _pull_push_service(
         "itisfoundation/sleeper", "1.0.0", docker_registry, node_meta_schema
     )
 
 
-@pytest.fixture(scope="session")
-def jupyter_service(docker_registry: str, node_meta_schema: dict) -> dict[str, Any]:
+@pytest_asyncio.fixture(scope="session", loop_scope="session")
+async def jupyter_service(
+    docker_registry: str, node_meta_schema: dict
+) -> dict[str, Any]:
     """Adds a itisfoundation/jupyter-base-notebook in docker registry"""
-    return _pull_push_service(
+    return await _pull_push_service(
         "itisfoundation/jupyter-base-notebook",
         "2.13.0",
         docker_registry,
         node_meta_schema,
     )
 
 
-@pytest.fixture(scope="session", params=["2.0.7"])
+@pytest_asyncio.fixture(scope="session", loop_scope="session", params=["2.0.7"])
 def dy_static_file_server_version(request: pytest.FixtureRequest):
     return request.param
 
 
-@pytest.fixture(scope="session")
-def dy_static_file_server_service(
+@pytest_asyncio.fixture(scope="session", loop_scope="session")
+async def dy_static_file_server_service(
     docker_registry: str, node_meta_schema: dict, dy_static_file_server_version: str
 ) -> dict[str, Any]:
     """
     Adds the below service in docker registry
     itisfoundation/dy-static-file-server
     """
-    return _pull_push_service(
+    return await _pull_push_service(
         "itisfoundation/dy-static-file-server",
         dy_static_file_server_version,
         docker_registry,
         node_meta_schema,
     )
 
 
-@pytest.fixture(scope="session")
-def dy_static_file_server_dynamic_sidecar_service(
+@pytest_asyncio.fixture(scope="session", loop_scope="session")
+async def dy_static_file_server_dynamic_sidecar_service(
     docker_registry: str, node_meta_schema: dict, dy_static_file_server_version: str
 ) -> dict[str, Any]:
     """
     Adds the below service in docker registry
     itisfoundation/dy-static-file-server-dynamic-sidecar
     """
-    return _pull_push_service(
+    return await _pull_push_service(
         "itisfoundation/dy-static-file-server-dynamic-sidecar",
         dy_static_file_server_version,
         docker_registry,
         node_meta_schema,
     )
 
 
-@pytest.fixture(scope="session")
-def dy_static_file_server_dynamic_sidecar_compose_spec_service(
+@pytest_asyncio.fixture(scope="session", loop_scope="session")
+async def dy_static_file_server_dynamic_sidecar_compose_spec_service(
     docker_registry: str, node_meta_schema: dict, dy_static_file_server_version: str
 ) -> dict[str, Any]:
     """
     Adds the below service in docker registry
     itisfoundation/dy-static-file-server-dynamic-sidecar-compose-spec
     """
-    return _pull_push_service(
+    return await _pull_push_service(
         "itisfoundation/dy-static-file-server-dynamic-sidecar-compose-spec",
         dy_static_file_server_version,
         docker_registry,

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

@@ -302,6 +302,29 @@ def __call__(self, message: str) -> bool:
             return False
 
 
+@dataclass
+class SocketIOWaitNodeForOutputs:
+    logger: logging.Logger
+    expected_number_of_outputs: int
+    node_id: str
+
+    def __call__(self, message: str) -> bool:
+        if message.startswith(SOCKETIO_MESSAGE_PREFIX):
+            decoded_message = decode_socketio_42_message(message)
+            if decoded_message.name == _OSparcMessages.NODE_UPDATED:
+                assert "data" in decoded_message.obj
+                assert "node_id" in decoded_message.obj
+                if decoded_message.obj["node_id"] == self.node_id:
+                    assert "outputs" in decoded_message.obj["data"]
+
+                    return (
+                        len(decoded_message.obj["data"]["outputs"])
+                        == self.expected_number_of_outputs
+                    )
+
+        return False
+
+
 @dataclass
 class SocketIOOsparcMessagePrinter:
     include_logger_messages: bool = False

services/agent/src/simcore_service_agent/services/backup.py

@@ -1,6 +1,7 @@
 import asyncio
+import json
 import logging
-import os
+import socket
 import tempfile
 from asyncio.streams import StreamReader
 from datetime import timedelta
@@ -9,6 +10,7 @@
 from typing import Final
 from uuid import uuid4
 
+import httpx
 from fastapi import FastAPI
 from servicelib.container_utils import run_command_in_container
 from settings_library.utils_r_clone import resolve_provider
@@ -112,8 +114,28 @@ def _log_expected_operation(
     _logger.log(log_level, formatted_message)
 
 
+def _get_self_container_ip() -> str:
+    return socket.gethostbyname(socket.gethostname())
+
+
+async def _get_self_container() -> str:
+    ip = _get_self_container_ip()
+
+    async with httpx.AsyncClient(
+        transport=httpx.AsyncHTTPTransport(uds="/var/run/docker.sock")
+    ) as client:
+        response = await client.get("http://localhost/containers/json")
+        for entry in response.json():
+            if ip in json.dumps(entry):
+                container_id: str = entry["Id"]
+                return container_id
+
+    msg = "Could not determine self container ID"
+    raise RuntimeError(msg)
+
+
 async def _ensure_permissions_on_source_dir(source_dir: Path) -> None:
-    self_container = os.environ["HOSTNAME"]
+    self_container = await _get_self_container()
     await run_command_in_container(
         self_container,
         command=f"chmod -R o+rX '{source_dir}'",

services/agent/tests/unit/test_services_backup.py

@@ -15,6 +15,8 @@
 from models_library.projects_nodes_io import NodeID
 from models_library.services_types import ServiceRunID
 from pydantic import NonNegativeInt
+from pytest_mock import MockerFixture
+from servicelib.container_utils import run_command_in_container
 from simcore_service_agent.core.settings import ApplicationSettings
 from simcore_service_agent.services.backup import backup_volume
 from simcore_service_agent.services.docker_utils import get_volume_details
@@ -42,7 +44,7 @@ def volume_content(tmpdir: Path) -> Path:
 @pytest.fixture
 async def mock_container_with_data(
     volume_content: Path, monkeypatch: pytest.MonkeyPatch
-) -> AsyncIterable[None]:
+) -> AsyncIterable[str]:
     async with aiodocker.Docker() as client:
         container = await client.containers.run(
             config={
@@ -56,7 +58,7 @@ async def mock_container_with_data(
         container_name = container_inspect["Name"][1:]
         monkeypatch.setenv("HOSTNAME", container_name)
 
-        yield None
+        yield container_inspect["Id"]
 
         await container.delete(force=True)
 
@@ -68,8 +70,24 @@ def downlaoded_from_s3(tmpdir: Path) -> Path:
     return path
 
 
+@pytest.fixture
+async def mock__get_self_container_ip(
+    mock_container_with_data: str,
+    mocker: MockerFixture,
+) -> None:
+    container_ip = await run_command_in_container(
+        mock_container_with_data, command="hostname -i"
+    )
+
+    mocker.patch(
+        "simcore_service_agent.services.backup._get_self_container_ip",
+        return_value=container_ip.strip(),
+    )
+
+
 async def test_backup_volume(
-    mock_container_with_data: None,
+    mock_container_with_data: str,
+    mock__get_self_container_ip: None,
     volume_content: Path,
     project_id: ProjectID,
     swarm_stack_name: str,

services/director-v2/src/simcore_service_director_v2/utils/computations.py

@@ -3,7 +3,7 @@
 from typing import Any
 
 import arrow
-from models_library.projects_state import RunningState
+from models_library.projects_state import RUNNING_STATE_COMPLETED_STATES, RunningState
 from models_library.services import ServiceKeyVersion
 from models_library.services_regex import SERVICE_KEY_RE
 from models_library.users import UserID
@@ -15,7 +15,7 @@
 
 _logger = logging.getLogger(__name__)
 
-_COMPLETED_STATES = (RunningState.ABORTED, RunningState.FAILED, RunningState.SUCCESS)
+
 _RUNNING_STATES = (RunningState.STARTED,)
 _TASK_TO_PIPELINE_CONVERSIONS = {
     # tasks are initially in NOT_STARTED state, then they transition to published
@@ -50,16 +50,16 @@
         RunningState.NOT_STARTED,
     ): RunningState.NOT_STARTED,
     # if there are only completed states with FAILED --> FAILED
-    (*_COMPLETED_STATES,): RunningState.FAILED,
+    (*RUNNING_STATE_COMPLETED_STATES,): RunningState.FAILED,
     # if there are only completed states with FAILED and not started ones --> NOT_STARTED
     (
-        *_COMPLETED_STATES,
+        *RUNNING_STATE_COMPLETED_STATES,
         RunningState.NOT_STARTED,
     ): RunningState.NOT_STARTED,
     # the generic case where we have a combination of completed states, running states,
     # or published/pending tasks, not_started is a started pipeline
     (
-        *_COMPLETED_STATES,
+        *RUNNING_STATE_COMPLETED_STATES,
         *_RUNNING_STATES,
         RunningState.PUBLISHED,
         RunningState.PENDING,

services/director-v2/tests/integration/02/test_dynamic_services_routes.py

@@ -276,11 +276,13 @@ async def _mocked_context_manger(*args, **kwargs) -> AsyncIterator[None]:
 async def key_version_expected(
     dy_static_file_server_dynamic_sidecar_service: dict,
     dy_static_file_server_service: dict,
-    docker_registry_image_injector: Callable,
+    docker_registry_image_injector: Callable[
+        [str, str, str | None], Awaitable[dict[str, Any]]
+    ],
 ) -> list[tuple[ServiceKeyVersion, bool]]:
     results: list[tuple[ServiceKeyVersion, bool]] = []
 
-    sleeper_service = docker_registry_image_injector(
+    sleeper_service = await docker_registry_image_injector(
         "itisfoundation/sleeper", "2.1.1", "[email protected]"
     )