Merge branch 'master' into enh/state-through-store

Author: odeimaiz

api/specs/web-server/_computations.py

@@ -6,7 +6,9 @@
 from models_library.api_schemas_webserver.computations import (
     ComputationCollectionRunListQueryParams,
     ComputationCollectionRunPathParams,
+    ComputationCollectionRunRestGet,
     ComputationCollectionRunTaskListQueryParams,
+    ComputationCollectionRunTaskRestGet,
     ComputationGet,
     ComputationPathParams,
     ComputationRunIterationsLatestListQueryParams,
@@ -102,7 +104,7 @@ async def list_computations_latest_iteration_tasks(
 
 @router.get(
     "/computation-collection-runs",
-    response_model=Page[ComputationTaskRestGet],
+    response_model=Page[ComputationCollectionRunRestGet],
 )
 async def list_computation_collection_runs(
     _query: Annotated[as_query(ComputationCollectionRunListQueryParams), Depends()],
@@ -111,7 +113,7 @@ async def list_computation_collection_runs(
 
 @router.get(
     "/computation-collection-runs/{collection_run_id}/tasks",
-    response_model=Page[ComputationTaskRestGet],
+    response_model=Page[ComputationCollectionRunTaskRestGet],
 )
 async def list_computation_collection_run_tasks(
     _query: Annotated[as_query(ComputationCollectionRunTaskListQueryParams), Depends()],

api/specs/web-server/_users.py

@@ -14,20 +14,22 @@
     MyProfilePatch,
     MyTokenCreate,
     MyTokenGet,
+    TokenPathParams,
     UserGet,
     UsersSearch,
 )
 from models_library.api_schemas_webserver.users_preferences import PatchRequestBody
 from models_library.generics import Envelope
 from models_library.user_preferences import PreferenceIdentifier
 from simcore_service_webserver._meta import API_VTAG
-from simcore_service_webserver.users._notifications import (
+from simcore_service_webserver.user_notifications._controller.rest.user_notification_rest import (
+    NotificationPathParams,
+)
+from simcore_service_webserver.user_notifications._models import (
     UserNotification,
     UserNotificationCreate,
     UserNotificationPatch,
 )
-from simcore_service_webserver.users._notifications_rest import _NotificationPathParams
-from simcore_service_webserver.users._tokens_rest import _TokenPathParams
 
 router = APIRouter(prefix=f"/{API_VTAG}", tags=["users"])
 
@@ -76,15 +78,15 @@ async def create_token(_body: MyTokenCreate): ...
     response_model=Envelope[MyTokenGet],
 )
 async def get_token(
-    _path: Annotated[_TokenPathParams, Depends()],
+    _path: Annotated[TokenPathParams, Depends()],
 ): ...
 
 
 @router.delete(
     "/me/tokens/{service}",
     status_code=status.HTTP_204_NO_CONTENT,
 )
-async def delete_token(_path: Annotated[_TokenPathParams, Depends()]): ...
+async def delete_token(_path: Annotated[TokenPathParams, Depends()]): ...
 
 
 @router.get(
@@ -108,7 +110,7 @@ async def create_user_notification(
     status_code=status.HTTP_204_NO_CONTENT,
 )
 async def mark_notification_as_read(
-    _path: Annotated[_NotificationPathParams, Depends()],
+    _path: Annotated[NotificationPathParams, Depends()],
     _body: UserNotificationPatch,
 ): ...
 

packages/models-library/src/models_library/api_schemas_webserver/users.py

@@ -11,6 +11,7 @@
 from models_library.rest_filters import Filters
 from models_library.rest_pagination import PageQueryParameters
 from pydantic import (
+    BaseModel,
     ConfigDict,
     EmailStr,
     Field,
@@ -334,6 +335,10 @@ def _consistency_check(cls, v, info: ValidationInfo):
 #
 
 
+class TokenPathParams(BaseModel):
+    service: str
+
+
 class MyTokenCreate(InputSchemaWithoutCamelCase):
     service: Annotated[
         IDStr,

packages/service-library/src/servicelib/aiohttp/observer.py

@@ -8,6 +8,7 @@
 from collections.abc import Callable
 
 from aiohttp import web
+from servicelib.aiohttp.application_setup import ensure_single_setup
 
 from ..utils import logged_gather
 
@@ -17,10 +18,10 @@
 _APP_OBSERVER_EVENTS_REGISTRY_KEY = "{__name__}.event_registry"
 
 
-class ObserverRegistryNotFoundError(RuntimeError):
-    ...
+class ObserverRegistryNotFoundError(RuntimeError): ...
 
 
+@ensure_single_setup(__name__, logger=log)
 def setup_observer_registry(app: web.Application):
     # only once
     app.setdefault(_APP_OBSERVER_EVENTS_REGISTRY_KEY, defaultdict(list))

services/director-v2/src/simcore_service_director_v2/api/rpc/_computations.py

@@ -114,6 +114,8 @@ async def list_computation_collection_runs_page(
         collection_run_ids = await comp_runs_repo.list_all_collection_run_ids_for_user_currently_running_computations(
             product_name=product_name, user_id=user_id
         )
+        if collection_run_ids == []:
+            return ComputationCollectionRunRpcGetPage(items=[], total=0)
 
     total, comp_runs_output = await comp_runs_repo.list_group_by_collection_run_id(
         product_name=product_name,

services/director-v2/src/simcore_service_director_v2/modules/db/repositories/comp_runs.py

@@ -458,13 +458,13 @@ async def list_group_by_collection_run_id(
             & (comp_runs.c.metadata["product_name"].astext == product_name)
         )
 
-        if project_ids_or_none:
+        if project_ids_or_none is not None:
             base_select_query = base_select_query.where(
                 comp_runs.c.project_uuid.in_(
                     [f"{project_id}" for project_id in project_ids_or_none]
                 )
             )
-        if collection_run_ids_or_none:
+        if collection_run_ids_or_none is not None:
             base_select_query = base_select_query.where(
                 comp_runs.c.collection_run_id.in_(
                     [

services/director-v2/tests/unit/with_dbs/comp_scheduler/test_api_rpc_computations.py

@@ -341,3 +341,82 @@ async def test_rpc_list_computation_collection_runs_page_and_collection_run_task
     assert len(output.items) == 1
     assert isinstance(output, ComputationCollectionRunRpcGetPage)
     assert len(output.items[0].project_ids) == 2
+
+
+async def test_rpc_list_computation_collection_runs_empty_ids_when_user_has_already_run_history(
+    fake_workbench_without_outputs: dict[str, Any],  # <-- Has 4 nodes
+    fake_workbench_adjacency: dict[str, Any],
+    create_registered_user: Callable[..., dict[str, Any]],
+    create_project: Callable[..., Awaitable[ProjectAtDB]],
+    create_pipeline: Callable[..., Awaitable[CompPipelineAtDB]],
+    create_tasks_from_project: Callable[..., Awaitable[list[CompTaskAtDB]]],
+    create_comp_run_snapshot_tasks: Callable[
+        ..., Awaitable[list[CompRunSnapshotTaskDBGet]]
+    ],
+    create_comp_run: Callable[..., Awaitable[CompRunsAtDB]],
+    rpc_client: RabbitMQRPCClient,
+    faker: Faker,
+    with_product: dict[str, Any],
+):
+    user = create_registered_user()
+    proj = await create_project(user, workbench=fake_workbench_without_outputs)
+
+    await create_pipeline(
+        project_id=f"{proj.uuid}",
+        dag_adjacency_list=fake_workbench_adjacency,
+    )
+    await create_tasks_from_project(
+        user=user, project=proj, state=RunningState.SUCCESS, progress=None
+    )
+    run = await create_comp_run(
+        user=user,
+        project=proj,
+        result=RunningState.SUCCESS,
+        started=datetime.now(tz=UTC) - timedelta(minutes=120),
+        ended=datetime.now(tz=UTC) - timedelta(minutes=100),
+        iteration=1,
+        dag_adjacency_list=fake_workbench_adjacency,
+    )
+    await create_comp_run_snapshot_tasks(
+        user=user,
+        project=proj,
+        run_id=run.run_id,
+    )
+
+    output = await rpc_computations.list_computation_collection_runs_page(
+        rpc_client, product_name="osparc", user_id=user["id"], project_ids=None
+    )
+    assert output.total == 1
+    assert len(output.items) == 1
+    assert isinstance(output, ComputationCollectionRunRpcGetPage)
+
+    # Test filtering only running collection runs
+    output = await rpc_computations.list_computation_collection_runs_page(
+        rpc_client,
+        product_name="osparc",
+        user_id=user["id"],
+        project_ids=None,
+        filter_only_running=True,  # <-- This is the tested filter
+    )
+    assert output.total == 0
+    assert len(output.items) == 0
+
+
+async def test_rpc_list_computation_collection_runs_empty_ids_when_user_do_not_have_run_history(
+    create_registered_user: Callable[..., dict[str, Any]],
+    rpc_client: RabbitMQRPCClient,
+    with_product: dict[str, Any],
+):
+    user = create_registered_user()
+
+    # Test with empty collection_run_ids
+    output = await rpc_computations.list_computation_collection_runs_page(
+        rpc_client,
+        product_name="osparc",
+        user_id=user["id"],
+        project_ids=None,
+        filter_only_running=True,  # This will result in empty collection_run_ids
+    )
+    assert output.total == 0
+    assert len(output.items) == 0
+    assert isinstance(output, ComputationCollectionRunRpcGetPage)