♻️ Maintenance: refactors guest user and save-state logic to improve flaky tests (ITISFoundation#2897)

Author: pcrespov

packages/models-library/src/models_library/projects.py

@@ -10,7 +10,7 @@
 from pydantic import BaseModel, EmailStr, Extra, Field, HttpUrl, constr, validator
 
 from .basic_regex import DATE_RE, UUID_RE
-from .projects_access import AccessRights, GroupID
+from .projects_access import AccessRights, GroupIDStr
 from .projects_nodes import Node
 from .projects_nodes_io import NodeIDStr
 from .projects_state import ProjectState
@@ -121,7 +121,7 @@ class Project(BaseProjectModel):
         examples=["2018-07-01T11:13:43Z"],
         alias="lastChangeDate",
     )
-    access_rights: Dict[GroupID, AccessRights] = Field(
+    access_rights: Dict[GroupIDStr, AccessRights] = Field(
         ...,
         description="object containing the GroupID as key and read/write/execution permissions as value",
         alias="accessRights",

packages/models-library/src/models_library/projects_access.py

@@ -7,7 +7,7 @@
 from pydantic import BaseModel, Extra, Field, constr
 from pydantic.types import PositiveInt
 
-GroupID = constr(regex=r"^\S+$")
+GroupIDStr = constr(regex=r"^\S+$")
 
 
 class AccessEnum(str, Enum):

packages/postgres-database/src/simcore_postgres_database/models/users.py

@@ -4,15 +4,24 @@
     - Users they have a role within the framework that provides
     them different access levels to it
 """
-import itertools
 from enum import Enum
+from functools import total_ordering
 
 import sqlalchemy as sa
 from sqlalchemy.sql import func
 
 from .base import metadata
 
+_USER_ROLE_TO_LEVEL = {
+    "ANONYMOUS": 0,
+    "GUEST": 10,
+    "USER": 20,
+    "TESTER": 30,
+    "ADMIN": 100,
+}
 
+
+@total_ordering
 class UserRole(Enum):
     """SORTED enumeration of user roles
 
@@ -36,11 +45,14 @@ class UserRole(Enum):
     TESTER = "TESTER"
     ADMIN = "ADMIN"
 
-    @classmethod
-    def super_users(cls):
-        return list(itertools.takewhile(lambda e: e != cls.USER, cls))
+    @property
+    def privilege_level(self) -> int:
+        return _USER_ROLE_TO_LEVEL[self.name]
 
-    # TODO: add comparison https://portingguide.readthedocs.io/en/latest/comparisons.html
+    def __lt__(self, other) -> bool:
+        if self.__class__ is other.__class__:
+            return self.privilege_level < other.privilege_level
+        return NotImplemented
 
 
 class UserStatus(Enum):

packages/postgres-database/tests/test_users.py

@@ -0,0 +1,50 @@
+from simcore_postgres_database.models.users import _USER_ROLE_TO_LEVEL, UserRole
+
+
+def test_user_role_to_level_map_in_sync():
+    # If fails, then update _USER_ROLE_TO_LEVEL map
+    assert set(_USER_ROLE_TO_LEVEL.keys()) == set(UserRole.__members__.keys())
+
+
+def test_user_role_comparison():
+
+    assert UserRole.ANONYMOUS < UserRole.ADMIN
+    assert UserRole.GUEST < UserRole.ADMIN
+    assert UserRole.USER < UserRole.ADMIN
+    assert UserRole.TESTER < UserRole.ADMIN
+    assert UserRole.ADMIN == UserRole.ADMIN
+
+    assert UserRole.ANONYMOUS < UserRole.TESTER
+    assert UserRole.GUEST < UserRole.TESTER
+    assert UserRole.USER < UserRole.TESTER
+    assert UserRole.TESTER == UserRole.TESTER
+    assert UserRole.ADMIN > UserRole.TESTER
+
+    assert UserRole.ANONYMOUS < UserRole.USER
+    assert UserRole.GUEST < UserRole.USER
+    assert UserRole.USER == UserRole.USER
+    assert UserRole.TESTER > UserRole.USER
+    assert UserRole.ADMIN > UserRole.USER
+
+    assert UserRole.ANONYMOUS < UserRole.GUEST
+    assert UserRole.GUEST == UserRole.GUEST
+    assert UserRole.USER > UserRole.GUEST
+    assert UserRole.TESTER > UserRole.GUEST
+    assert UserRole.ADMIN > UserRole.GUEST
+
+    assert UserRole.ANONYMOUS == UserRole.ANONYMOUS
+    assert UserRole.GUEST > UserRole.ANONYMOUS
+    assert UserRole.USER > UserRole.ANONYMOUS
+    assert UserRole.TESTER > UserRole.ANONYMOUS
+    assert UserRole.ADMIN > UserRole.ANONYMOUS
+
+    # < and >
+    assert UserRole.TESTER < UserRole.ADMIN
+    assert UserRole.ADMIN > UserRole.TESTER
+
+    # >=, == and <=
+    assert UserRole.TESTER <= UserRole.ADMIN
+    assert UserRole.ADMIN >= UserRole.TESTER
+
+    assert UserRole.ADMIN <= UserRole.ADMIN
+    assert UserRole.ADMIN == UserRole.ADMIN

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

@@ -1,5 +1,5 @@
 import re
-from typing import Any, Dict
+from typing import TypedDict
 
 from aiohttp import web
 from aiohttp.test_utils import TestClient
@@ -12,9 +12,22 @@
 
 from .utils_assert import assert_status
 
-# WARNING: UserDict is already in https://docs.python.org/3/library/collections.html#collections.UserDict
-# TODO: move this to future simcore_service_webserver.users_models.py
-AUserDict = Dict[str, Any]
+
+# 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
+    raw_password: str
+    status: UserStatus
+    role: UserRole
+
+
+class UserInfoDict(_UserInfoDictRequired, total=False):
+    created_ip: int
+    password_hash: str
+
 
 TEST_MARKS = re.compile(r"TEST (\w+):(.*)")
 
@@ -37,7 +50,7 @@ def parse_link(text):
     return URL(link).path
 
 
-async def create_user(db: AsyncpgStorage, data=None) -> AUserDict:
+async def create_user(db: AsyncpgStorage, data=None) -> UserInfoDict:
     data = data or {}
     password = get_random_string(10)
     params = {
@@ -56,7 +69,7 @@ async def create_user(db: AsyncpgStorage, data=None) -> AUserDict:
 
 async def log_client_in(
     client: TestClient, user_data=None, *, enable_check=True
-) -> AUserDict:
+) -> UserInfoDict:
     # creates user directly in db
     db: AsyncpgStorage = get_plugin_storage(client.app)
     cfg: LoginOptions = get_plugin_options(client.app)
@@ -99,7 +112,7 @@ def __init__(self, client, params=None, *, check_if_succeeds=True):
         self.client = client
         self.enable_check = check_if_succeeds
 
-    async def __aenter__(self) -> AUserDict:
+    async def __aenter__(self) -> UserInfoDict:
         self.user = await log_client_in(
             self.client, self.params, enable_check=self.enable_check
         )

services/web/server/src/simcore_service_webserver/catalog_client.py

@@ -4,7 +4,7 @@
 import asyncio
 import logging
 import urllib.parse
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Mapping, Optional
 
 from aiohttp import ClientSession, ClientTimeout, web
 from aiohttp.client_exceptions import (
@@ -60,7 +60,7 @@ async def make_request_and_envelope_response(
     app: web.Application,
     method: str,
     url: URL,
-    headers: Optional[Dict[str, str]] = None,
+    headers: Optional[Mapping[str, str]] = None,
     data: Optional[bytes] = None,
 ) -> web.Response:
     """