Cleanup

 - joined and named mockers from directory-api app module
 - renaming
 - retry for background

Author: pcrespov

services/web/server/src/simcore_service_webserver/projects/projects_api.py

@@ -178,11 +178,11 @@ async def start_project_interactive_services(
 async def delete_project(request: web.Request, project_uuid: str, user_id: int) -> None:
     await delete_project_from_db(request.app, project_uuid, user_id)
 
-    async def remove_services_and_data():
+    async def _remove_services_and_data():
         await remove_project_interactive_services(user_id, project_uuid, request.app)
         await delete_project_data(request, project_uuid, user_id)
 
-    fire_and_forget_task(remove_services_and_data())
+    fire_and_forget_task(_remove_services_and_data())
 
 
 ## PROJECT NODES -----------------------------------------------------

services/web/server/src/simcore_service_webserver/resource_manager/garbage_collector.py

@@ -51,25 +51,25 @@
 logger = logging.getLogger(__name__)
 
 
-async def _setup_garbage_collector_task(app: web.Application):
-
-    app[APP_GARBAGE_COLLECTOR_KEY] = asyncio.create_task(
-        garbage_collector_task(app), name="garbage-collector"
-    )
-
-    yield
+def setup_garbage_collector(app: web.Application):
+    async def _setup_background_task(app: web.Application):
+        # on_startup
+        app[APP_GARBAGE_COLLECTOR_KEY] = asyncio.create_task(
+            collect_garbage_periodically(app), name="garbage-collector"
+        )
 
-    with suppress(asyncio.CancelledError):
-        task = app[APP_GARBAGE_COLLECTOR_KEY]
-        task.cancel()
-        await task
+        yield
 
+        # on_cleanup
+        with suppress(asyncio.CancelledError):
+            task = app[APP_GARBAGE_COLLECTOR_KEY]
+            task.cancel()
+            await task
 
-def setup_garbage_collector(app: web.Application):
-    app.cleanup_ctx.append(_setup_garbage_collector_task)
+    app.cleanup_ctx.append(_setup_background_task)
 
 
-async def garbage_collector_task(app: web.Application):
+async def collect_garbage_periodically(app: web.Application):
 
     while True:
         logger.info("Starting garbage collector...")
@@ -358,8 +358,8 @@ async def remove_orphaned_services(
             logger.info("Will remove service %s", interactive_service["service_host"])
             try:
                 await stop_service(app, node_id)
-            except (ServiceNotFoundError, DirectorException) as e:
-                logger.warning("Error while stopping service: %s", e)
+            except (ServiceNotFoundError, DirectorException) as err:
+                logger.warning("Error while stopping service: %s", err)
 
     logger.info("Finished orphaned services removal")
 

services/web/server/src/simcore_service_webserver/socketio/handlers.py

@@ -11,10 +11,9 @@
 from typing import Any, Dict, List, Optional
 
 from aiohttp import web
-from socketio.exceptions import ConnectionRefusedError as SocketIOConnectionError
-
 from servicelib.observer import observe
 from servicelib.utils import fire_and_forget_task, logged_gather
+from socketio.exceptions import ConnectionRefusedError as SocketIOConnectionError
 
 from ..groups_api import list_user_groups
 from ..login.decorators import RQT_USERID_KEY, login_required
@@ -68,8 +67,7 @@ async def connect(sid: str, environ: Dict, app: web.Application) -> bool:
 async def authenticate_user(
     sid: str, app: web.Application, request: web.Request
 ) -> None:
-    """throws web.HTTPUnauthorized when the user is not recognized. Keeps the original request.
-    """
+    """throws web.HTTPUnauthorized when the user is not recognized. Keeps the original request."""
     user_id = request.get(RQT_USERID_KEY, ANONYMOUS_USER_ID)
     log.debug("client %s authenticated", user_id)
     client_session_id = request.query.get("client_session_id", None)
@@ -116,7 +114,7 @@ async def disconnect_other_sockets(sio, sockets: List[str]) -> None:
 
 
 @observe(event="SIGNAL_USER_LOGOUT")
-async def user_logged_out(
+async def on_user_logout(
     user_id: str, client_session_id: Optional[str], app: web.Application
 ) -> None:
     log.debug("user %s must be disconnected", user_id)

services/web/server/tests/unit/with_dbs/conftest.py

@@ -229,40 +229,25 @@ def asyncpg_storage_system_mock(mocker):
     return mocked_method
 
 
-@pytest.fixture
-def mocked_director_subsystem(mocker):
-    mock_director_api = {
-        "get_running_interactive_services": mocker.patch(
-            "simcore_service_webserver.director.director_api.get_running_interactive_services",
-            return_value="",
-        ),
-        "start_service": mocker.patch(
-            "simcore_service_webserver.director.director_api.start_service",
-            return_value="",
-        ),
-        "stop_service": mocker.patch(
-            "simcore_service_webserver.director.director_api.stop_service",
-            return_value="",
-        ),
-    }
-    return mock_director_api
-
-
 @pytest.fixture
 async def mocked_director_api(loop, mocker):
+    # NOTE: patches are done at 'simcore_service_webserver.director.director_api'
+    #
+    #  Read carefully "where to patch" in https://docs.python.org/3/library/unittest.mock.html#id6
+    #
     mocks = {}
-    mocked_running_services = mocker.patch(
-        "simcore_service_webserver.director.director_api.get_running_interactive_services",
-        return_value="",
-    )
-    mocks["get_running_interactive_services"] = mocked_running_services
-    mocked_stop_service = mocker.patch(
-        "simcore_service_webserver.director.director_api.stop_service",
-        return_value="",
-    )
-    mocks["stop_service"] = mocked_stop_service
+    for func_name, fake_return in [
+        ("get_running_interactive_services", []),
+        ("start_service", []),
+        ("stop_service", None),
+    ]:
+        mocks[func_name] = mocker.patch(
+            f"simcore_service_webserver.director.director_api.{func_name}",
+            return_value=fake_return,
+            name=f"{__name__}.mocked_director_api::director_api.{func_name}",
+        )
 
-    yield mocks
+    return mocks
 
 
 @pytest.fixture

services/web/server/tests/unit/with_dbs/medium/test_resource_manager.py

@@ -3,6 +3,7 @@
 # pylint:disable=redefined-outer-name
 
 
+import logging
 from asyncio import sleep
 from copy import deepcopy
 from typing import Callable
@@ -30,10 +31,27 @@
 from simcore_service_webserver.session import setup_session
 from simcore_service_webserver.socketio import setup_socketio
 from simcore_service_webserver.users import setup_users
+from tenacity import (
+    AsyncRetrying,
+    after_log,
+    retry_if_exception_type,
+    stop_after_attempt,
+    wait_fixed,
+)
+
+logger = logging.getLogger(__name__)
+
 
 API_VERSION = "v0"
 GARBAGE_COLLECTOR_INTERVAL = 1
 SERVICE_DELETION_DELAY = 1
+CHECK_BACKGROUND_RETRY_POLICY = dict(
+    stop=stop_after_attempt(2),
+    wait=wait_fixed(SERVICE_DELETION_DELAY + GARBAGE_COLLECTOR_INTERVAL),
+    retry=retry_if_exception_type(AssertionError),
+    after=after_log(logger, logging.INFO),
+    reraise=True,
+)
 
 
 @pytest.fixture
@@ -352,11 +370,15 @@ async def test_interactive_services_removed_after_logout(
     r = await client.post(logout_url, json={"client_session_id": client_session_id1})
     assert r.url_obj.path == logout_url.path
     await assert_status(r, web.HTTPOk)
-    # ensure sufficient time is wasted here
-    await sleep(SERVICE_DELETION_DELAY + GARBAGE_COLLECTOR_INTERVAL + 1)
-    # assert dynamic service is removed
-    calls = [call(client.server.app, service["service_uuid"])]
-    mocked_director_api["stop_service"].assert_has_calls(calls)
+
+    # check result perfomed by background task
+    async for attempt in AsyncRetrying(**CHECK_BACKGROUND_RETRY_POLICY):
+        with attempt:
+            print(".")
+            # assert dynamic service is removed
+            mocked_director_api["stop_service"].assert_awaited_with(
+                client.server.app, service["service_uuid"]
+            )
 
 
 @pytest.mark.parametrize(