Merge branch 'master' into 8065-instrument-asyncpg

Author: bisgaard-itis

api/specs/web-server/_auth.py

@@ -15,7 +15,7 @@
 from models_library.rest_error import EnvelopedError, Log
 from pydantic import BaseModel, Field, confloat
 from simcore_service_webserver._meta import API_VTAG
-from simcore_service_webserver.login._controller.rest.auth import (
+from simcore_service_webserver.login._controller.rest.auth_schemas import (
     LoginBody,
     LoginNextPage,
     LoginTwoFactorAuthBody,
@@ -30,7 +30,7 @@
     PhoneConfirmationBody,
     ResetPasswordConfirmation,
 )
-from simcore_service_webserver.login._controller.rest.registration import (
+from simcore_service_webserver.login._controller.rest.registration_schemas import (
     InvitationCheck,
     InvitationInfo,
     RegisterBody,

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

@@ -5,11 +5,11 @@
 
 from aiohttp.test_utils import TestClient
 from servicelib.aiohttp import status
-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 (
     get_plugin_storage,
 )
+from simcore_service_webserver.login.constants import MSG_LOGGED_IN
 from simcore_service_webserver.security import security_service
 from yarl import URL
 

services/director-v2/src/simcore_service_director_v2/api/routes/computations.py

@@ -97,8 +97,8 @@ async def _check_pipeline_not_running_or_raise_409(
     computation: ComputationCreate,
 ) -> None:
     with contextlib.suppress(ComputationalRunNotFoundError):
-        last_run = await comp_runs_repo.get(
-            user_id=computation.user_id, project_id=computation.project_id
+        last_run = await comp_runs_repo.get_latest_run_by_project(
+            project_id=computation.project_id
         )
         pipeline_state = last_run.result
 
@@ -367,8 +367,8 @@ async def create_or_update_or_start_computation(  # noqa: PLR0913 # pylint: disa
         last_run: CompRunsAtDB | None = None
         pipeline_state = RunningState.NOT_STARTED
         with contextlib.suppress(ComputationalRunNotFoundError):
-            last_run = await comp_runs_repo.get(
-                user_id=computation.user_id, project_id=computation.project_id
+            last_run = await comp_runs_repo.get_latest_run_by_project(
+                project_id=computation.project_id
             )
             pipeline_state = last_run.result
 
@@ -467,7 +467,7 @@ async def get_computation(
     last_run: CompRunsAtDB | None = None
     pipeline_state = RunningState.NOT_STARTED
     with contextlib.suppress(ComputationalRunNotFoundError):
-        last_run = await comp_runs_repo.get(user_id=user_id, project_id=project_id)
+        last_run = await comp_runs_repo.get_latest_run_by_project(project_id=project_id)
         pipeline_state = last_run.result
 
     _logger.debug(
@@ -542,8 +542,8 @@ async def stop_computation(
         last_run: CompRunsAtDB | None = None
         pipeline_state = RunningState.UNKNOWN
         with contextlib.suppress(ComputationalRunNotFoundError):
-            last_run = await comp_runs_repo.get(
-                user_id=computation_stop.user_id, project_id=project_id
+            last_run = await comp_runs_repo.get_latest_run_by_project(
+                project_id=project_id
             )
             pipeline_state = last_run.result
             if utils.is_pipeline_running(last_run.result):
@@ -601,8 +601,8 @@ async def delete_computation(
         # check if current state allow to stop the computation
         pipeline_state = RunningState.UNKNOWN
         with contextlib.suppress(ComputationalRunNotFoundError):
-            last_run = await comp_runs_repo.get(
-                user_id=computation_stop.user_id, project_id=project_id
+            last_run = await comp_runs_repo.get_latest_run_by_project(
+                project_id=project_id
             )
             pipeline_state = last_run.result
         if utils.is_pipeline_running(pipeline_state):
@@ -636,8 +636,8 @@ def return_last_value(retry_state: Any) -> Any:
                 before_sleep=before_sleep_log(_logger, logging.INFO),
             )
             async def check_pipeline_stopped() -> bool:
-                last_run = await comp_runs_repo.get(
-                    user_id=computation_stop.user_id, project_id=project_id
+                last_run = await comp_runs_repo.get_latest_run_by_project(
+                    project_id=project_id
                 )
                 pipeline_state = last_run.result
                 return utils.is_pipeline_stopped(pipeline_state)

services/director-v2/src/simcore_service_director_v2/modules/db/repositories/comp_runs.py

@@ -145,6 +145,22 @@ async def get(
                 raise ComputationalRunNotFoundError
             return CompRunsAtDB.model_validate(row)
 
+    async def get_latest_run_by_project(
+        self,
+        project_id: ProjectID,
+    ) -> CompRunsAtDB:
+        async with pass_or_acquire_connection(self.db_engine) as conn:
+            result = await conn.execute(
+                sa.select(comp_runs)
+                .where(comp_runs.c.project_uuid == f"{project_id}")
+                .order_by(desc(comp_runs.c.run_id))
+                .limit(1)
+            )
+            row = result.one_or_none()
+            if not row:
+                raise ComputationalRunNotFoundError
+            return CompRunsAtDB.model_validate(row)
+
     async def list_(
         self,
         *,

services/director-v2/tests/unit/with_dbs/comp_scheduler/test_db_repositories_comp_runs.py

@@ -1245,3 +1245,78 @@ async def test_list_group_by_collection_run_id_state_priority_precedence(
     assert len(items) == 1
     collection_item = items[0]
     assert collection_item.state == RunningState.FAILED
+
+
+async def test_get_latest_run_by_project(
+    sqlalchemy_async_engine: AsyncEngine,
+    run_metadata: RunMetadataDict,
+    faker: Faker,
+    publish_project: Callable[[], Awaitable[PublishedProject]],
+    create_registered_user: Callable[..., dict[str, Any]],
+    with_product: dict[str, Any],
+):
+    """Test that get() with user_id=None retrieves the latest run regardless of user"""
+    published_project = await publish_project()
+
+    # Create a second user
+    second_user = create_registered_user()
+
+    # Create comp runs for the original user
+    comp_run_user1_iter1 = await CompRunsRepository(sqlalchemy_async_engine).create(
+        user_id=published_project.user["id"],
+        project_id=published_project.project.uuid,
+        iteration=None,
+        metadata=run_metadata,
+        use_on_demand_clusters=faker.pybool(),
+        dag_adjacency_list=published_project.pipeline.dag_adjacency_list,
+        collection_run_id=CollectionRunID(faker.uuid4()),
+    )
+
+    # Create comp runs for the second user (this should increment iteration)
+    comp_run_user2_iter2 = await CompRunsRepository(sqlalchemy_async_engine).create(
+        user_id=second_user["id"],
+        project_id=published_project.project.uuid,
+        iteration=None,
+        metadata=run_metadata,
+        use_on_demand_clusters=faker.pybool(),
+        dag_adjacency_list=published_project.pipeline.dag_adjacency_list,
+        collection_run_id=CollectionRunID(faker.uuid4()),
+    )
+
+    # Create another run for the first user (should be iteration 3)
+    comp_run_user1_iter3 = await CompRunsRepository(sqlalchemy_async_engine).create(
+        user_id=published_project.user["id"],
+        project_id=published_project.project.uuid,
+        iteration=None,
+        metadata=run_metadata,
+        use_on_demand_clusters=faker.pybool(),
+        dag_adjacency_list=published_project.pipeline.dag_adjacency_list,
+        collection_run_id=CollectionRunID(faker.uuid4()),
+    )
+
+    # Verify iterations are correct
+    assert comp_run_user1_iter1.iteration == 1
+    assert comp_run_user2_iter2.iteration == 1
+    assert comp_run_user1_iter3.iteration == 2
+
+    # Test get with user_id=None should return the latest run (highest iteration)
+    latest_run = await CompRunsRepository(
+        sqlalchemy_async_engine
+    ).get_latest_run_by_project(
+        project_id=published_project.project.uuid,
+    )
+    assert latest_run == comp_run_user1_iter3
+    assert latest_run.iteration == 2
+
+    # Test get with specific user_id still works
+    user1_latest = await CompRunsRepository(sqlalchemy_async_engine).get(
+        user_id=published_project.user["id"],
+        project_id=published_project.project.uuid,
+    )
+    assert user1_latest == comp_run_user1_iter3
+
+    user2_latest = await CompRunsRepository(sqlalchemy_async_engine).get(
+        user_id=second_user["id"],
+        project_id=published_project.project.uuid,
+    )
+    assert user2_latest == comp_run_user2_iter2

services/web/server/src/simcore_service_webserver/login/__init__.py

@@ -1,5 +0,0 @@
-""" webserver's login subsystem
-
-
-    This sub-package is based on aiohttp-login https://github.com/imbolc/aiohttp-login
-"""

services/web/server/src/simcore_service_webserver/login/_auth_service.py

@@ -12,8 +12,8 @@
 from ..products.models import Product
 from ..security import security_service
 from . import _login_service
-from ._constants import MSG_UNKNOWN_EMAIL, MSG_WRONG_PASSWORD
 from ._login_repository_legacy import AsyncpgStorage, get_plugin_storage
+from .constants import MSG_UNKNOWN_EMAIL, MSG_WRONG_PASSWORD
 
 
 async def get_user_by_email(app: web.Application, *, email: str) -> dict[str, Any]:

services/web/server/src/simcore_service_webserver/login/_controller/rest/_rest_exceptions.py

@@ -8,7 +8,7 @@
     to_exceptions_handlers_map,
 )
 from ....users.exceptions import AlreadyPreRegisteredError
-from ..._constants import MSG_2FA_UNAVAILABLE
+from ...constants import MSG_2FA_UNAVAILABLE
 from ...errors import SendingVerificationEmailError, SendingVerificationSmsError
 
 _TO_HTTP_ERROR_MAP: ExceptionToHttpErrorMap = {

services/web/server/src/simcore_service_webserver/login/_controller/rest/auth.py

@@ -3,8 +3,7 @@
 from aiohttp import web
 from aiohttp.web import RouteTableDef
 from models_library.authentification import TwoFactorAuthentificationMethod
-from models_library.emails import LowerCaseEmailStr
-from pydantic import BaseModel, Field, PositiveInt, SecretStr, TypeAdapter
+from pydantic import TypeAdapter
 from servicelib.aiohttp import status
 from servicelib.aiohttp.requests_validation import parse_request_body_as
 from servicelib.logging_utils import get_log_record_extra, log_context
@@ -21,9 +20,9 @@
     session_access_required,
 )
 from ....users import preferences_api as user_preferences_api
-from ....utils_aiohttp import NextPage
+from ....web_utils import envelope_response, flash_response
 from ... import _auth_service, _login_service, _security_service, _twofa_service
-from ..._constants import (
+from ...constants import (
     CODE_2FA_EMAIL_CODE_REQUIRED,
     CODE_2FA_SMS_CODE_REQUIRED,
     CODE_PHONE_NUMBER_REQUIRED,
@@ -37,31 +36,17 @@
     MSG_WRONG_2FA_CODE__EXPIRED,
     MSG_WRONG_2FA_CODE__INVALID,
 )
-from ..._models import InputSchema
 from ...decorators import login_required
 from ...settings import LoginSettingsForProduct, get_plugin_settings
 from ._rest_exceptions import handle_rest_requests_exceptions
+from .auth_schemas import LoginBody, LoginTwoFactorAuthBody, LogoutBody
 
 log = logging.getLogger(__name__)
 
 
 routes = RouteTableDef()
 
 
-class LoginBody(InputSchema):
-    email: LowerCaseEmailStr
-    password: SecretStr
-
-
-class CodePageParams(BaseModel):
-    message: str
-    expiration_2fa: PositiveInt | None = None
-    next_url: str | None = None
-
-
-class LoginNextPage(NextPage[CodePageParams]): ...
-
-
 @routes.post(f"/{API_VTAG}/auth/login", name="auth_login")
 @on_success_grant_session_access_to(
     name="auth_register_phone",
@@ -134,7 +119,7 @@ async def login(request: web.Request):
         user_2fa_authentification_method == TwoFactorAuthentificationMethod.SMS
         and not user["phone"]
     ):
-        return _login_service.envelope_response(
+        return envelope_response(
             # LoginNextPage
             {
                 "name": CODE_PHONE_NUMBER_REQUIRED,
@@ -169,7 +154,7 @@ async def login(request: web.Request):
             user_id=user["id"],
         )
 
-        return _login_service.envelope_response(
+        return envelope_response(
             # LoginNextPage
             {
                 "name": CODE_2FA_SMS_CODE_REQUIRED,
@@ -196,7 +181,7 @@ async def login(request: web.Request):
         product=product,
         user_id=user["id"],
     )
-    return _login_service.envelope_response(
+    return envelope_response(
         {
             "name": CODE_2FA_EMAIL_CODE_REQUIRED,
             "parameters": {
@@ -208,11 +193,6 @@ async def login(request: web.Request):
     )
 
 
-class LoginTwoFactorAuthBody(InputSchema):
-    email: LowerCaseEmailStr
-    code: SecretStr
-
-
 @routes.post(f"/{API_VTAG}/auth/validate-code-login", name="auth_login_2fa")
 @session_access_required(
     "auth_login_2fa",
@@ -259,12 +239,6 @@ async def login_2fa(request: web.Request):
     return await _security_service.login_granted_response(request, user=dict(user))
 
 
-class LogoutBody(InputSchema):
-    client_session_id: str | None = Field(
-        None, examples=["5ac57685-c40f-448f-8711-70be1936fd63"]
-    )
-
-
 @routes.post(f"/{API_VTAG}/auth/logout", name="auth_logout")
 @login_required
 @handle_rest_requests_exceptions
@@ -282,7 +256,7 @@ async def logout(request: web.Request) -> web.Response:
         f"{logout_.client_session_id=}",
         extra=get_log_record_extra(user_id=user_id),
     ):
-        response = _login_service.flash_response(MSG_LOGGED_OUT, "INFO")
+        response = flash_response(MSG_LOGGED_OUT, "INFO")
         await _login_service.notify_user_logout(
             request.app, user_id, logout_.client_session_id
         )

services/web/server/src/simcore_service_webserver/login/_controller/rest/auth_schemas.py

@@ -0,0 +1,33 @@
+from aiohttp.web import RouteTableDef
+from models_library.emails import LowerCaseEmailStr
+from pydantic import BaseModel, Field, PositiveInt, SecretStr
+
+from ....utils_aiohttp import NextPage
+from ..._models import InputSchema
+
+routes = RouteTableDef()
+
+
+class LoginBody(InputSchema):
+    email: LowerCaseEmailStr
+    password: SecretStr
+
+
+class CodePageParams(BaseModel):
+    message: str
+    expiration_2fa: PositiveInt | None = None
+    next_url: str | None = None
+
+
+class LoginNextPage(NextPage[CodePageParams]): ...
+
+
+class LoginTwoFactorAuthBody(InputSchema):
+    email: LowerCaseEmailStr
+    code: SecretStr
+
+
+class LogoutBody(InputSchema):
+    client_session_id: str | None = Field(
+        None, examples=["5ac57685-c40f-448f-8711-70be1936fd63"]
+    )