✨ [Backend] Add async function for paginated listing of service summaries and refactor mock setup in tests

Author: pcrespov

packages/models-library/src/models_library/api_schemas_catalog/services.py

@@ -182,6 +182,41 @@ class ServiceSummary(CatalogOutputSchema):
 
     contact: LowerCaseEmailStr | None
 
+    service_type: Annotated[ServiceType, Field(alias="type")]
+
+    @staticmethod
+    def _update_json_schema_extra(schema: JsonDict) -> None:
+        schema.update(
+            {
+                "examples": [
+                    {
+                        "key": _EXAMPLE_SLEEPER["key"],
+                        "version": _EXAMPLE_SLEEPER["version"],
+                        "name": _EXAMPLE_SLEEPER["name"],
+                        "description": _EXAMPLE_SLEEPER["description"],
+                        "version_display": _EXAMPLE_SLEEPER["version_display"],
+                        "contact": _EXAMPLE_SLEEPER["contact"],
+                        "type": _EXAMPLE_SLEEPER["type"],
+                    },
+                    {
+                        "key": _EXAMPLE_FILEPICKER["key"],
+                        "version": _EXAMPLE_FILEPICKER["version"],
+                        "name": _EXAMPLE_FILEPICKER["name"],
+                        "description": _EXAMPLE_FILEPICKER["description"],
+                        "version_display": None,
+                        "contact": _EXAMPLE_FILEPICKER["contact"],
+                        "type": _EXAMPLE_FILEPICKER["type"],
+                    },
+                ]
+            }
+        )
+
+    model_config = ConfigDict(
+        extra="ignore",
+        populate_by_name=True,
+        json_schema_extra=_update_json_schema_extra,
+    )
+
 
 class _BaseServiceGetV2(ServiceSummary):
     # Model used in catalog's rpc and rest interfaces
@@ -190,8 +225,6 @@ class _BaseServiceGetV2(ServiceSummary):
 
     description_ui: bool = False
 
-    service_type: Annotated[ServiceType, Field(alias="type")]
-
     authors: Annotated[list[Author], Field(min_length=1)]
     owner: Annotated[
         LowerCaseEmailStr | None,

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

@@ -13,6 +13,7 @@
     LatestServiceGet,
     ServiceGetV2,
     ServiceListFilters,
+    ServiceSummary,
     ServiceUpdateV2,
 )
 from models_library.api_schemas_catalog.services_ports import ServicePortGet
@@ -200,6 +201,62 @@ async def get_service_ports(
             ServicePortGet.model_json_schema()["examples"],
         )
 
+    @validate_call(config={"arbitrary_types_allowed": True})
+    async def list_all_services_summaries_paginated(
+        self,
+        rpc_client: RabbitMQRPCClient | MockType,
+        *,
+        product_name: ProductName,
+        user_id: UserID,
+        limit: PageLimitInt,
+        offset: NonNegativeInt,
+        filters: ServiceListFilters | None = None,
+    ):
+        assert rpc_client
+        assert product_name
+        assert user_id
+
+        service_summaries = TypeAdapter(list[ServiceSummary]).validate_python(
+            ServiceSummary.model_json_schema()["examples"],
+        )
+        if filters:
+            filtered_summaries = []
+            for summary in service_summaries:
+                # Match service type if specified
+                if (
+                    filters.service_type
+                    and summary.service_type != filters.service_type
+                ):
+                    continue
+
+                # Match service key pattern if specified
+                if filters.service_key_pattern and not fnmatch.fnmatch(
+                    summary.key, filters.service_key_pattern
+                ):
+                    continue
+
+                # Match version display pattern if specified
+                if filters.version_display_pattern and (
+                    summary.version_display is None
+                    or not fnmatch.fnmatch(
+                        summary.version_display, filters.version_display_pattern
+                    )
+                ):
+                    continue
+
+                filtered_summaries.append(summary)
+
+            service_summaries = filtered_summaries
+
+        total_count = len(service_summaries)
+
+        return PageRpc[ServiceSummary].create(
+            service_summaries[offset : offset + limit],
+            total=total_count,
+            limit=limit,
+            offset=offset,
+        )
+
 
 @dataclass
 class ZeroListingCatalogRpcSideEffects:
@@ -216,3 +273,11 @@ async def list_my_service_history_latest_first(self, *args, **kwargs):
             limit=10,
             offset=0,
         )
+
+    async def list_all_services_summaries_paginated(self, *args, **kwargs):
+        return PageRpc[ServiceSummary].create(
+            [],
+            total=0,
+            limit=10,
+            offset=0,
+        )

services/api-server/tests/unit/conftest.py

@@ -509,38 +509,27 @@ def mocked_catalog_rpc_api(
         services as catalog_rpc,  # keep import here
     )
 
-    return {
-        "list_services_paginated": mocker.patch.object(
-            catalog_rpc,
-            "list_services_paginated",
-            autospec=True,
-            side_effect=catalog_rpc_side_effects.list_services_paginated,
-        ),
-        "get_service": mocker.patch.object(
-            catalog_rpc,
-            "get_service",
-            autospec=True,
-            side_effect=catalog_rpc_side_effects.get_service,
-        ),
-        "update_service": mocker.patch.object(
-            catalog_rpc,
-            "update_service",
-            autospec=True,
-            side_effect=catalog_rpc_side_effects.update_service,
-        ),
-        "list_my_service_history_latest_first": mocker.patch.object(
-            catalog_rpc,
-            "list_my_service_history_latest_first",
-            autospec=True,
-            side_effect=catalog_rpc_side_effects.list_my_service_history_latest_first,
-        ),
-        "get_service_ports": mocker.patch.object(
-            catalog_rpc,
-            "get_service_ports",
-            autospec=True,
-            side_effect=catalog_rpc_side_effects.get_service_ports,
-        ),
-    }
+    mocks = {}
+
+    # Get all callable methods from the side effects class that are not built-ins
+    side_effect_methods = [
+        method_name
+        for method_name in dir(catalog_rpc_side_effects)
+        if not method_name.startswith("_")
+        and callable(getattr(catalog_rpc_side_effects, method_name))
+    ]
+
+    # Create mocks for each method in catalog_rpc that has a corresponding side effect
+    for method_name in side_effect_methods:
+        if hasattr(catalog_rpc, method_name):
+            mocks[method_name] = mocker.patch.object(
+                catalog_rpc,
+                method_name,
+                autospec=True,
+                side_effect=getattr(catalog_rpc_side_effects, method_name),
+            )
+
+    return mocks
 
 
 #