tests

Author: pcrespov

packages/postgres-database/src/simcore_postgres_database/utils_tags.py

@@ -287,7 +287,9 @@ async def create_or_update_access_rights(
         write: bool,
         delete: bool,
     ) -> TagAccessRightsDict:
-
+        # NOTE that there is no reference to the caller which implies
+        # that we assume that the rigths to create or update these access_rights
+        # have been checked (via has_access_rights) before calling this function
         async with transaction_context(self.engine, connection) as conn:
             result = await conn.execute(
                 upsert_tags_access_rights_stmt(
@@ -316,6 +318,9 @@ async def delete_access_rights(
         tag_id: int,
         group_id: int,
     ) -> bool:
+        # NOTE that there is no reference to the caller which implies
+        # that we assume that the rigths to delete these access_rights
+        # have been checked (via has_access_rights) before calling this function
         async with transaction_context(self.engine, connection) as conn:
             deleted: bool = await conn.scalar(
                 delete_tag_access_rights_stmt(tag_id=tag_id, group_id=group_id)

packages/postgres-database/tests/test_utils_tags.py

@@ -630,9 +630,6 @@ async def test_tags_repo_create(
         name="T1",
         description="my first tag",
         color="pink",
-        read=True,
-        write=True,
-        delete=True,
     )
     assert tag_1 == {
         "id": 1,
@@ -654,6 +651,108 @@ async def test_tags_repo_create(
         == user.primary_gid
     )
 
+    # Checks defaults to full ownership
+    assert await tags_repo.has_access_rights(
+        caller_id=user.id,
+        tag_id=tag_1["id"],
+        read=True,
+        write=True,
+        delete=True,
+    )
+
+
+async def test_tags_repo_access_rights(
+    asyncpg_engine: AsyncEngine,
+    user: RowProxy,
+    group: RowProxy,
+    other_user: RowProxy,
+):
+    tags_repo = TagsRepo(asyncpg_engine)
+    tag = await tags_repo.create(
+        user_id=user.id,
+        name="T1",
+        description="my first tag",
+        color="pink",
+    )
+
+    # check ownership
+    tag_accesses = await tags_repo.list_access_rights(tag_id=tag["id"])
+    assert len(tag_accesses) == 1
+    user_access = tag_accesses[0]
+    assert user_access == {
+        "group_id": user.primary_gid,
+        "tag_id": tag["id"],
+        "read": True,
+        "write": True,
+        "delete": True,
+    }
+
+    assert await tags_repo.has_access_rights(
+        caller_id=user.id,
+        tag_id=tag["id"],
+        read=True,
+        write=True,
+        delete=True,
+    )
+
+    # CREATE access for other_user
+    other_user_access = await tags_repo.create_or_update_access_rights(
+        tag_id=tag["id"],
+        group_id=other_user.primary_gid,
+        read=True,
+        write=False,
+        delete=False,
+    )
+
+    assert not await tags_repo.has_access_rights(
+        caller_id=other_user.id,
+        tag_id=tag["id"],
+        read=user_access["read"],
+        write=user_access["write"],
+        delete=user_access["delete"],
+    )
+
+    assert await tags_repo.has_access_rights(
+        caller_id=other_user.id,
+        tag_id=tag["id"],
+        read=other_user_access["read"],
+        write=other_user_access["write"],
+        delete=other_user_access["delete"],
+    )
+
+    tag_accesses = await tags_repo.list_access_rights(tag_id=tag["id"])
+    assert len(tag_accesses) == 2
+
+    # UPDATE access
+    updated_access = await tags_repo.create_or_update_access_rights(
+        tag_id=tag["id"],
+        group_id=other_user.primary_gid,
+        read=False,  # <--
+        write=False,
+        delete=False,
+    )
+    assert updated_access != other_user_access
+
+    # checks partial
+    assert await tags_repo.has_access_rights(
+        caller_id=other_user.id,
+        tag_id=tag["id"],
+        read=False,
+    )
+
+    assert not await tags_repo.has_access_rights(
+        caller_id=other_user.id, tag_id=tag["id"], write=True
+    )
+
+    # DELETE access to other-user
+    await tags_repo.delete_access_rights(
+        tag_id=tag["id"],
+        group_id=other_user.primary_gid,
+    )
+
+    tag_accesses = await tags_repo.list_access_rights(tag_id=tag["id"])
+    assert len(tag_accesses) == 1
+
 
 def test_building_tags_sql_statements():
     def _check(func_smt, **kwargs):

services/web/server/src/simcore_service_webserver/api/v0/openapi.yaml

@@ -14069,22 +14069,12 @@ components:
         delete:
           type: boolean
           title: Delete
-        created:
-          type: string
-          format: date-time
-          title: Created
-        modified:
-          type: string
-          format: date-time
-          title: Modified
       type: object
       required:
       - gid
       - read
       - write
       - delete
-      - created
-      - modified
       title: TagGroupGet
     TagUpdate:
       properties:

services/web/server/src/simcore_service_webserver/tags/_service.py

@@ -20,6 +20,7 @@
 async def create_tag(
     app: web.Application, user_id: UserID, new_tag: TagCreate
 ) -> TagGet:
+    """Creates tag and user_id takes ownership"""
     engine: AsyncEngine = get_async_engine(app)
 
     repo = TagsRepo(engine)
@@ -30,7 +31,7 @@ async def create_tag(
         delete=True,
         **new_tag.model_dump(exclude_unset=True),
     )
-    return TagGet.from_db(tag)
+    return TagGet.from_model(tag)
 
 
 async def list_tags(
@@ -40,7 +41,7 @@ async def list_tags(
     engine: AsyncEngine = get_async_engine(app)
     repo = TagsRepo(engine)
     tags = await repo.list_all(user_id=user_id)
-    return [TagGet.from_db(t) for t in tags]
+    return [TagGet.from_model(t) for t in tags]
 
 
 async def update_tag(
@@ -54,7 +55,7 @@ async def update_tag(
         tag_id=tag_id,
         **tag_updates.model_dump(exclude_unset=True),
     )
-    return TagGet.from_db(tag)
+    return TagGet.from_model(tag)
 
 
 async def delete_tag(app: web.Application, user_id: UserID, tag_id: IdInt):

services/web/server/src/simcore_service_webserver/tags/schemas.py

@@ -55,7 +55,7 @@ class TagGet(OutputSchema):
     access_rights: TagAccessRights = Field(..., alias="accessRights")
 
     @classmethod
-    def from_db(cls, tag: TagDict) -> "TagGet":
+    def from_model(cls, tag: TagDict) -> Self:
         # NOTE: cls(access_rights=tag, **tag) would also work because of Config
         return cls(
             id=tag["id"],