Adds various small patches to new score calibration model

- Removed the `investigator_provided` field from `ScoreCalibrationBase` and adjusted related validation logic.
- Updated `SavedScoreCalibration` to include `investigator_provided` as a boolean field.
- Modified validation error messages to include specific locations in the request body.
- Changed references from `InvestigatorProvidedScoreCalibrationCreate` to `ScoreCalibrationCreate` in `ScoreSetCreate`.
- Updated test constants and removed unnecessary test cases related to `investigator_provided`.
- Enhanced tests for creating score calibrations to ensure correct behavior for investigator-provided settings.
- Adjusted tests to reflect changes in the model structure and validation logic.

Author: bencap

src/mavedb/lib/permissions.py

@@ -28,6 +28,7 @@ class Action(Enum):
     ADD_ROLE = "add_role"
     PUBLISH = "publish"
     ADD_BADGE = "add_badge"
+    CHANGE_RANK = "change_rank"
 
 
 class PermissionResponse:
@@ -409,11 +410,12 @@ def has_permission(user_data: Optional[UserData], item: Base, action: Action) ->
             else:
                 return PermissionResponse(False, 403, f"insufficient permissions for URN '{item.urn}'")
         elif action == Action.UPDATE:
-            if user_may_edit:
-                return PermissionResponse(True)
-            # Roles which may perform this operation.
-            elif roles_permitted(active_roles, [UserRole.admin]):
+            if roles_permitted(active_roles, [UserRole.admin]):
                 return PermissionResponse(True)
+            # TODO#549: Allow editing of certain fields even if published. For now,
+            # Owner may only edit if a calibration is not published.
+            elif user_may_edit:
+                return PermissionResponse(not published, 403, f"insufficient permissions for URN '{item.urn}'")
             elif private:
                 # Do not acknowledge the existence of a private entity.
                 return PermissionResponse(False, 404, f"score calibration with URN '{item.urn}' not found")
@@ -422,12 +424,12 @@ def has_permission(user_data: Optional[UserData], item: Base, action: Action) ->
             else:
                 return PermissionResponse(False, 403, f"insufficient permissions for URN '{item.urn}'")
         elif action == Action.DELETE:
-            # Owner may only delete a calibration if it has not already been published.
-            if user_may_edit:
-                return PermissionResponse(not published, 403, f"insufficient permissions for URN '{item.urn}'")
             # Roles which may perform this operation.
-            elif roles_permitted(active_roles, [UserRole.admin]):
+            if roles_permitted(active_roles, [UserRole.admin]):
                 return PermissionResponse(True)
+            # Owner may only delete a calibration if it has not already been published.
+            elif user_may_edit:
+                return PermissionResponse(not published, 403, f"insufficient permissions for URN '{item.urn}'")
             elif private:
                 # Do not acknowledge the existence of a private entity.
                 return PermissionResponse(False, 404, f"score calibration with URN '{item.urn}' not found")
@@ -444,6 +446,13 @@ def has_permission(user_data: Optional[UserData], item: Base, action: Action) ->
                 return PermissionResponse(False, 404, f"score calibration with URN '{item.urn}' not found")
             else:
                 return PermissionResponse(False)
+        elif action == Action.CHANGE_RANK:
+            if user_may_edit:
+                return PermissionResponse(True)
+            elif roles_permitted(active_roles, [UserRole.admin]):
+                return PermissionResponse(True)
+            else:
+                return PermissionResponse(False, 403, f"insufficient permissions for URN '{item.urn}'")
 
         else:
             raise NotImplementedError(f"has_permission(User, ScoreCalibration, {action}, Role)")

src/mavedb/lib/score_calibrations.py

@@ -145,6 +145,14 @@ async def create_score_calibration_in_score_set(
     calibration = await _create_score_calibration(db, calibration_create, user)
     calibration.score_set = containing_score_set
 
+    if user.username in [contributor.orcid_id for contributor in containing_score_set.contributors] + [
+        containing_score_set.created_by.username,
+        containing_score_set.modified_by.username,
+    ]:
+        calibration.investigator_provided = True
+    else:
+        calibration.investigator_provided = False
+
     db.add(calibration)
     return calibration
 

src/mavedb/routers/score_calibrations.py

@@ -264,7 +264,7 @@ async def promote_score_calibration_to_primary_route(
         logger.debug("The requested score calibration does not exist", extra=logging_context())
         raise HTTPException(status_code=404, detail="The requested score calibration does not exist")
 
-    assert_permission(user_data, item, Action.UPDATE)
+    assert_permission(user_data, item, Action.CHANGE_RANK)
 
     if item.primary:
         logger.debug("The requested score calibration is already primary", extra=logging_context())
@@ -293,7 +293,7 @@ async def promote_score_calibration_to_primary_route(
             detail="A primary score calibration already exists for this score set. Demote it first or pass demoteExistingPrimary=True.",
         )
     elif existing_primary_calibration and demote_existing_primary:
-        assert_permission(user_data, existing_primary_calibration, Action.UPDATE)
+        assert_permission(user_data, existing_primary_calibration, Action.CHANGE_RANK)
 
     promoted_calibration = promote_score_calibration_to_primary(db, item, user_data.user, demote_existing_primary)
     db.commit()
@@ -323,7 +323,7 @@ def demote_score_calibration_from_primary_route(
         logger.debug("The requested score calibration does not exist", extra=logging_context())
         raise HTTPException(status_code=404, detail="The requested score calibration does not exist")
 
-    assert_permission(user_data, item, Action.UPDATE)
+    assert_permission(user_data, item, Action.CHANGE_RANK)
 
     if not item.primary:
         logger.debug("The requested score calibration is not primary", extra=logging_context())

src/mavedb/routers/score_sets.py

@@ -811,6 +811,7 @@ async def create_score_set(
     if item_create.score_calibrations:
         for calibration_create in item_create.score_calibrations:
             created_calibration_item = await create_score_calibration(db, calibration_create, user_data.user)
+            created_calibration_item.investigator_provided = True  # necessarily true on score set creation
             score_calibrations.append(created_calibration_item)
 
     targets: list[TargetGene] = []

src/mavedb/view_models/score_calibration.py

@@ -178,7 +178,6 @@ class ScoreCalibrationBase(BaseModel):
     """
 
     title: str
-    investigator_provided: bool
     research_use_only: bool = False
 
     baseline_score: Optional[float] = None
@@ -225,7 +224,7 @@ def test_overlap(range_test: FunctionalRangeBase, range_check: FunctionalRangeBa
                 if test_overlap(a, b):
                     raise ValidationError(
                         f"Score ranges may not overlap; `{a.label}` ({a.range}) overlaps with `{b.label}` ({b.range}).",
-                        custom_loc=["body", "scoreCalibration", "functionalRanges", i, "range"],
+                        custom_loc=["body", i, "range"],
                     )
         return field_value
 
@@ -236,16 +235,17 @@ def functional_range_labels_must_be_unique(self: "ScoreCalibrationBase") -> "Sco
             return self
 
         seen, dupes = set(), set()
-        for fr in self.functional_ranges:
+        for i, fr in enumerate(self.functional_ranges):
             fr.label = fr.label.strip()
             if fr.label in seen:
-                dupes.add(fr.label)
+                dupes.add((fr.label, i))
             else:
                 seen.add(fr.label)
 
         if dupes:
             raise ValidationError(
-                f"Detected repeated label(s): {', '.join(dupes)}. Functional range labels must be unique."
+                f"Detected repeated label(s): {', '.join(label for label, _ in dupes)}. Functional range labels must be unique.",
+                custom_loc=["body", "functionalRanges", dupes.pop()[1], "label"],
             )
 
         return self
@@ -261,8 +261,9 @@ def validate_baseline_score(self: "ScoreCalibrationBase") -> "ScoreCalibrationBa
 
         for fr in self.functional_ranges:
             if fr.is_contained_by_range(self.baseline_score) and fr.classification != "normal":
-                raise ValueError(
-                    f"The provided baseline score of {self.baseline_score} falls within a non-normal range ({fr.label}). Baseline scores may not fall within non-normal ranges."
+                raise ValidationError(
+                    f"The provided baseline score of {self.baseline_score} falls within a non-normal range ({fr.label}). Baseline scores may not fall within non-normal ranges.",
+                    custom_loc=["body", "baselineScore"],
                 )
 
         return self
@@ -288,12 +289,6 @@ class ScoreCalibrationCreate(ScoreCalibrationModify):
     method_sources: Optional[Sequence[PublicationIdentifierCreate]] = None
 
 
-class InvestigatorProvidedScoreCalibrationCreate(ScoreCalibrationCreate):
-    """Model used to create a new investigator-provided score calibration. Enforces fixed field values for certain properties."""
-
-    investigator_provided: Literal[True] = True
-
-
 class SavedScoreCalibration(ScoreCalibrationBase):
     """Persisted score calibration model (includes identifiers and source lists)."""
 
@@ -304,6 +299,7 @@ class SavedScoreCalibration(ScoreCalibrationBase):
 
     score_set_id: int
 
+    investigator_provided: bool
     primary: bool = False
     private: bool = True
 
@@ -338,15 +334,20 @@ def publication_identifiers_validator(cls, value: Any) -> Optional[list[Publicat
     def primary_calibrations_may_not_be_research_use_only(self: "SavedScoreCalibration") -> "SavedScoreCalibration":
         """Primary calibrations may not be marked as research use only."""
         if self.primary and self.research_use_only:
-            raise ValidationError("Primary score calibrations may not be marked as research use only.")
+            raise ValidationError(
+                "Primary score calibrations may not be marked as research use only.",
+                custom_loc=["body", "researchUseOnly"],
+            )
 
         return self
 
     @model_validator(mode="after")
     def primary_calibrations_may_not_be_private(self: "SavedScoreCalibration") -> "SavedScoreCalibration":
         """Primary calibrations may not be marked as private."""
         if self.primary and self.private:
-            raise ValidationError("Primary score calibrations may not be marked as private.")
+            raise ValidationError(
+                "Primary score calibrations may not be marked as private.", custom_loc=["body", "private"]
+            )
 
         return self
 

src/mavedb/view_models/score_set.py

@@ -33,7 +33,7 @@
 )
 from mavedb.view_models.score_calibration import (
     SavedScoreCalibration,
-    InvestigatorProvidedScoreCalibrationCreate,
+    ScoreCalibrationCreate,
     ScoreCalibration,
 )
 from mavedb.view_models.target_gene import (
@@ -178,7 +178,7 @@ class ScoreSetCreate(ScoreSetModify):
     #       If this propertie ever became available during calibration creation,
     #       validation criteria which enforces constraints on there being a single primary
     #       calibration per score set would need to be added at this model level.
-    score_calibrations: Optional[Sequence[InvestigatorProvidedScoreCalibrationCreate]] = None
+    score_calibrations: Optional[Sequence[ScoreCalibrationCreate]] = None
 
     @field_validator("superseded_score_set_urn")
     def validate_superseded_score_set_urn(cls, v: Optional[str]) -> Optional[str]:

tests/helpers/constants.py

@@ -1502,7 +1502,6 @@
 TEST_BRNICH_SCORE_CALIBRATION = {
     "title": "Test BRNICH Score Calibration",
     "research_use_only": False,
-    "investigator_provided": False,
     "baseline_score": TEST_BASELINE_SCORE,
     "baseline_score_description": "Test baseline score description",
     "functional_ranges": [
@@ -1536,6 +1535,7 @@
     "methodSources": [SAVED_PUBMED_PUBLICATION],
     "id": 1,
     "urn": VALID_CALIBRATION_URN,
+    "investigatorProvided": True,
     "primary": True,
     "private": False,
     "scoreSetId": 1,
@@ -1555,20 +1555,9 @@
     "modificationDate": date.today().isoformat(),
 }
 
-TEST_INVESTIGATOR_PROVIDED_SCORE_CALIBRATION = {
-    **TEST_BRNICH_SCORE_CALIBRATION,
-    "investigator_provided": True,
-}
-
-TEST_SAVED_INVESTIGATOR_PROVIDED_SCORE_CALIBRATION = {
-    **TEST_SAVED_BRNICH_SCORE_CALIBRATION,
-    "investigatorProvided": True,
-}
-
 TEST_PATHOGENICITY_SCORE_CALIBRATION = {
     "title": "Test Pathogenicity Score Calibration",
     "research_use_only": False,
-    "investigator_provided": False,
     "baseline_score": TEST_BASELINE_SCORE,
     "baseline_score_description": "Test baseline score description",
     "functional_ranges": [
@@ -1596,6 +1585,7 @@
     "classificationSources": None,
     "methodSources": None,
     "id": 2,
+    "investigatorProvided": True,
     "primary": False,
     "private": False,
     "urn": VALID_CALIBRATION_URN,

tests/lib/conftest.py

@@ -64,13 +64,18 @@ def setup_lib_db_with_score_set(session, setup_lib_db):
     """
     Sets up the lib test db with a user, reference, license, and a score set.
     """
+    user = session.query(User).filter(User.username == TEST_USER["username"]).first()
 
     experiment_set = ExperimentSet(**TEST_EXPERIMENT_SET, urn=VALID_EXPERIMENT_SET_URN)
+    experiment_set.created_by = user
+    experiment_set.modified_by = user
     session.add(experiment_set)
     session.commit()
     session.refresh(experiment_set)
 
     experiment = Experiment(**TEST_EXPERIMENT, urn=VALID_EXPERIMENT_URN, experiment_set_id=experiment_set.id)
+    experiment.created_by = user
+    experiment.modified_by = user
     session.add(experiment)
     session.commit()
     session.refresh(experiment)
@@ -80,7 +85,8 @@ def setup_lib_db_with_score_set(session, setup_lib_db):
     score_set = ScoreSet(
         **score_set_scaffold, urn=VALID_SCORE_SET_URN, experiment_id=experiment.id, licence_id=TEST_LICENSE["id"]
     )
-
+    score_set.created_by = user
+    score_set.modified_by = user
     session.add(score_set)
     session.commit()
     session.refresh(score_set)

tests/lib/test_score_calibrations.py

@@ -34,7 +34,9 @@
     TEST_PUBMED_IDENTIFIER,
     TEST_SEQ_SCORESET,
     VALID_SCORE_SET_URN,
+    EXTRA_USER,
 )
+from tests.helpers.util.contributor import add_contributor
 from tests.helpers.util.score_calibration import create_test_score_calibration_in_score_set
 
 ################################################################################
@@ -88,6 +90,82 @@ async def test_create_score_calibration_in_score_set_creates_score_calibration_w
     assert calibration.score_set == setup_lib_db_with_score_set
 
 
+@pytest.mark.asyncio
+async def test_create_score_calibration_in_score_set_investigator_provided_set_when_creator_is_owner(
+    setup_lib_db_with_score_set, session, mock_user
+):
+    test_user = session.execute(select(User)).scalars().first()
+
+    MockCalibrationCreate = create_model(
+        "MockCalibrationCreate",
+        score_set_urn=(str | None, setup_lib_db_with_score_set.urn),
+        threshold_sources=(list, []),
+        classification_sources=(list, []),
+        method_sources=(list, []),
+    )
+
+    calibration = await create_score_calibration_in_score_set(session, MockCalibrationCreate(), test_user)
+    assert calibration is not None
+    assert calibration.score_set == setup_lib_db_with_score_set
+    assert calibration.created_by == test_user
+    assert calibration.modified_by == test_user
+    assert calibration.investigator_provided is True
+
+
+@pytest.mark.asyncio
+async def test_create_score_calibration_in_score_set_investigator_provided_set_when_creator_is_contributor(
+    setup_lib_db_with_score_set, session
+):
+    extra_user = session.execute(select(User).where(User.username == EXTRA_USER["username"])).scalars().first()
+
+    add_contributor(
+        session,
+        setup_lib_db_with_score_set.urn,
+        ScoreSet,
+        EXTRA_USER["username"],
+        EXTRA_USER["first_name"],
+        EXTRA_USER["last_name"],
+    )
+
+    MockCalibrationCreate = create_model(
+        "MockCalibrationCreate",
+        score_set_urn=(str | None, setup_lib_db_with_score_set.urn),
+        threshold_sources=(list, []),
+        classification_sources=(list, []),
+        method_sources=(list, []),
+    )
+
+    calibration = await create_score_calibration_in_score_set(session, MockCalibrationCreate(), extra_user)
+    assert calibration is not None
+    assert calibration.score_set == setup_lib_db_with_score_set
+    assert calibration.created_by == extra_user
+    assert calibration.modified_by == extra_user
+    assert calibration.investigator_provided is True
+
+
+@pytest.mark.asyncio
+async def test_create_score_calibration_in_score_set_investigator_provided_not_set_when_creator_not_owner(
+    setup_lib_db_with_score_set, session
+):
+    MockCalibrationCreate = create_model(
+        "MockCalibrationCreate",
+        score_set_urn=(str | None, setup_lib_db_with_score_set.urn),
+        threshold_sources=(list, []),
+        classification_sources=(list, []),
+        method_sources=(list, []),
+    )
+
+    # invoke from a different user context
+    extra_user = session.execute(select(User).where(User.username == EXTRA_USER["username"])).scalars().first()
+
+    calibration = await create_score_calibration_in_score_set(session, MockCalibrationCreate(), extra_user)
+    assert calibration is not None
+    assert calibration.score_set == setup_lib_db_with_score_set
+    assert calibration.created_by == extra_user
+    assert calibration.modified_by == extra_user
+    assert calibration.investigator_provided is False
+
+
 ### create_score_calibration
 
 
@@ -605,7 +683,8 @@ async def test_modify_score_calibration_new_score_set(setup_lib_db_with_score_se
         experiment_id=existing_experiment.id,
         licence_id=TEST_LICENSE["id"],
     )
-
+    new_containing_score_set.created_by = setup_lib_db_with_score_set.created_by
+    new_containing_score_set.modified_by = setup_lib_db_with_score_set.modified_by
     session.add(new_containing_score_set)
     session.commit()
     session.refresh(new_containing_score_set)