added cancellation test

Author: Andrei Neagu

packages/service-library/src/servicelib/long_running_interfaces/_client.py

@@ -296,5 +296,8 @@ async def ensure_result(
             # when completed remove the task form memory both on the server and the client
             # NOTE: unsure if these should be caught and ignored. In the case we decide to
             # catch them there should be some automatic cleanup in case they are forgotten
+
+            # remove from remote executor
             await self._rpc_interface.remove(unique_id)
+            # remove distributed storage (redis)
             await self._store_interface.remove(unique_id)

packages/service-library/src/servicelib/long_running_interfaces/_server.py

@@ -40,14 +40,14 @@ def __init__(
         long_running_namespace: LongRunningNamespace,
         job_interface: BaseServerJobInterface,
     ) -> None:
-        self._rpc_interface = ServerRPCInterface(
+        self.rpc_interface = ServerRPCInterface(
             rabbit_settings, long_running_namespace, job_interface
         )
 
     # TODO: register jobs! and then use the interface to run them!!!
 
     async def setup(self) -> None:
-        await self._rpc_interface.setup()
+        await self.rpc_interface.setup()
 
     async def teardown(self) -> None:
-        await self._rpc_interface.teardown()
+        await self.rpc_interface.teardown()

packages/service-library/src/servicelib/long_running_interfaces/runners/asyncio_tasks/_core.py

@@ -70,6 +70,3 @@ async def get_result(self, unique_id: JobUniqueId) -> Any | None:
 
         task = self._tasks[unique_id]
         return task.result()
-
-
-# TODO: add tests individually for these so we knoe they work when registring starting and stopping and all the operations. Also cancellation

packages/service-library/tests/long_running_interfaces/test_workflow.py

@@ -9,6 +9,7 @@
 
 import pytest
 from pydantic import NonNegativeInt, ValidationError
+from servicelib.async_utils import cancel_wait_task
 from servicelib.long_running_interfaces import (
     Client,
     FinishedWithError,
@@ -22,6 +23,12 @@
 )
 from settings_library.rabbit import RabbitSettings
 from settings_library.redis import RedisSettings
+from tenacity import (
+    retry,
+    retry_if_exception_type,
+    stop_after_delay,
+    wait_fixed,
+)
 
 pytest_simcore_core_services_selection = [
     "rabbit",
@@ -157,3 +164,31 @@ async def test_timeout_error_retry_count_zero(server: Server, client: Client):
 
     assert "retry_count" in f"{exec_info.value}"
     assert "Input should be greater than 0" in f"{exec_info.value}"
+
+
+@retry(
+    wait=wait_fixed(0.1),
+    stop=stop_after_delay(5),
+    retry=retry_if_exception_type(AssertionError),
+)  # NOTE: function has to be async or the loop does not get a cahce to switch between retries
+async def _assert_tasks_count(server: Server, count: int) -> None:
+    assert isinstance(server.rpc_interface.job_interface, AsyncioTasksJobInterface)
+    task_count = len(
+        server.rpc_interface.job_interface._tasks.values()  # noqa: SLF001 # pylint:disable=protected-access
+    )
+    assert task_count == count
+
+
+async def test_cancellation_from_client(server: Server, client: Client):
+    async def _to_run() -> None:
+        await client.ensure_result(
+            "sleep_forever_f", expected_type=type(None), timeout=timedelta(seconds=1)
+        )
+
+    await _assert_tasks_count(server, count=0)
+
+    task = asyncio.create_task(_to_run())
+    await _assert_tasks_count(server, count=1)
+
+    await cancel_wait_task(task, max_delay=5)
+    await _assert_tasks_count(server, count=0)