fixed test

Author: Andrei Neagu

services/storage/src/simcore_service_storage/api/rpc/_async_jobs.py

@@ -19,7 +19,6 @@
     JobNotDoneError,
     JobSchedulerError,
 )
-from servicelib.logging_utils import log_catch
 from servicelib.rabbitmq import RPCRouter
 
 from ...modules.celery import get_celery_client
@@ -96,14 +95,10 @@ async def result(
     if _status.task_state == TaskState.ABORTED:
         raise JobAbortedError(job_id=job_id)
     if _status.task_state == TaskState.ERROR:
-        exc_type = ""
-        exc_msg = ""
-        with log_catch(logger=_logger, reraise=False):
-            # NOTE: recover original error from wrapped error
-            exception = pickle.loads(base64.b64decode(_result.args[0]))
-            exc_type = type(exception).__name__
-            exc_msg = f"{exception}"
-
+        # NOTE: recover original error from wrapped error
+        exception = pickle.loads(base64.b64decode(_result.args[0]))
+        exc_type = type(exception).__name__
+        exc_msg = f"{exception}"
         raise JobError(job_id=job_id, exc_type=exc_type, exc_msg=exc_msg, exc=exception)
 
     return AsyncJobResult(result=_result)

services/storage/src/simcore_service_storage/modules/celery/_task.py

@@ -28,6 +28,14 @@
 R = TypeVar("R")
 
 
+def _wrap_exception(exc: Exception) -> Exception:
+    return Exception(
+        base64.b64encode(pickle.dumps(exc, protocol=pickle.HIGHEST_PROTOCOL)).decode(
+            "ascii"
+        )
+    )
+
+
 def _async_task_wrapper(
     app: Celery,
 ) -> Callable[
@@ -73,15 +81,10 @@ def wrapper(task: Task, *args: Any, **kwargs: Any) -> Any:
                 # by running it's magic, it looses the context of some errors
                 # this allows to recreate the same error in the caller side excatly as
                 # it was raised in this context
-                wrapping_exception = Exception(
-                    base64.b64encode(
-                        pickle.dumps(exc, protocol=pickle.HIGHEST_PROTOCOL)
-                    ).decode("ascii")
-                )
                 raise task.retry(
                     max_retries=max_retries,
                     countdown=delay_between_retries.total_seconds(),
-                    exc=wrapping_exception,
+                    exc=_wrap_exception(exc),
                 )
 
         return wrapper

services/storage/tests/unit/test_async_jobs.py

@@ -28,6 +28,7 @@
 from pytest_mock import MockerFixture
 from servicelib.rabbitmq import RabbitMQRPCClient
 from servicelib.rabbitmq.rpc_interfaces.async_jobs import async_jobs
+from simcore_service_storage.modules.celery._task import _wrap_exception
 from simcore_service_storage.modules.celery.client import TaskUUID
 from simcore_service_storage.modules.celery.models import TaskState, TaskStatus
 
@@ -68,7 +69,7 @@ async def get_task_result(self, *args, **kwargs) -> Any:
         _ = kwargs
         assert self.get_task_result_object is not None
         if isinstance(self.get_task_result_object, Exception):
-            raise self.get_task_result_object
+            return _wrap_exception(self.get_task_result_object)
         return self.get_task_result_object
 
     async def get_task_uuids(self, *args, **kwargs) -> set[TaskUUID]:
@@ -318,7 +319,7 @@ async def test_async_jobs_result_success(
                     task_state=TaskState.ERROR,
                     progress_report=ProgressReport(actual_value=1.0, total=1.0),
                 ),
-                "get_task_result_object": _faker.text(),
+                "get_task_result_object": Exception("generic exception"),
                 "get_task_uuids_object": [AsyncJobId(_faker.uuid4())],
             },
             JobError,