tests

services/web/server/src/simcore_service_webserver/catalog/_api_units.py

@@ -1,9 +1,5 @@
 from typing import Any
 
-from models_library.api_schemas_webserver.catalog import (
-    ServiceInputGet,
-    ServiceOutputGet,
-)
 from models_library.services import BaseServiceIOModel, ServiceInput, ServiceOutput
 from pint import PintError, UnitRegistry
 
@@ -45,20 +41,25 @@ async def replace_service_input_outputs(
 ):
     """Thin wrapper to replace i/o ports in returned service model"""
     # This is a fast solution until proper models are available for the web API
-    for input_key in service["inputs"]:
-        new_input: ServiceInputGet = (
-            await ServiceInputGetFactory.from_catalog_service_api_model(
-                service=service, input_key=input_key, ureg=unit_registry
-            )
+    new_inputs = [
+        await ServiceInputGetFactory.from_catalog_service_api_model(
+            service=service, input_key=input_key, ureg=unit_registry
         )
-        service["inputs"][input_key] = new_input.dict(**export_options)
+        for input_key in service["inputs"]
+    ]
 
-    for output_key in service["outputs"]:
-        new_output: ServiceOutputGet = (
-            await ServiceOutputGetFactory.from_catalog_service_api_model(
-                service=service, output_key=output_key, ureg=unit_registry
-            )
+    new_outputs = [
+        await ServiceOutputGetFactory.from_catalog_service_api_model(
+            service=service, output_key=output_key, ureg=unit_registry
         )
+        for output_key in service["outputs"]
+    ]
+
+    # replace if above is successful
+    for input_key, new_input in zip(service["inputs"], new_inputs, strict=True):
+        service["inputs"][input_key] = new_input.dict(**export_options)
+
+    for output_key, new_output in zip(service["outputs"], new_outputs, strict=True):
         service["outputs"][output_key] = new_output.dict(**export_options)
 
 

services/web/server/tests/unit/isolated/test_catalog_models.py

@@ -2,6 +2,7 @@
 # pylint: disable=unused-argument
 # pylint: disable=unused-variable
 
+import asyncio
 import json
 from copy import deepcopy
 
@@ -69,14 +70,16 @@ def test_from_catalog_to_webapi_service(
         "owner": "[email protected]",
     }
 
-    def _run():
+    def _run_async_test():
         s = deepcopy(catalog_service)
-        replace_service_input_outputs(
-            s, unit_registry=unit_registry, **RESPONSE_MODEL_POLICY
+        asyncio.get_event_loop().run_until_complete(
+            replace_service_input_outputs(
+                s, unit_registry=unit_registry, **RESPONSE_MODEL_POLICY
+            )
         )
         return s
 
-    result = benchmark(_run)
+    result = benchmark(_run_async_test)
 
     # check result
     got = json.dumps(result, indent=1)