Modify multiple functions that relates to superseding score set and add some related tests.

Author: EstelleDa

src/mavedb/lib/score_sets.py

@@ -8,7 +8,7 @@
 import numpy as np
 import pandas as pd
 from pandas.testing import assert_index_equal
-from sqlalchemy import Integer, cast, func, or_, select
+from sqlalchemy import and_, Integer, cast, func, or_, select
 from sqlalchemy.orm import Session, aliased, contains_eager, joinedload, selectinload
 
 from mavedb.lib.exceptions import ValidationError
@@ -74,13 +74,21 @@ def search_score_sets(db: Session, owner_or_contributor: Optional["UserData"], s
     query = db.query(ScoreSet)  # \
     # .filter(ScoreSet.private.is_(False))
 
-    if owner_or_contributor is not None:
-        query = query.filter(
-            or_(
-                ScoreSet.created_by_id == owner_or_contributor.user.id,
-                ScoreSet.contributors.any(Contributor.orcid_id == owner_or_contributor.user.username),
+    if owner_or_contributor is not None and search.me is not None:
+        if search.me:
+            query = query.filter(
+                or_(
+                    ScoreSet.created_by_id == owner_or_contributor.user.id,
+                    ScoreSet.contributors.any(Contributor.orcid_id == owner_or_contributor.user.username),
+                )
+            )
+        else:
+            query = query.filter(
+                and_(
+                    ScoreSet.created_by_id != owner_or_contributor.user.id,
+                    ~ScoreSet.contributors.any(Contributor.orcid_id == owner_or_contributor.user.username),
+                )
             )
-        )
 
     if search.published is not None:
         if search.published:
@@ -256,27 +264,36 @@ def search_score_sets(db: Session, owner_or_contributor: Optional["UserData"], s
         .order_by(Experiment.title)
         .all()
     )
+    print(score_sets)
     # Remove superseded score set
     if not score_sets:
+        print("if no score set")
         final_score_sets: list[ScoreSet] = []
     else:
-        published_filter = search.published if search.published is not None else None
-        print(len(score_sets))
-        filtered_score_sets = [
-            find_superseded_score_set_tail(
-                score_set,
-                Action.READ,
-                owner_or_contributor,
-                published_filter
-            ) for score_set in score_sets
-        ]
-        print(len(filtered_score_sets))
+        if search.published:
+            filtered_score_sets_tail = [
+                find_publish_or_private_superseded_score_set_tail(
+                    score_set,
+                    Action.READ,
+                    owner_or_contributor,
+                    search.published
+                ) for score_set in score_sets
+            ]
+        else:
+            print("filtered_tail")
+            filtered_score_sets_tail = [
+                find_superseded_score_set_tail(
+                    score_set,
+                    Action.READ,
+                    owner_or_contributor
+                ) for score_set in score_sets
+            ]
+        print(len(filtered_score_sets_tail))
         # Remove None item.
-        filtered_score_sets = [score_set for score_set in filtered_score_sets if score_set is not None]
+        filtered_score_sets = [score_set for score_set in filtered_score_sets_tail if score_set is not None]
         print(len(filtered_score_sets))
         if filtered_score_sets:
-            # final_score_sets = sorted(set(filtered_score_sets), key=attrgetter("urn"))
-            final_score_sets = filtered_score_sets.sort(key=attrgetter("urn"))
+            final_score_sets = sorted(set(filtered_score_sets), key=attrgetter("urn"))
             for f in filtered_score_sets:
                 print(f.urn)
         else:
@@ -335,48 +352,54 @@ def find_meta_analyses_for_experiment_sets(db: Session, urns: list[str]) -> list
 def find_superseded_score_set_tail(
         score_set: ScoreSet,
         action: Optional["Action"] = None,
-        user_data: Optional["UserData"] = None,
-        publish: Optional[bool] = None) -> Optional[ScoreSet]:
+        user_data: Optional["UserData"] = None) -> Optional[ScoreSet]:
     from mavedb.lib.permissions import has_permission
-    if publish is not None:
-        if publish is True:
-            while score_set.superseding_score_set is not None:
-                next_score_set_in_chain = score_set.superseding_score_set
-                # Find the final published one.
-                if action is not None and has_permission(user_data, score_set, action).permitted \
-                        and next_score_set_in_chain.published_date is None:
-                    return score_set
-                score_set = next_score_set_in_chain
-        else:
-            # Unpublished score set should not be superseded.
-            # It should not have superseding score set, but possible have superseded score set.
-            if action is not None and score_set.published_date is None \
-                    and has_permission(user_data, score_set, action).permitted:
-                return score_set
+    while score_set.superseding_score_set is not None:
+        next_score_set_in_chain = score_set.superseding_score_set
+
+        # If we were given a permission to check and the next score set in the chain does not have that permission,
+        # pretend like we have reached the end of the chain. Otherwise, continue to the next score set.
+        if action is not None and not has_permission(user_data, next_score_set_in_chain, action).permitted:
+            return score_set
+
+        score_set = next_score_set_in_chain
+
+    # Handle unpublished superseding score set case.
+    # The score set has a published superseded score set but has not superseding score set.
+    if action is not None and not has_permission(user_data, score_set, action).permitted:
+        while score_set.superseded_score_set is not None:
+            next_score_set_in_chain = score_set.superseded_score_set
+            if has_permission(user_data, next_score_set_in_chain, action).permitted:
+                return next_score_set_in_chain
             else:
-                return None
-    else:
+                score_set = next_score_set_in_chain
+        return None
+
+    return score_set
+
+
+def find_publish_or_private_superseded_score_set_tail(
+        score_set: ScoreSet,
+        action: Optional["Action"] = None,
+        user_data: Optional["UserData"] = None,
+        publish: bool = True) -> Optional[ScoreSet]:
+    from mavedb.lib.permissions import has_permission
+    if publish:
         while score_set.superseding_score_set is not None:
             next_score_set_in_chain = score_set.superseding_score_set
-
-            # If we were given a permission to check and the next score set in the chain does not have that permission,
-            # pretend like we have reached the end of the chain. Otherwise, continue to the next score set.
-            if action is not None and not has_permission(user_data, next_score_set_in_chain, action).permitted:
+            # Find the final published one.
+            if action is not None and has_permission(user_data, score_set, action).permitted \
+                    and next_score_set_in_chain.published_date is None:
                 return score_set
-
             score_set = next_score_set_in_chain
-
-        # Handle unpublished superseding score set case.
-        # The score set has superseded score set but has not superseding score set.
-        if action is not None and not has_permission(user_data, score_set, action).permitted:
-            while score_set.superseded_score_set is not None:
-                next_score_set_in_chain = score_set.superseded_score_set
-                if has_permission(user_data, next_score_set_in_chain, action).permitted:
-                    return next_score_set_in_chain
-                else:
-                    score_set = next_score_set_in_chain
+    else:
+        # Unpublished score set should not be superseded.
+        # It should not have superseding score set, but possible have superseded score set.
+        if action is not None and score_set.published_date is None \
+                and has_permission(user_data, score_set, action).permitted:
+            return score_set
+        else:
             return None
-
     return score_set
 
 

src/mavedb/routers/experiments.py

@@ -171,25 +171,24 @@ def get_experiment_score_sets(
         .all()
     )
 
-    superseded_score_set_tails = [
+    filter_superseded_score_set_tails = [
         find_superseded_score_set_tail(
             score_set,
             Action.READ,
-            user_data,
-            None
+            user_data
         ) for score_set in score_set_result
     ]
-
-    if not superseded_score_set_tails:
+    filtered_score_sets = [score_set for score_set in filter_superseded_score_set_tails if score_set is not None]
+    if not filtered_score_sets:
         save_to_logging_context({"associated_resources": []})
         logger.info(msg="No score sets are associated with the requested experiment.", extra=logging_context())
 
         raise HTTPException(status_code=404, detail="no associated score sets")
     else:
-        superseded_score_set_tails.sort(key=attrgetter("urn"))
+        filtered_score_sets.sort(key=attrgetter("urn"))
         save_to_logging_context({"associated_resources": [item.urn for item in score_set_result]})
 
-    return superseded_score_set_tails
+    return filtered_score_sets
 
 
 @router.post(

src/mavedb/routers/score_sets.py

@@ -119,11 +119,15 @@ async def fetch_score_set_by_urn(
 
 
 @router.post("/score-sets/search", status_code=200, response_model=list[score_set.ShortScoreSet])
-def search_score_sets(search: ScoreSetsSearch, db: Session = Depends(deps.get_db)) -> Any:  # = Body(..., embed=True),
+def search_score_sets(
+    search: ScoreSetsSearch,
+    db: Session = Depends(deps.get_db),
+    user_data: Optional[UserData] = Depends(get_current_user),
+) -> Any:  # = Body(..., embed=True),
     """
     Search score sets.
     """
-    return _search_score_sets(db, None, search)
+    return _search_score_sets(db, user_data, search)
 
 
 @router.post(
@@ -139,6 +143,7 @@ def search_my_score_sets(
     """
     Search score sets created by the current user..
     """
+    search.me = True
     return _search_score_sets(db, user_data, search)
 
 

src/mavedb/view_models/search.py

@@ -25,6 +25,7 @@ class ScoreSetsSearch(BaseModel):
     publication_identifiers: Optional[list[str]]
     keywords: Optional[list[str]]
     text: Optional[str]
+    me: Optional[bool]
 
 
 class TextSearch(BaseModel):

tests/routers/test_experiments.py

@@ -24,6 +24,7 @@
     TEST_MEDRXIV_IDENTIFIER,
     TEST_MINIMAL_EXPERIMENT,
     TEST_MINIMAL_EXPERIMENT_RESPONSE,
+    TEST_MINIMAL_SEQ_SCORESET,
     TEST_ORCID_ID,
     TEST_PUBMED_IDENTIFIER,
     TEST_USER,
@@ -991,6 +992,112 @@ def test_search_score_sets_for_experiments(session, client, setup_router_db, dat
     assert response.json()[0]["urn"] == published_score_set["urn"]
 
 
+# Creator created a superseding score set but not published it yet.
+def test_owner_searches_score_sets_with_unpublished_superseding_score_sets_for_experiments(session, client, setup_router_db, data_files, data_provider):
+    experiment = create_experiment(client)
+    unpublished_score_set = create_seq_score_set_with_variants(
+        client, session, data_provider, experiment["urn"], data_files / "scores.csv"
+    )
+    publish_score_set_response = client.post(f"/api/v1/score-sets/{unpublished_score_set['urn']}/publish")
+    assert publish_score_set_response.status_code == 200
+    published_score_set = publish_score_set_response.json()
+    score_set_post_payload = deepcopy(TEST_MINIMAL_SEQ_SCORESET)
+    score_set_post_payload["experimentUrn"] = published_score_set["experiment"]["urn"]
+    score_set_post_payload["supersededScoreSetUrn"] = published_score_set["urn"]
+    superseding_score_set_response = client.post("/api/v1/score-sets/", json=score_set_post_payload)
+    assert superseding_score_set_response.status_code == 200
+    superseding_score_set = superseding_score_set_response.json()
+
+    # On score set publication, the experiment will get a new urn
+    experiment_urn = published_score_set["experiment"]["urn"]
+    response = client.get(f"/api/v1/experiments/{experiment_urn}/score-sets")
+    assert response.status_code == 200
+    assert len(response.json()) == 1
+    assert response.json()[0]["urn"] == superseding_score_set["urn"]
+
+
+def test_non_owner_searches_score_sets_with_unpublished_superseding_score_sets_for_experiments(session, client, setup_router_db, data_files, data_provider):
+    experiment = create_experiment(client)
+    unpublished_score_set = create_seq_score_set_with_variants(
+        client, session, data_provider, experiment["urn"], data_files / "scores.csv"
+    )
+    publish_score_set_response = client.post(f"/api/v1/score-sets/{unpublished_score_set['urn']}/publish")
+    assert publish_score_set_response.status_code == 200
+    published_score_set = publish_score_set_response.json()
+    score_set_post_payload = deepcopy(TEST_MINIMAL_SEQ_SCORESET)
+    score_set_post_payload["experimentUrn"] = published_score_set["experiment"]["urn"]
+    score_set_post_payload["supersededScoreSetUrn"] = published_score_set["urn"]
+    superseding_score_set_response = client.post("/api/v1/score-sets/", json=score_set_post_payload)
+    assert superseding_score_set_response.status_code == 200
+    superseding_score_set = superseding_score_set_response.json()
+    change_ownership(session, published_score_set["urn"], ScoreSetDbModel)
+    change_ownership(session, superseding_score_set["urn"], ScoreSetDbModel)
+    # On score set publication, the experiment will get a new urn
+    experiment_urn = published_score_set["experiment"]["urn"]
+    response = client.get(f"/api/v1/experiments/{experiment_urn}/score-sets")
+    assert response.status_code == 200
+    assert len(response.json()) == 1
+    assert response.json()[0]["urn"] == published_score_set["urn"]
+
+
+def test_owner_searches_published_superseding_score_sets_for_experiments(session, client, setup_router_db, data_files, data_provider):
+    experiment = create_experiment(client)
+    unpublished_score_set = create_seq_score_set_with_variants(
+        client, session, data_provider, experiment["urn"], data_files / "scores.csv"
+    )
+    publish_score_set_response = client.post(f"/api/v1/score-sets/{unpublished_score_set['urn']}/publish")
+    assert publish_score_set_response.status_code == 200
+    published_score_set = publish_score_set_response.json()
+
+    superseding_score_set = create_seq_score_set_with_variants(
+        client,
+        session,
+        data_provider,
+        published_score_set["experiment"]["urn"],
+        data_files / "scores.csv",
+        update={"supersededScoreSetUrn": published_score_set["urn"]},
+    )
+    published_superseding_score_set_response = client.post(f"/api/v1/score-sets/{superseding_score_set['urn']}/publish")
+    assert published_superseding_score_set_response.status_code == 200
+    published_superseding_score_set = published_superseding_score_set_response.json()
+    # On score set publication, the experiment will get a new urn
+    experiment_urn = published_score_set["experiment"]["urn"]
+    response = client.get(f"/api/v1/experiments/{experiment_urn}/score-sets")
+    assert response.status_code == 200
+    assert len(response.json()) == 1
+    assert response.json()[0]["urn"] == published_superseding_score_set["urn"]
+
+
+def test_non_owner_searches_published_superseding_score_sets_for_experiments(session, client, setup_router_db, data_files, data_provider):
+    experiment = create_experiment(client)
+    unpublished_score_set = create_seq_score_set_with_variants(
+        client, session, data_provider, experiment["urn"], data_files / "scores.csv"
+    )
+    publish_score_set_response = client.post(f"/api/v1/score-sets/{unpublished_score_set['urn']}/publish")
+    assert publish_score_set_response.status_code == 200
+    published_score_set = publish_score_set_response.json()
+
+    superseding_score_set = create_seq_score_set_with_variants(
+        client,
+        session,
+        data_provider,
+        published_score_set["experiment"]["urn"],
+        data_files / "scores.csv",
+        update={"supersededScoreSetUrn": published_score_set["urn"]},
+    )
+    published_superseding_score_set_response = client.post(f"/api/v1/score-sets/{superseding_score_set['urn']}/publish")
+    assert published_superseding_score_set_response.status_code == 200
+    published_superseding_score_set = published_superseding_score_set_response.json()
+    change_ownership(session, published_score_set["urn"], ScoreSetDbModel)
+    change_ownership(session, published_superseding_score_set["urn"], ScoreSetDbModel)
+    # On score set publication, the experiment will get a new urn
+    experiment_urn = published_score_set["experiment"]["urn"]
+    response = client.get(f"/api/v1/experiments/{experiment_urn}/score-sets")
+    assert response.status_code == 200
+    assert len(response.json()) == 1
+    assert response.json()[0]["urn"] == published_superseding_score_set["urn"]
+
+
 def test_search_score_sets_for_contributor_experiments(session, client, setup_router_db, data_files, data_provider):
     experiment = create_experiment(client)
     score_set_pub = create_seq_score_set_with_variants(