added new test

sanderegg · sanderegg · commit cd482972a2a3 · 2025-06-30T16:40:04.000+02:00
diff --git a/packages/pytest-simcore/src/pytest_simcore/db_entries_mocks.py b/packages/pytest-simcore/src/pytest_simcore/db_entries_mocks.py
@@ -10,11 +10,13 @@
 import pytest
 import sqlalchemy as sa
 from faker import Faker
+from models_library.products import ProductName
 from models_library.projects import ProjectAtDB, ProjectID
 from models_library.projects_nodes_io import NodeID
 from simcore_postgres_database.models.comp_pipeline import StateType, comp_pipeline
 from simcore_postgres_database.models.comp_tasks import comp_tasks
 from simcore_postgres_database.models.projects import ProjectType, projects
+from simcore_postgres_database.models.projects_to_products import projects_to_products
 from simcore_postgres_database.models.users import UserRole, UserStatus, users
 from simcore_postgres_database.utils_projects_nodes import (
     ProjectNodeCreate,
@@ -64,7 +66,7 @@ def creator(**user_kwargs) -> dict[str, Any]:
 
 @pytest.fixture
 async def project(
-    sqlalchemy_async_engine: AsyncEngine, faker: Faker
+    sqlalchemy_async_engine: AsyncEngine, faker: Faker, product_name: ProductName
 ) -> AsyncIterator[Callable[..., Awaitable[ProjectAtDB]]]:
     created_project_ids: list[str] = []
 
@@ -112,6 +114,12 @@ async def creator(
                     for node_id in inserted_project.workbench
                 ],
             )
+            await con.execute(
+                projects_to_products.insert().values(
+                    project_uuid=f"{inserted_project.uuid}",
+                    product_name=product_name,
+                )
+            )
         print(f"--> created {inserted_project=}")
         created_project_ids.append(f"{inserted_project.uuid}")
         return inserted_project
diff --git a/services/web/server/src/simcore_service_webserver/db_listener/_db_comp_tasks_listening_task.py b/services/web/server/src/simcore_service_webserver/db_listener/_db_comp_tasks_listening_task.py
@@ -18,6 +18,7 @@
 from models_library.projects_state import RunningState
 from pydantic.types import PositiveInt
 from servicelib.background_task import periodic_task
+from servicelib.logging_utils import log_catch
 from simcore_postgres_database.models.comp_tasks import comp_tasks
 from simcore_postgres_database.webserver_models import DB_CHANNEL_NAME, projects
 from sqlalchemy.sql import select
@@ -73,56 +74,57 @@ async def _get_changed_comp_task_row(
 async def _handle_db_notification(
     app: web.Application, payload: CompTaskNotificationPayload, conn: SAConnection
 ) -> None:
-    try:
-        the_project_owner = await _get_project_owner(conn, payload.project_id)
-        changed_row = await _get_changed_comp_task_row(conn, payload.task_id)
-        if not changed_row:
+    with log_catch(_logger, reraise=False):
+        try:
+            the_project_owner = await _get_project_owner(conn, payload.project_id)
+            changed_row = await _get_changed_comp_task_row(conn, payload.task_id)
+            if not changed_row:
+                _logger.warning(
+                    "No comp_tasks row found for project_id=%s node_id=%s",
+                    payload.project_id,
+                    payload.node_id,
+                )
+                return
+
+            if any(f in payload.changes for f in ["outputs", "run_hash"]):
+                await update_node_outputs(
+                    app,
+                    the_project_owner,
+                    payload.project_id,
+                    payload.node_id,
+                    changed_row.outputs,
+                    changed_row.run_hash,
+                    node_errors=changed_row.errors,
+                    ui_changed_keys=None,
+                )
+
+            if "state" in payload.changes and (changed_row.state is not None):
+                await _update_project_state(
+                    app,
+                    the_project_owner,
+                    payload.project_id,
+                    payload.node_id,
+                    convert_state_from_db(changed_row.state),
+                    node_errors=changed_row.errors,
+                )
+
+        except exceptions.ProjectNotFoundError as exc:
             _logger.warning(
-                "No comp_tasks row found for project_id=%s node_id=%s",
-                payload.project_id,
-                payload.node_id,
+                "Project %s was not found and cannot be updated. Maybe was it deleted?",
+                exc.project_uuid,
             )
-            return
-
-        if any(f in payload.changes for f in ["outputs", "run_hash"]):
-            await update_node_outputs(
-                app,
-                the_project_owner,
-                payload.project_id,
-                payload.node_id,
-                changed_row.outputs,
-                changed_row.run_hash,
-                node_errors=changed_row.errors,
-                ui_changed_keys=None,
+        except exceptions.ProjectOwnerNotFoundError as exc:
+            _logger.warning(
+                "Project owner of project %s could not be found, is the project valid?",
+                exc.project_uuid,
             )
-
-        if "state" in payload.changes and (changed_row.state is not None):
-            await _update_project_state(
-                app,
-                the_project_owner,
-                payload.project_id,
-                payload.node_id,
-                convert_state_from_db(changed_row.state),
-                node_errors=changed_row.errors,
+        except exceptions.NodeNotFoundError as exc:
+            _logger.warning(
+                "Node %s of project %s not found and cannot be updated. Maybe was it deleted?",
+                exc.node_uuid,
+                exc.project_uuid,
             )
 
-    except exceptions.ProjectNotFoundError as exc:
-        _logger.warning(
-            "Project %s was not found and cannot be updated. Maybe was it deleted?",
-            exc.project_uuid,
-        )
-    except exceptions.ProjectOwnerNotFoundError as exc:
-        _logger.warning(
-            "Project owner of project %s could not be found, is the project valid?",
-            exc.project_uuid,
-        )
-    except exceptions.NodeNotFoundError as exc:
-        _logger.warning(
-            "Node %s of project %s not found and cannot be updated. Maybe was it deleted?",
-            exc.node_uuid,
-            exc.project_uuid,
-        )
-
 
 async def _listen(app: web.Application) -> NoReturn:
     listen_query = f"LISTEN {DB_CHANNEL_NAME};"
diff --git a/services/web/server/tests/data/workbench_2connected_jupyterlabs.json b/services/web/server/tests/data/workbench_2connected_jupyterlabs.json
@@ -0,0 +1,37 @@
+{
+  "e8eae2cd-ae0f-4ba6-ae0b-86eeadf99b42": {
+    "key": "simcore/services/dynamic/jupyter-math",
+    "version": "3.0.5",
+    "label": "JupyterLab Math (Python+Octave)",
+    "inputs": {},
+    "inputsRequired": [],
+    "inputNodes": []
+  },
+  "f7d6dc1e-a6dc-44e1-9588-a2f4b05d3d9c": {
+    "key": "simcore/services/dynamic/jupyter-math",
+    "version": "3.0.5",
+    "label": "JupyterLab Math (Python+Octave)_2",
+    "inputs": {
+      "input_1": {
+        "nodeUuid": "e8eae2cd-ae0f-4ba6-ae0b-86eeadf99b42",
+        "output": "output_1"
+      },
+      "input_2": {
+        "nodeUuid": "e8eae2cd-ae0f-4ba6-ae0b-86eeadf99b42",
+        "output": "output_2"
+      },
+      "input_3": {
+        "nodeUuid": "e8eae2cd-ae0f-4ba6-ae0b-86eeadf99b42",
+        "output": "output_3"
+      },
+      "input_4": {
+        "nodeUuid": "e8eae2cd-ae0f-4ba6-ae0b-86eeadf99b42",
+        "output": "output_4"
+      }
+    },
+    "inputsRequired": [],
+    "inputNodes": [
+      "e8eae2cd-ae0f-4ba6-ae0b-86eeadf99b42"
+    ]
+  }
+}
diff --git a/services/web/server/tests/unit/with_dbs/04/notifications/test_notifications__db_comp_tasks_listening_task.py b/services/web/server/tests/unit/with_dbs/04/notifications/test_notifications__db_comp_tasks_listening_task.py
@@ -5,6 +5,9 @@
 # pylint:disable=too-many-arguments
 # pylint:disable=protected-access
 
+from ast import Assert
+import asyncio
+from datetime import timedelta
 import json
 import logging
 import secrets
@@ -14,6 +17,10 @@
 from unittest import mock
 
 import pytest
+from tenacity import stop_after_attempt
+from common_library.async_tools import delayed_start
+from models_library.projects_nodes import InputsDict
+from pytest_simcore.helpers.logging_tools import log_context
 import simcore_service_webserver
 import simcore_service_webserver.db_listener
 import simcore_service_webserver.db_listener._db_comp_tasks_listening_task
@@ -36,6 +43,8 @@
 from tenacity.stop import stop_after_delay
 from tenacity.wait import wait_fixed
 
+from simcore_service_webserver.projects.models import ProjectDict
+
 logger = logging.getLogger(__name__)
 
 
@@ -205,6 +214,181 @@ async def test_db_listener_triggers_on_event_with_multiple_tasks(
         assert any(
             call.args[1] == updated_task_id
             for call in spied_get_changed_comp_task_row.call_args_list
-        ), f"_get_changed_comp_task_row was not called with task_id={updated_task_id}. Calls: {spied_get_changed_comp_task_row.call_args_list}"
+        ), (
+            f"_get_changed_comp_task_row was not called with task_id={updated_task_id}. Calls: {spied_get_changed_comp_task_row.call_args_list}"
+        )
     else:
         spied_get_changed_comp_task_row.assert_not_called()
+
+
+from pathlib import Path
+
+
+@pytest.fixture
+def fake_2connected_jupyterlabs_workbench(tests_data_dir: Path) -> dict[str, Any]:
+    fpath = tests_data_dir / "workbench_2connected_jupyterlabs.json"
+    assert fpath.exists()
+    return json.loads(fpath.read_text())
+
+
+@pytest.fixture
+async def mock_dynamic_service_rpc(
+    mocker: MockerFixture,
+) -> mock.AsyncMock:
+    """
+    Mocks the dynamic service RPC calls to avoid actual service calls during tests.
+    """
+    return mocker.patch(
+        "servicelib.rabbitmq.rpc_interfaces.dynamic_scheduler.services.retrieve_inputs",
+        autospec=True,
+    )
+
+
+async def _check_for_stability(
+    function: Callable[..., Awaitable[None]], *args, **kwargs
+) -> None:
+    async for attempt in AsyncRetrying(
+        stop=stop_after_attempt(5),
+        wait=wait_fixed(1),
+        retry=retry_if_exception_type(),
+        reraise=True,
+    ):
+        with attempt:  # noqa: SIM117
+            with log_context(
+                logging.INFO,
+                msg=f"check stability of {function.__name__} {attempt.retry_state.retry_object.statistics}",
+            ) as log_ctx:
+                await function(*args, **kwargs)
+                log_ctx.logger.info(
+                    "stable for %s...", attempt.retry_state.seconds_since_start
+                )
+
+
+@pytest.mark.testit
+@pytest.mark.parametrize("user_role", [UserRole.USER])
+async def test_db_listener_upgrades_projects_row_correctly(
+    with_started_listening_task: None,
+    mock_dynamic_service_rpc: mock.AsyncMock,
+    sqlalchemy_async_engine: AsyncEngine,
+    logged_user: UserInfoDict,
+    project: Callable[..., Awaitable[ProjectAtDB]],
+    fake_2connected_jupyterlabs_workbench: dict[str, Any],
+    pipeline: Callable[..., dict[str, Any]],
+    comp_task: Callable[..., dict[str, Any]],
+    spied_get_changed_comp_task_row: MockType,
+    faker: Faker,
+):
+    some_project = await project(
+        logged_user, workbench=fake_2connected_jupyterlabs_workbench
+    )
+
+    # create the corresponding comp_task entries for the project workbench
+    pipeline(project_id=f"{some_project.uuid}")
+    tasks = [
+        comp_task(
+            project_id=f"{some_project.uuid}",
+            node_id=node_id,
+            outputs=node_data.get("outputs", {}),
+            node_class=NodeClass.INTERACTIVE
+            if "dynamic" in node_data["key"]
+            else NodeClass.COMPUTATIONAL,
+            inputs=node_data.get("inputs", InputsDict()),
+        )
+        for node_id, node_data in fake_2connected_jupyterlabs_workbench.items()
+    ]
+    assert len(tasks) == 2, "Expected two tasks for the two JupyterLab nodes"
+    first_jupyter_task = tasks[0]
+    second_jupyter_task = tasks[1]
+    assert len(second_jupyter_task["inputs"]) > 0, (
+        "Expected inputs for the second JupyterLab task"
+    )
+    number_of_inputs_linked = len(second_jupyter_task["inputs"])
+
+    # simulate a concurrent change in all the outputs of first jupyterlab
+    async def _update_first_jupyter_task_output(
+        port_index: int, data: dict[str, Any]
+    ) -> None:
+        with log_context(logging.INFO, msg=f"Updating output {port_index + 1}"):
+            async with sqlalchemy_async_engine.begin() as conn:
+                # For JSON columns, we need to use jsonb_set or fetch-modify-update
+                # Since it's JSON (not JSONB), let's use the safer fetch-modify approach
+                # Use SELECT FOR UPDATE to lock the row for concurrent access
+                result = await conn.execute(
+                    comp_tasks.select()
+                    .with_only_columns([comp_tasks.c.outputs])
+                    .where(comp_tasks.c.task_id == first_jupyter_task["task_id"])
+                    .with_for_update()
+                )
+                row = result.first()
+                current_outputs = row[0] if row and row[0] else {}
+
+                # Update/add the new key while preserving existing keys
+                current_outputs[f"output_{port_index + 1}"] = data
+
+                # Write back the updated outputs
+                await conn.execute(
+                    comp_tasks.update()
+                    .values(outputs=current_outputs)
+                    .where(comp_tasks.c.task_id == first_jupyter_task["task_id"])
+                )
+
+    # await asyncio.gather(
+    #     *(
+    #         _update_first_jupyter_task_output(i, {"data": i})
+    #         for i in range(number_of_inputs_linked)
+    #     )
+    # )
+
+    @delayed_start(timedelta(seconds=2))
+    async def _change_outputs_sequentially(sleep: float = 0.1) -> None:
+        """
+        Sequentially updates the outputs of the second JupyterLab task to trigger the dynamic service RPC.
+        """
+        for i in range(number_of_inputs_linked):
+            await _update_first_jupyter_task_output(i, {"data": i})
+            await asyncio.sleep(sleep)
+
+    # this runs in a task
+    sequential_task = asyncio.create_task(_change_outputs_sequentially(5))
+    assert sequential_task is not None, "Failed to create the sequential task"
+
+    async def _check_retrieve_rpc_called(expected_ports_retrieved: int) -> None:
+        async for attempt in AsyncRetrying(
+            stop=stop_after_delay(60),
+            wait=wait_fixed(1),
+            retry=retry_if_exception_type(AssertionError),
+            reraise=True,
+        ):
+            with attempt:  # noqa: SIM117
+                with log_context(
+                    logging.INFO,
+                    msg=f"Checking if dynamic service retrieve RPC was called and "
+                    f"all expected ports were retrieved {expected_ports_retrieved} "
+                    f"times,  {attempt.retry_state.retry_object.statistics}",
+                ) as log_ctx:
+                    if mock_dynamic_service_rpc.call_count > 0:
+                        log_ctx.logger.info(
+                            "call arguments: %s",
+                            mock_dynamic_service_rpc.call_args_list,
+                        )
+                    # Assert that the dynamic service RPC was called
+                    assert mock_dynamic_service_rpc.call_count > 0, (
+                        "Dynamic service retrieve RPC was not called"
+                    )
+                    # now get we check which ports were retrieved, we expect all of them
+                    all_ports = set()
+                    for call in mock_dynamic_service_rpc.call_args_list:
+                        retrieved_ports = call[1]["port_keys"]
+                        all_ports.update(retrieved_ports)
+                    assert len(all_ports) == expected_ports_retrieved, (
+                        f"Expected {expected_ports_retrieved} ports to be retrieved, "
+                        f"but got {len(all_ports)}: {all_ports}"
+                    )
+                    log_ctx.logger.info(
+                        "Dynamic service retrieve RPC was called with all expected ports!"
+                    )
+
+    await _check_for_stability(_check_retrieve_rpc_called, number_of_inputs_linked)
+
+    assert sequential_task.done(), "Sequential task did not complete"
+    assert not sequential_task.cancelled(), "Sequential task was cancelled unexpectedly"
