♻️✅ Refactor webserver pytest helpers to isolate user and login setup logic

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

@@ -1,48 +1,21 @@
 import contextlib
 import re
 from collections.abc import AsyncIterator
-from datetime import datetime
-from typing import Any, TypedDict
+from typing import Any
 
-from aiohttp import web
 from aiohttp.test_utils import TestClient
-from models_library.users import UserID
 from servicelib.aiohttp import status
-from simcore_service_webserver.db.models import UserRole, UserStatus
-from simcore_service_webserver.groups.api import auto_add_user_to_product_group
 from simcore_service_webserver.login._constants import MSG_LOGGED_IN
 from simcore_service_webserver.login._invitations_service import create_invitation_token
 from simcore_service_webserver.login._login_repository_legacy import (
-    AsyncpgStorage,
     get_plugin_storage,
 )
-from simcore_service_webserver.products.products_service import list_products
 from simcore_service_webserver.security import security_service
 from yarl import URL
 
 from .assert_checks import assert_status
-from .faker_factories import DEFAULT_FAKER, DEFAULT_TEST_PASSWORD, random_user
-
-
-# WARNING: DO NOT use UserDict is already in https://docs.python.org/3/library/collections.html#collections.UserDictclass UserRowDict(TypedDict):
-# NOTE: this is modified dict version of packages/postgres-database/src/simcore_postgres_database/models/users.py for testing purposes
-class _UserInfoDictRequired(TypedDict, total=True):
-    id: int
-    name: str
-    email: str
-    primary_gid: str
-    raw_password: str
-    status: UserStatus
-    role: UserRole
-
-
-class UserInfoDict(_UserInfoDictRequired, total=False):
-    created_at: datetime
-    password_hash: str
-    first_name: str
-    last_name: str
-    phone: str
-
+from .faker_factories import DEFAULT_FAKER
+from .webserver_users import NewUser, UserInfoDict, _create_account_in_db
 
 TEST_MARKS = re.compile(r"TEST (\w+):(.*)")
 
@@ -65,76 +38,21 @@ def parse_link(text):
     return URL(link).path
 
 
-async def _create_user(app: web.Application, data=None) -> UserInfoDict:
-    db: AsyncpgStorage = get_plugin_storage(app)
-
-    # create
-    data = data or {}
-    data.setdefault("status", UserStatus.ACTIVE.name)
-    data.setdefault("role", UserRole.USER.name)
-    data.setdefault("password", DEFAULT_TEST_PASSWORD)
-    user = await db.create_user(random_user(**data))
-
-    # get
-    user = await db.get_user({"id": user["id"]})
-    assert "first_name" in user
-    assert "last_name" in user
-
-    # adds extras
-    extras = {"raw_password": data["password"]}
-
-    return UserInfoDict(
-        **{
-            key: user[key]
-            for key in [
-                "id",
-                "name",
-                "email",
-                "primary_gid",
-                "status",
-                "role",
-                "created_at",
-                "password_hash",
-                "first_name",
-                "last_name",
-                "phone",
-            ]
-        },
-        **extras,
-    )
-
-
-async def _register_user_in_default_product(app: web.Application, user_id: UserID):
-    products = list_products(app)
-    assert products
-    product_name = products[0].name
-
-    return await auto_add_user_to_product_group(app, user_id, product_name=product_name)
-
-
-async def _create_account(
-    app: web.Application,
-    user_data: dict[str, Any] | None = None,
-) -> UserInfoDict:
-    # users, groups in db
-    user = await _create_user(app, user_data)
-    # user has default product
-    await _register_user_in_default_product(app, user_id=user["id"])
-    return user
-
-
 async def log_client_in(
     client: TestClient,
     user_data: dict[str, Any] | None = None,
     *,
-    enable_check=True,
+    exit_stack: contextlib.AsyncExitStack,
+    enable_check: bool = True,
 ) -> UserInfoDict:
     assert client.app
 
     # create account
-    user = await _create_account(client.app, user_data=user_data)
+    user = await _create_account_in_db(
+        client.app, exit_stack=exit_stack, user_data=user_data
+    )
 
-    # login
+    # login (requires)
     url = client.app.router["auth_login"].url_for()
     reponse = await client.post(
         str(url),
@@ -150,26 +68,6 @@ async def log_client_in(
     return user
 
 
-class NewUser:
-    def __init__(
-        self,
-        user_data: dict[str, Any] | None = None,
-        app: web.Application | None = None,
-    ):
-        self.user_data = user_data
-        self.user = None
-        assert app
-        self.db = get_plugin_storage(app)
-        self.app = app
-
-    async def __aenter__(self) -> UserInfoDict:
-        self.user = await _create_account(self.app, self.user_data)
-        return self.user
-
-    async def __aexit__(self, *args):
-        await self.db.delete_user(self.user)
-
-
 class LoggedUser(NewUser):
     def __init__(self, client: TestClient, user_data=None, *, check_if_succeeds=True):
         super().__init__(user_data, client.app)
@@ -179,7 +77,10 @@ def __init__(self, client: TestClient, user_data=None, *, check_if_succeeds=True
 
     async def __aenter__(self) -> UserInfoDict:
         self.user = await log_client_in(
-            self.client, self.user_data, enable_check=self.enable_check
+            self.client,
+            self.user_data,
+            exit_stack=self.exit_stack,
+            enable_check=self.enable_check,
         )
         return self.user
 
@@ -231,11 +132,12 @@ def __init__(
         self.confirmation = None
         self.trial_days = trial_days
         self.extra_credits_in_usd = extra_credits_in_usd
+        self.db = get_plugin_storage(self.app)
 
     async def __aenter__(self) -> "NewInvitation":
         # creates host user
         assert self.client.app
-        self.user = await _create_user(self.client.app, self.user_data)
+        self.user = await super().__aenter__()
 
         self.confirmation = await create_invitation_token(
             self.db,

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

@@ -0,0 +1,143 @@
+import contextlib
+from datetime import datetime
+from typing import Any, TypedDict
+
+from aiohttp import web
+from models_library.users import UserID
+from simcore_postgres_database.models.users import users as users_table
+from simcore_service_webserver.db.models import UserRole, UserStatus
+from simcore_service_webserver.db.plugin import get_asyncpg_engine
+from simcore_service_webserver.groups import api as groups_service
+from simcore_service_webserver.products.products_service import list_products
+from sqlalchemy.ext.asyncio import AsyncEngine
+
+from .faker_factories import DEFAULT_TEST_PASSWORD, random_user
+from .postgres_tools import insert_and_get_row_lifespan
+
+
+# WARNING: DO NOT use UserDict is already in https://docs.python.org/3/library/collections.html#collections.UserDictclass UserRowDict(TypedDict):
+# NOTE: this is modified dict version of packages/postgres-database/src/simcore_postgres_database/models/users.py for testing purposes
+class _UserInfoDictRequired(TypedDict, total=True):
+    id: int
+    name: str
+    email: str
+    primary_gid: str
+    raw_password: str
+    status: UserStatus
+    role: UserRole
+
+
+class UserInfoDict(_UserInfoDictRequired, total=False):
+    created_at: datetime
+    password_hash: str
+    first_name: str
+    last_name: str
+    phone: str
+
+
+async def _create_user_in_db(
+    sqlalchemy_async_engine: AsyncEngine,
+    exit_stack: contextlib.AsyncExitStack,
+    data: dict | None = None,
+) -> UserInfoDict:
+
+    # create fake
+    data = data or {}
+    data.setdefault("status", UserStatus.ACTIVE.name)
+    data.setdefault("role", UserRole.USER.name)
+    data.setdefault("password", DEFAULT_TEST_PASSWORD)
+
+    raw_password = data["password"]
+
+    # inject in db
+    user = await exit_stack.enter_async_context(
+        insert_and_get_row_lifespan(  # pylint:disable=contextmanager-generator-missing-cleanup
+            sqlalchemy_async_engine,
+            table=users_table,
+            values=random_user(**data),
+            pk_col=users_table.c.id,
+        )
+    )
+    assert "first_name" in user
+    assert "last_name" in user
+
+    return UserInfoDict(
+        # required
+        #  - in db
+        id=user["id"],
+        name=user["name"],
+        email=user["email"],
+        primary_gid=user["primary_gid"],
+        status=(
+            UserStatus(user["status"])
+            if not isinstance(user["status"], UserStatus)
+            else user["status"]
+        ),
+        role=(
+            UserRole(user["role"])
+            if not isinstance(user["role"], UserRole)
+            else user["role"]
+        ),
+        # optional
+        #  - in db
+        created_at=(
+            user["created_at"]
+            if isinstance(user["created_at"], datetime)
+            else datetime.fromisoformat(user["created_at"])
+        ),
+        password_hash=user["password_hash"],
+        first_name=user["first_name"],
+        last_name=user["last_name"],
+        phone=user["phone"],
+        # extras
+        raw_password=raw_password,
+    )
+
+
+async def _register_user_in_default_product(app: web.Application, user_id: UserID):
+    products = list_products(app)
+    assert products
+    product_name = products[0].name
+
+    return await groups_service.auto_add_user_to_product_group(
+        app, user_id, product_name=product_name
+    )
+
+
+async def _create_account_in_db(
+    app: web.Application,
+    exit_stack: contextlib.AsyncExitStack,
+    user_data: dict[str, Any] | None = None,
+) -> UserInfoDict:
+    # users, groups in db
+    user = await _create_user_in_db(
+        get_asyncpg_engine(app), exit_stack=exit_stack, data=user_data
+    )
+
+    # user has default product
+    await _register_user_in_default_product(app, user_id=user["id"])
+    return user
+
+
+class NewUser:
+    def __init__(
+        self,
+        user_data: dict[str, Any] | None = None,
+        app: web.Application | None = None,
+    ):
+        self.user_data = user_data
+        self.user = None
+
+        assert app
+        self.app = app
+
+        self.exit_stack = contextlib.AsyncExitStack()
+
+    async def __aenter__(self) -> UserInfoDict:
+        self.user = await _create_account_in_db(
+            self.app, self.exit_stack, self.user_data
+        )
+        return self.user
+
+    async def __aexit__(self, *args):
+        await self.exit_stack.aclose()

packages/pytest-simcore/src/pytest_simcore/simcore_webserver_groups_fixtures.py

@@ -3,9 +3,9 @@
 # pylint: disable=unused-variable
 """
 
-    Fixtures for groups
+Fixtures for groups
 
-    NOTE: These fixtures are used in integration and unit tests
+NOTE: These fixtures are used in integration and unit tests
 """
 
 
@@ -18,7 +18,7 @@
 from models_library.api_schemas_webserver.groups import GroupGet
 from models_library.groups import GroupsByTypeTuple, StandardGroupCreate
 from models_library.users import UserID
-from pytest_simcore.helpers.webserver_login import NewUser, UserInfoDict
+from pytest_simcore.helpers.webserver_users import NewUser, UserInfoDict
 from simcore_service_webserver.groups._groups_service import (
     add_user_in_group,
     create_standard_group,

services/web/server/tests/conftest.py

@@ -4,6 +4,7 @@
 # pylint: disable=unused-variable
 
 import asyncio
+import contextlib
 import json
 import logging
 import random
@@ -26,7 +27,8 @@
 from pytest_mock import MockerFixture
 from pytest_simcore.helpers.assert_checks import assert_status
 from pytest_simcore.helpers.monkeypatch_envs import EnvVarsDict, setenvs_from_dict
-from pytest_simcore.helpers.webserver_login import LoggedUser, NewUser, UserInfoDict
+from pytest_simcore.helpers.webserver_login import LoggedUser
+from pytest_simcore.helpers.webserver_users import NewUser, UserInfoDict
 from pytest_simcore.simcore_webserver_projects_rest_api import NEW_PROJECT
 from servicelib.aiohttp import status
 from servicelib.common_headers import (
@@ -88,6 +90,13 @@
 ]
 
 
+@pytest.fixture
+async def exit_stack() -> AsyncIterator[contextlib.AsyncExitStack]:
+    """Provides an AsyncExitStack that gets cleaned up after each test"""
+    async with contextlib.AsyncExitStack() as stack:
+        yield stack
+
+
 @pytest.fixture(scope="session")
 def package_dir() -> Path:
     """osparc-simcore installed directory"""