tests groups 2 passing

Author: pcrespov

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

@@ -10,7 +10,7 @@
 
 
 from collections.abc import AsyncIterator
-from typing import Any, Protocol
+from typing import Any
 
 import pytest
 from aiohttp import web
@@ -19,41 +19,25 @@
 from models_library.groups import GroupsByTypeTuple
 from models_library.users import UserID
 from pytest_simcore.helpers.webserver_login import NewUser, UserInfoDict
-from simcore_service_webserver.groups import _groups_db
 from simcore_service_webserver.groups._groups_api import (
     add_user_in_group,
+    create_organization,
     delete_organization,
     list_user_groups_with_read_access,
 )
 
 
-def _to_group_get_json(group, access_rights) -> dict[str, Any]:
+def _groupget_model_dump(group, access_rights) -> dict[str, Any]:
     return GroupGet.from_model(group, access_rights).model_dump(mode="json")
 
 
-#
-# FACTORY FIXTURES
-#
-
-
-class CreateUserGroupCallable(Protocol):
-    async def __call__(
-        self, app: web.Application, user_id: UserID, new_group: dict
-    ) -> dict[str, Any]:
-        ...
-
-
-@pytest.fixture
-def create_user_group() -> CreateUserGroupCallable:
-    async def _create(
-        app: web.Application, user_id: UserID, new_group: dict
-    ) -> dict[str, Any]:
-        group, access_rights = await _groups_db.create_user_group(
-            app, user_id=user_id, new_group=new_group
-        )
-        return _to_group_get_json(group=group, access_rights=access_rights)
-
-    return _create
+async def _create_organization(
+    app: web.Application, user_id: UserID, new_group: dict
+) -> dict[str, Any]:
+    group, access_rights = await create_organization(
+        app, user_id=user_id, new_group_values=new_group
+    )
+    return _groupget_model_dump(group=group, access_rights=access_rights)
 
 
 #
@@ -62,36 +46,13 @@ async def _create(
 
 
 @pytest.fixture
-async def logged_user_groups_by_type(
-    client: TestClient, logged_user: UserInfoDict
-) -> GroupsByTypeTuple:
-    assert client.app
-
-    groups_by_type = await list_user_groups_with_read_access(
-        client.app, user_id=logged_user["id"]
-    )
-    assert groups_by_type.primary
-    assert groups_by_type.everyone
-    return groups_by_type
-
-
-@pytest.fixture
-async def primary_group(
-    client: TestClient,
-    logged_user_groups_by_type: GroupsByTypeTuple,
-) -> dict[str, Any]:
-    assert client.app
-    assert logged_user_groups_by_type.primary
-    return _to_group_get_json(*logged_user_groups_by_type.primary)
-
-
-@pytest.fixture
-async def standard_groups(
+async def standard_groups_owner(
     client: TestClient,
     logged_user: UserInfoDict,
-    logged_user_groups_by_type: GroupsByTypeTuple,
-    create_user_group: CreateUserGroupCallable,
-) -> AsyncIterator[list[dict[str, Any]]]:
+) -> AsyncIterator[UserInfoDict]:
+    """
+    standard_groups_owner creates TWO organizations and adds logged_user in them
+    """
 
     assert client.app
     # create a separate account to own standard groups
@@ -102,8 +63,9 @@ async def standard_groups(
         },
         client.app,
     ) as owner_user:
+
         # creates two groups
-        sparc_group = await create_user_group(
+        sparc_group = await _create_organization(
             app=client.app,
             user_id=owner_user["id"],
             new_group={
@@ -113,7 +75,7 @@ async def standard_groups(
                 "inclusionRules": {"email": r"@(sparc)+\.(io|com)$"},
             },
         )
-        team_black_group = await create_user_group(
+        team_black_group = await _create_organization(
             app=client.app,
             user_id=owner_user["id"],
             new_group={
@@ -141,11 +103,7 @@ async def standard_groups(
             new_user_email=logged_user["email"],
         )
 
-        standard_groups = [
-            _to_group_get_json(*sg) for sg in logged_user_groups_by_type.standard
-        ]
-
-        yield standard_groups
+        yield owner_user
 
         # clean groups
         await delete_organization(
@@ -157,11 +115,41 @@ async def standard_groups(
 
 
 @pytest.fixture
-async def all_group(
-    client: TestClient,
+async def logged_user_groups_by_type(
+    client: TestClient, logged_user: UserInfoDict, standard_groups_owner: UserInfoDict
+) -> GroupsByTypeTuple:
+    assert client.app
+
+    assert logged_user["id"] != standard_groups_owner["id"]
+
+    groups_by_type = await list_user_groups_with_read_access(
+        client.app, user_id=logged_user["id"]
+    )
+    assert groups_by_type.primary
+    assert groups_by_type.everyone
+    return groups_by_type
+
+
+@pytest.fixture
+def primary_group(
     logged_user_groups_by_type: GroupsByTypeTuple,
 ) -> dict[str, Any]:
-    assert client.app
-    assert logged_user_groups_by_type.everyone
+    """`logged_user`'s primary group"""
+    assert logged_user_groups_by_type.primary
+    return _groupget_model_dump(*logged_user_groups_by_type.primary)
 
-    return _to_group_get_json(*logged_user_groups_by_type.everyone)
+
+@pytest.fixture
+def standard_groups(
+    logged_user_groups_by_type: GroupsByTypeTuple,
+) -> list[dict[str, Any]]:
+    """owned by `standard_groups_owner` and shared with `logged_user`"""
+    return [_groupget_model_dump(*sg) for sg in logged_user_groups_by_type.standard]
+
+
+@pytest.fixture
+def all_group(
+    logged_user_groups_by_type: GroupsByTypeTuple,
+) -> dict[str, Any]:
+    assert logged_user_groups_by_type.everyone
+    return _groupget_model_dump(*logged_user_groups_by_type.everyone)

services/web/server/tests/unit/with_dbs/01/groups/test_groups_2.py

@@ -0,0 +1,137 @@
+# pylint: disable=redefined-outer-name
+# pylint: disable=too-many-arguments
+# pylint: disable=too-many-statements
+# pylint: disable=unused-argument
+# pylint: disable=unused-variable
+
+
+import operator
+
+import pytest
+from aiohttp.test_utils import TestClient
+from models_library.api_schemas_webserver.groups import GroupGet, MyGroupsGet
+from pydantic import TypeAdapter
+from pytest_simcore.helpers.assert_checks import assert_status
+from pytest_simcore.helpers.webserver_login import UserInfoDict
+from pytest_simcore.helpers.webserver_parametrizations import (
+    ExpectedResponse,
+    standard_role_response,
+)
+from servicelib.aiohttp import status
+from servicelib.aiohttp.application import create_safe_application
+from simcore_postgres_database.models.users import UserRole
+from simcore_service_webserver._meta import API_VTAG
+from simcore_service_webserver.application_settings import setup_settings
+from simcore_service_webserver.db.plugin import setup_db
+from simcore_service_webserver.groups.plugin import setup_groups
+from simcore_service_webserver.login.plugin import setup_login
+from simcore_service_webserver.rest.plugin import setup_rest
+from simcore_service_webserver.security.plugin import setup_security
+from simcore_service_webserver.session.plugin import setup_session
+from simcore_service_webserver.users.plugin import setup_users
+
+
+@pytest.fixture
+def client(
+    event_loop,
+    aiohttp_client,
+    app_environment,
+    postgres_db,
+) -> TestClient:
+    app = create_safe_application()
+
+    setup_settings(app)
+    setup_db(app)
+    setup_session(app)
+    setup_security(app)
+    setup_rest(app)
+    setup_login(app)
+    setup_users(app)
+    setup_groups(app)
+
+    return event_loop.run_until_complete(aiohttp_client(app))
+
+
+@pytest.mark.parametrize(*standard_role_response(), ids=str)
+async def test_list_groups_access_rights(
+    client: TestClient,
+    logged_user: UserInfoDict,
+    user_role: UserRole,
+    expected: ExpectedResponse,
+    primary_group: dict[str, str],
+    standard_groups: list[dict[str, str]],
+    all_group: dict[str, str],
+):
+    assert client.app
+    url = client.app.router["list_groups"].url_for()
+    assert f"{url}" == f"/{API_VTAG}/groups"
+
+    response = await client.get(f"{url}")
+    await assert_status(
+        response, expected.ok if user_role != UserRole.GUEST else status.HTTP_200_OK
+    )
+
+
+@pytest.mark.parametrize("user_role", [UserRole.USER])
+async def test_list_user_groups(
+    client: TestClient,
+    user_role: UserRole,
+    standard_groups_owner: UserInfoDict,
+    logged_user: UserInfoDict,
+    primary_group: dict[str, str],
+    standard_groups: list[dict[str, str]],
+    all_group: dict[str, str],
+):
+    assert client.app
+    assert logged_user["id"] != standard_groups_owner["id"]
+    assert logged_user["role"] == user_role.value
+
+    # List all groups (organizations, primary, everyone and products) I belong to
+    url = client.app.router["list_groups"].url_for()
+    assert f"{url}" == f"/{API_VTAG}/groups"
+
+    response = await client.get(f"{url}")
+    data, error = await assert_status(response, status.HTTP_200_OK)
+
+    my_groups = MyGroupsGet.model_validate(data)
+    assert not error
+
+    assert my_groups.me.model_dump() == primary_group
+    assert my_groups.all.model_dump() == all_group
+
+    assert my_groups.organizations
+    assert len(my_groups.organizations) == len(standard_groups)
+
+    by_gid = operator.itemgetter("gid")
+    assert sorted(
+        TypeAdapter(list[GroupGet]).dump_python(my_groups.organizations, mode="json"),
+        key=by_gid,
+    ) == sorted(standard_groups, key=by_gid)
+
+    for group in standard_groups:
+        # try to delete a group
+        url = client.app.router["delete_group"].url_for(gid=f"{group['gid']}")
+        response = await client.delete(f"{url}")
+        await assert_status(response, status.HTTP_403_FORBIDDEN)
+
+        # try to add some user in the group
+        url = client.app.router["add_group_user"].url_for(gid=f"{group['gid']}")
+        response = await client.post(f"{url}", json={"uid": logged_user["id"]})
+        await assert_status(response, status.HTTP_403_FORBIDDEN)
+
+        # try to modify the user in the group
+        url = client.app.router["update_group_user"].url_for(
+            gid=f"{group['gid']}", uid=f"{logged_user['id']}"
+        )
+        response = await client.patch(
+            f"{url}",
+            json={"accessRights": {"read": True, "write": True, "delete": True}},
+        )
+        await assert_status(response, status.HTTP_403_FORBIDDEN)
+
+        # try to remove the user from the group
+        url = client.app.router["delete_group_user"].url_for(
+            gid=f"{group['gid']}", uid=f"{logged_user['id']}"
+        )
+        response = await client.delete(f"{url}")
+        await assert_status(response, status.HTTP_403_FORBIDDEN)

services/web/server/tests/unit/with_dbs/01/test_groups.py

@@ -104,71 +104,6 @@ def _assert__group_user(
     assert "gid" in actual_user
 
 
-@pytest.mark.parametrize(*standard_role_response(), ids=str)
-async def test_list_groups(
-    client: TestClient,
-    logged_user: UserInfoDict,
-    user_role: UserRole,
-    expected: ExpectedResponse,
-    primary_group: dict[str, str],
-    standard_groups: list[dict[str, str]],
-    all_group: dict[str, str],
-):
-    assert client.app
-    url = client.app.router["list_groups"].url_for()
-    assert f"{url}" == f"/{API_VTAG}/groups"
-
-    response = await client.get(f"{url}")
-    data, error = await assert_status(
-        response, expected.ok if user_role != UserRole.GUEST else status.HTTP_200_OK
-    )
-
-    if not error:
-        assert isinstance(data, dict)
-
-        assert "me" in data
-        _assert_group(data["me"])
-        assert data["me"] == primary_group
-
-        assert "organizations" in data
-        assert isinstance(data["organizations"], list)
-        for group in data["organizations"]:
-            _assert_group(group)
-        assert data["organizations"] == standard_groups
-
-        assert "all" in data
-        _assert_group(data["all"])
-        assert data["all"] == all_group
-
-        for group in standard_groups:
-            # try to delete a group
-            url = client.app.router["delete_group"].url_for(gid=f"{group['gid']}")
-            response = await client.delete(f"{url}")
-            await assert_status(response, status.HTTP_403_FORBIDDEN)
-
-            # try to add some user in the group
-            url = client.app.router["add_group_user"].url_for(gid=f"{group['gid']}")
-            response = await client.post(f"{url}", json={"uid": logged_user["id"]})
-            await assert_status(response, status.HTTP_403_FORBIDDEN)
-
-            # try to modify the user in the group
-            url = client.app.router["update_group_user"].url_for(
-                gid=f"{group['gid']}", uid=f"{logged_user['id']}"
-            )
-            response = await client.patch(
-                f"{url}",
-                json={"accessRights": {"read": True, "write": True, "delete": True}},
-            )
-            await assert_status(response, status.HTTP_403_FORBIDDEN)
-
-            # try to remove the user from the group
-            url = client.app.router["delete_group_user"].url_for(
-                gid=f"{group['gid']}", uid=f"{logged_user['id']}"
-            )
-            response = await client.delete(f"{url}")
-            await assert_status(response, status.HTTP_403_FORBIDDEN)
-
-
 @pytest.mark.parametrize(*standard_role_response())
 async def test_group_creation_workflow(
     client: TestClient,