fixing tests

Author: pcrespov

services/catalog/tests/unit/with_dbs/test_service_access_rights.py

@@ -5,6 +5,7 @@
 from collections.abc import Callable
 from typing import Any
 
+import pytest
 import simcore_service_catalog.service.access_rights
 from fastapi import FastAPI
 from models_library.groups import GroupID
@@ -31,6 +32,37 @@
 ]
 
 
+@pytest.fixture
+def new_service_metadata_published(user: dict[str, Any]) -> ServiceMetaDataPublished:
+    MOST_UPDATED_EXAMPLE = -1
+    metadata = ServiceMetaDataPublished.model_validate(
+        ServiceMetaDataPublished.model_json_schema()["examples"][MOST_UPDATED_EXAMPLE]
+    )
+    metadata.contact = user["email"]
+    metadata.authors = [
+        Author(name=user["name"], email=user["email"], affiliation=None)
+    ]
+    metadata.version = TypeAdapter(ServiceVersion).validate_python("1.0.11")
+    metadata.icon = None  # Remove icon to test inheritance
+    return metadata
+
+
+@pytest.fixture
+def app_with_repo(
+    sqlalchemy_async_engine: AsyncEngine,
+    target_product: ProductName,
+    mocker: MockerFixture,
+) -> tuple[FastAPI, ServicesRepository]:
+    """Creates FastAPI app with services repository setup."""
+    app = FastAPI()
+    app.state.engine = sqlalchemy_async_engine
+    app.state.settings = mocker.Mock()
+    app.state.default_product_name = target_product
+
+    services_repo = ServicesRepository(app.state.engine)
+    return app, services_repo
+
+
 def test_reduce_access_rights():
     sample = ServiceAccessRightsDB.model_validate(
         {
@@ -86,43 +118,104 @@ def test_reduce_access_rights():
     }
 
 
-async def test_service_upgrade_metadata_inheritance(
-    sqlalchemy_async_engine: AsyncEngine,
-    user: dict[str, Any],
+async def test_service_upgrade_metadata_inheritance_old_service(
     user_groups_ids: list[GroupID],
     target_product: ProductName,
-    other_product: ProductName,
     services_db_tables_injector: Callable,
     create_fake_service_data: CreateFakeServiceDataCallable,
     mocker: MockerFixture,
+    new_service_metadata_published: ServiceMetaDataPublished,
+    app_with_repo: tuple[FastAPI, ServicesRepository],
 ):
+    """Test inheritance behavior when the service is considered old"""
     everyone_gid, user_gid, team_gid = user_groups_ids
+    app, services_repo = app_with_repo
 
-    assert user_gid == user["primary_gid"]
-
-    # Avoids calls to director API
+    # Mock to make the service appear as old
     mocker.patch.object(
         simcore_service_catalog.service.access_rights,
         "_is_old_service",
-        return_value=False,
+        return_value=True,
     )
 
-    # SETUP ---
-    MOST_UPDATED_EXAMPLE = -1
-    new_service_metadata = ServiceMetaDataPublished.model_validate(
-        ServiceMetaDataPublished.model_json_schema()["examples"][MOST_UPDATED_EXAMPLE]
+    # Create latest-release service for testing inheritance
+    latest_release_service, *latest_release_service_access_rights = (
+        create_fake_service_data(
+            new_service_metadata_published.key,
+            "1.0.10",
+            team_access="x",
+            everyone_access=None,
+            product=target_product,
+        )
+    )
+
+    latest_release_service["icon"] = "https://foo/previous_icon.svg"
+    latest_release = (latest_release_service, *latest_release_service_access_rights)
+
+    await services_db_tables_injector([latest_release])
+
+    # DEFAULT policies for old service
+    owner_gid, service_access_rights = (
+        await evaluate_default_service_ownership_and_rights(
+            app, service=new_service_metadata_published, product_name=target_product
+        )
     )
-    new_service_metadata.contact = user["email"]
-    new_service_metadata.authors = [
-        Author(name=user["name"], email=user["email"], affiliation=None)
-    ]
-    new_service_metadata.version = TypeAdapter(ServiceVersion).validate_python("1.0.11")
-    new_service_metadata.icon = None  # Remove icon to test inheritance
 
-    # latest-release
+    # For old services, everyone should have access
+    assert owner_gid == user_gid
+    assert len(service_access_rights) == 2  # Owner + everyone
+    assert {a.gid for a in service_access_rights} == {owner_gid, everyone_gid}
+
+    # Check owner access
+    owner_access = next(a for a in service_access_rights if a.gid == owner_gid)
+    assert owner_access.model_dump(include={"execute_access", "write_access"}) == {
+        "execute_access": True,
+        "write_access": True,
+    }
+
+    # Check everyone access
+    everyone_access = next(a for a in service_access_rights if a.gid == everyone_gid)
+    assert everyone_access.model_dump(include={"execute_access", "write_access"}) == {
+        "execute_access": True,
+        "write_access": False,  # Everyone can execute but not modify
+    }
+
+    # Inheritance policy (both access rights and metadata)
+    inherited_data = await inherit_from_latest_compatible_release(
+        services_repo, service_metadata=new_service_metadata_published
+    )
+
+    # Check metadata inheritance
+    inherited_metadata = inherited_data["metadata_updates"]
+    assert "icon" in inherited_metadata
+    assert inherited_metadata["icon"] == latest_release_service["icon"]
+
+
+async def test_service_upgrade_metadata_inheritance_new_service_multi_product(
+    user_groups_ids: list[GroupID],
+    target_product: ProductName,
+    other_product: ProductName,
+    services_db_tables_injector: Callable,
+    create_fake_service_data: CreateFakeServiceDataCallable,
+    mocker: MockerFixture,
+    new_service_metadata_published: ServiceMetaDataPublished,
+    app_with_repo: tuple[FastAPI, ServicesRepository],
+):
+    """Test inheritance behavior when the service is new and latest version exists in multiple products"""
+    everyone_gid, user_gid, team_gid = user_groups_ids
+    app, services_repo = app_with_repo
+
+    # Avoids calls to director API - service is new
+    mocker.patch.object(
+        simcore_service_catalog.service.access_rights,
+        "_is_old_service",
+        return_value=False,
+    )
+
+    # Create latest-release service
     latest_release_service, *latest_release_service_access_rights = (
         create_fake_service_data(
-            new_service_metadata.key,
+            new_service_metadata_published.key,
             "1.0.10",
             team_access="x",
             everyone_access=None,
@@ -136,7 +229,7 @@ async def test_service_upgrade_metadata_inheritance(
     # latest-release in other product
     _, *latest_release_service_access_rights_in_other_product = (
         create_fake_service_data(
-            new_service_metadata.key,
+            new_service_metadata_published.key,
             latest_release_service["version"],
             team_access="x",
             everyone_access=None,
@@ -149,51 +242,40 @@ async def test_service_upgrade_metadata_inheritance(
         *latest_release_service_access_rights_in_other_product,  # <-- different product
     )
 
-    # we have three versions of the service in the database for which the sorting matters: (1.0.11 should inherit from 1.0.10 not 1.0.9)
+    # Setup multiple versions in database
     await services_db_tables_injector(
         [
             create_fake_service_data(
-                new_service_metadata.key,
+                new_service_metadata_published.key,
                 "1.0.1",
                 team_access=None,
                 everyone_access=None,
                 product=target_product,
             ),
             create_fake_service_data(
-                new_service_metadata.key,
+                new_service_metadata_published.key,
                 "1.0.9",
                 team_access=None,
                 everyone_access=None,
                 product=target_product,
             ),
-            # new release is a patch on released 1.0.X
-            # which were released in two different product
             latest_release,
             latest_release_in_other_product,
         ]
     )
 
-    # ------------
-
-    app = FastAPI()
-    app.state.engine = sqlalchemy_async_engine
-    app.state.settings = mocker.Mock()
-    app.state.default_product_name = target_product
-
-    services_repo = ServicesRepository(app.state.engine)
-
     # DEFAULT policies
     owner_gid, service_access_rights = (
         await evaluate_default_service_ownership_and_rights(
-            app, service=new_service_metadata, product_name=target_product
+            app, service=new_service_metadata_published, product_name=target_product
         )
     )
     assert owner_gid == user_gid
-    assert len(service_access_rights) == 1
+    assert len(service_access_rights) == 1  # Only owner for new service
     assert {a.gid for a in service_access_rights} == {owner_gid}
     assert service_access_rights[0].model_dump() == {
-        "key": new_service_metadata.key,
-        "version": new_service_metadata.version,
+        "key": new_service_metadata_published.key,
+        "version": new_service_metadata_published.version,
         "gid": user_gid,
         "product_name": target_product,
         "execute_access": True,
@@ -203,7 +285,7 @@ async def test_service_upgrade_metadata_inheritance(
 
     # Inheritance policy (both access rights and metadata)
     inherited_data = await inherit_from_latest_compatible_release(
-        services_repo, service_metadata=new_service_metadata
+        services_repo, service_metadata=new_service_metadata_published
     )
 
     # Check access rights inheritance