From 2d9c43fe84c03a7adcf26ba0ea8640c99516d8df Mon Sep 17 00:00:00 2001 From: Estelle Da Date: Mon, 15 Jun 2026 16:37:51 +1000 Subject: [PATCH 1/4] Add superseded_calibration in Calibration. Modified related functions and tests. --- ...7c60b_add_calibration_superseded_column.py | 32 +++++++++++++ src/mavedb/lib/score_calibrations.py | 47 +++++++++++++++---- src/mavedb/lib/validation/urn_re.py | 8 ++++ src/mavedb/models/score_calibration.py | 13 ++++- src/mavedb/routers/score_calibrations.py | 3 +- src/mavedb/routers/score_sets.py | 2 +- src/mavedb/scripts/load_calibration_csv.py | 4 +- .../scripts/load_excalibr_calibrations.py | 4 +- src/mavedb/view_models/score_calibration.py | 16 +++++++ tests/routers/test_score_calibrations.py | 23 ++++++--- 10 files changed, 133 insertions(+), 19 deletions(-) create mode 100644 alembic/versions/adb481b7c60b_add_calibration_superseded_column.py diff --git a/alembic/versions/adb481b7c60b_add_calibration_superseded_column.py b/alembic/versions/adb481b7c60b_add_calibration_superseded_column.py new file mode 100644 index 000000000..d6a064ad2 --- /dev/null +++ b/alembic/versions/adb481b7c60b_add_calibration_superseded_column.py @@ -0,0 +1,32 @@ +"""add_calibration_superseded_column + +Revision ID: adb481b7c60b +Revises: 398067c53257 +Create Date: 2026-06-01 16:45:35.507837 + +""" +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + +# revision identifiers, used by Alembic. +revision = 'adb481b7c60b' +down_revision = '398067c53257' +branch_labels = None +depends_on = None + + +def upgrade(): + # ### commands auto generated by Alembic - please adjust! ### + op.add_column('score_calibrations', sa.Column('replaces_id', sa.Integer(), nullable=True)) + op.create_index(op.f('ix_score_calibrations_replaces_id'), 'score_calibrations', ['replaces_id'], unique=False) + op.create_foreign_key(None, 'score_calibrations', 'score_calibrations', ['replaces_id'], ['id']) + # ### end Alembic commands ### + + +def downgrade(): + # ### commands auto generated by Alembic - please adjust! ### + op.drop_constraint(None, 'score_calibrations', type_='foreignkey') + op.drop_index(op.f('ix_score_calibrations_replaces_id'), table_name='score_calibrations') + op.drop_column('score_calibrations', 'replaces_id') + # ### end Alembic commands ### diff --git a/src/mavedb/lib/score_calibrations.py b/src/mavedb/lib/score_calibrations.py index 11e1b2e88..a4615e4a6 100644 --- a/src/mavedb/lib/score_calibrations.py +++ b/src/mavedb/lib/score_calibrations.py @@ -4,11 +4,14 @@ from typing import Optional, Union import pandas as pd +from fastapi import Depends from sqlalchemy import Float, and_, select from sqlalchemy.orm import Session from mavedb.lib.acmg import find_or_create_acmg_classification from mavedb.lib.identifiers import find_or_create_publication_identifier +from mavedb.lib.permissions import Action, has_permission +from mavedb.lib.types.authentication import UserData from mavedb.lib.types.score_calibrations import ClassificationDict from mavedb.lib.validation.constants.general import ( calibration_class_column_name, @@ -25,6 +28,7 @@ from mavedb.models.user import User from mavedb.models.variant import Variant from mavedb.view_models import score_calibration +from mavedb.view_models.score_calibration import ScoreCalibrationCreate def create_functional_classification( @@ -97,7 +101,7 @@ def create_functional_classification( async def _create_score_calibration( db: Session, calibration_create: score_calibration.ScoreCalibrationCreate, - user: User, + user_data: UserData, variant_classes: Optional[ClassificationDict] = None, containing_score_set: Optional[ScoreSet] = None, ) -> ScoreCalibration: @@ -176,6 +180,8 @@ async def _create_score_calibration( db.add(pub) db.flush() + superseded_calibration = validate_superseded_score_calibration(db, calibration_create, user_data) + calibration = ScoreCalibration( **calibration_create.model_dump( by_alias=False, @@ -185,12 +191,14 @@ async def _create_score_calibration( "evidence_sources", "method_sources", "score_set_urn", + "superseded_calibration_urn" }, ), publication_identifier_associations=calibration_pub_assocs, functional_classifications=[], - created_by=user, - modified_by=user, + superseded_calibration=superseded_calibration, + created_by=user_data.user, + modified_by=user_data.user, ) # type: ignore[call-arg] if containing_score_set: @@ -210,7 +218,7 @@ async def _create_score_calibration( async def create_score_calibration_in_score_set( db: Session, calibration_create: score_calibration.ScoreCalibrationCreate, - user: User, + user_data: UserData, variant_classes: Optional[ClassificationDict] = None, ) -> ScoreCalibration: """ @@ -251,9 +259,9 @@ async def create_score_calibration_in_score_set( raise ValueError("score_set_urn must be provided to create a score calibration within a score set.") containing_score_set = db.query(ScoreSet).where(ScoreSet.urn == calibration_create.score_set_urn).one() - calibration = await _create_score_calibration(db, calibration_create, user, variant_classes, containing_score_set) + calibration = await _create_score_calibration(db, calibration_create, user_data, variant_classes, containing_score_set) - if user.username in [contributor.orcid_id for contributor in containing_score_set.contributors] + [ + if user_data.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, ]: @@ -268,7 +276,7 @@ async def create_score_calibration_in_score_set( async def create_score_calibration( db: Session, calibration_create: score_calibration.ScoreCalibrationCreate, - user: User, + user_data: UserData, variant_classes: Optional[ClassificationDict] = None, ) -> ScoreCalibration: """ @@ -321,7 +329,7 @@ async def create_score_calibration( raise ValueError("score_set_urn must not be provided to create a score calibration outside a score set.") created_calibration = await _create_score_calibration( - db, calibration_create, user, variant_classes, containing_score_set=None + db, calibration_create, user_data, variant_classes, containing_score_set=None ) db.add(created_calibration) @@ -652,6 +660,29 @@ def delete_score_calibration(db: Session, calibration: ScoreCalibration) -> None return None +def validate_superseded_score_calibration( + db: Session, + calibration_create: ScoreCalibrationCreate, + user_data: Optional[UserData], +) -> Optional[ScoreCalibration]: + if not calibration_create.superseded_calibration_urn: + return None + + superseded_calibration = ( + db.query(ScoreCalibration) + .where(ScoreCalibration.urn == calibration_create.superseded_calibration_urn) + .one_or_none() + ) + + if superseded_calibration is None: + raise ValueError("Superseded calibration does not exist.") + + if not has_permission(user_data, superseded_calibration, Action.READ).permitted: + raise ValueError("No access right to supersede this calibration.") + + return superseded_calibration + + def variants_for_functional_classification( db: Session, functional_classification: ScoreCalibrationFunctionalClassification, diff --git a/src/mavedb/lib/validation/urn_re.py b/src/mavedb/lib/validation/urn_re.py index 82feb19a2..595c07e3d 100644 --- a/src/mavedb/lib/validation/urn_re.py +++ b/src/mavedb/lib/validation/urn_re.py @@ -32,6 +32,14 @@ MAVEDB_COLLECTION_URN_PATTERN = r"urn:mavedb:collection-[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}" MAVEDB_COLLECTION_URN_RE = re.compile(MAVEDB_COLLECTION_URN_PATTERN) +# Temp calibration URN +MAVEDB_TMP_CALIBRATION_URN_PATTERN = r"tmp:mavedb.calibration-[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}" +MAVEDB_TMP_CALIBRATION_URN_RE = re.compile(MAVEDB_TMP_CALIBRATION_URN_PATTERN) + +# Calibration URN +MAVEDB_CALIBRATION_URN_PATTERN = r"urn:mavedb:calibration-[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}" +MAVEDB_CALIBRATION_URN_RE = re.compile(MAVEDB_CALIBRATION_URN_PATTERN) + # Any URN MAVEDB_ANY_URN_PATTERN = "|".join( [ diff --git a/src/mavedb/models/score_calibration.py b/src/mavedb/models/score_calibration.py index 38ce1f286..f4c283010 100644 --- a/src/mavedb/models/score_calibration.py +++ b/src/mavedb/models/score_calibration.py @@ -3,7 +3,7 @@ from __future__ import annotations from datetime import date -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Optional from sqlalchemy import Boolean, Column, Date, Float, ForeignKey, Integer, String from sqlalchemy.dialects.postgresql import JSONB @@ -60,6 +60,17 @@ class ScoreCalibration(Base): calibration_metadata = Column(JSONB(none_as_null=True), nullable=True) + superseded_calibration_id = Column("replaces_id", Integer, ForeignKey("score_calibrations.id"), index=True, nullable=True) + superseded_calibration: Mapped[Optional["ScoreCalibration"]] = relationship( + "ScoreCalibration", + uselist=False, + foreign_keys="ScoreCalibration.superseded_calibration_id", + remote_side=[id], + ) + superseding_calibration: Mapped[Optional["ScoreCalibration"]] = relationship( + "ScoreCalibration", uselist=False, back_populates="superseded_calibration" + ) + created_by_id = Column(Integer, ForeignKey("users.id"), index=True, nullable=False) created_by: Mapped["User"] = relationship("User", foreign_keys="ScoreCalibration.created_by_id") modified_by_id = Column(Integer, ForeignKey("users.id"), index=True, nullable=False) diff --git a/src/mavedb/routers/score_calibrations.py b/src/mavedb/routers/score_calibrations.py index e8b31478c..eff22df22 100644 --- a/src/mavedb/routers/score_calibrations.py +++ b/src/mavedb/routers/score_calibrations.py @@ -132,6 +132,7 @@ async def get_score_calibrations_for_score_set( calibrations = ( db.query(ScoreCalibration) .filter(ScoreCalibration.score_set_id == score_set.id) + .filter(~ScoreCalibration.superseding_calibration.has(ScoreCalibration.private == False)) .options(selectinload(ScoreCalibration.score_set).selectinload(ScoreSet.contributors)) .all() ) @@ -339,7 +340,7 @@ async def create_score_calibration_route( ) created_calibration = await create_score_calibration_in_score_set( - db, calibration, user_data.user, variant_classes if classes_file else None + db, calibration, user_data, variant_classes if classes_file else None ) db.commit() diff --git a/src/mavedb/routers/score_sets.py b/src/mavedb/routers/score_sets.py index 74f2a5c0b..349080cac 100644 --- a/src/mavedb/routers/score_sets.py +++ b/src/mavedb/routers/score_sets.py @@ -1742,7 +1742,7 @@ async def create_score_set( ) created_calibration_item = await create_score_calibration( - db, calibration_create, user_data.user, variant_classes=None + db, calibration_create, user_data, variant_classes=None ) created_calibration_item.investigator_provided = True # necessarily true on score set creation score_calibrations.append(created_calibration_item) diff --git a/src/mavedb/scripts/load_calibration_csv.py b/src/mavedb/scripts/load_calibration_csv.py index 066509e39..2a3c7b482 100644 --- a/src/mavedb/scripts/load_calibration_csv.py +++ b/src/mavedb/scripts/load_calibration_csv.py @@ -100,6 +100,7 @@ from mavedb.lib.acmg import ACMGCriterion, StrengthOfEvidenceProvided from mavedb.lib.oddspaths import oddspaths_evidence_strength_equivalent from mavedb.lib.score_calibrations import create_score_calibration_in_score_set +from mavedb.lib.types.authentication import UserData from mavedb.models import score_calibration from mavedb.models.enums.functional_classification import FunctionalClassification as FunctionalClassifcationOptions from mavedb.models.score_set import ScoreSet @@ -414,8 +415,9 @@ def main(db: Session, csv_path: str, delimiter: str, overwrite: bool, purge_publ system_user = db.query(User).filter(User.id == 1).one() calibration_user = score_set.created_by if calibration_is_investigator_provided else system_user + calibration_user_data = UserData(calibration_user, calibration_user.roles) new_calibration_object = asyncio.run( - create_score_calibration_in_score_set(db, created_score_calibration, calibration_user) + create_score_calibration_in_score_set(db, created_score_calibration, calibration_user_data) ) new_calibration_object.primary = primary new_calibration_object.private = False diff --git a/src/mavedb/scripts/load_excalibr_calibrations.py b/src/mavedb/scripts/load_excalibr_calibrations.py index 72f7ed2cd..208754806 100644 --- a/src/mavedb/scripts/load_excalibr_calibrations.py +++ b/src/mavedb/scripts/load_excalibr_calibrations.py @@ -49,6 +49,7 @@ from sqlalchemy.orm import Session from mavedb.lib.score_calibrations import create_score_calibration_in_score_set +from mavedb.lib.types.authentication import UserData from mavedb.models.enums.functional_classification import FunctionalClassification as FunctionalClassificationOptions from mavedb.models.score_calibration import ScoreCalibration from mavedb.models.score_set import ScoreSet @@ -238,8 +239,9 @@ def main(db: Session, csv_path: str, dataset_map: str, overwrite: bool, remove: method_sources=[EXCALIBR_CALIBRATION_CITATION], ) + system_user_data = UserData(system_user, system_user.roles) new_calibration_object = asyncio.run( - create_score_calibration_in_score_set(db, score_calibration_create, system_user) + create_score_calibration_in_score_set(db, score_calibration_create, system_user_data) ) new_calibration_object.primary = False new_calibration_object.private = False diff --git a/src/mavedb/view_models/score_calibration.py b/src/mavedb/view_models/score_calibration.py index 857d4e49f..06e12885b 100644 --- a/src/mavedb/view_models/score_calibration.py +++ b/src/mavedb/view_models/score_calibration.py @@ -10,6 +10,7 @@ from pydantic import Field, field_validator, model_validator from mavedb.lib.oddspaths import oddspaths_evidence_strength_equivalent +from mavedb.lib.validation import urn_re from mavedb.lib.validation.exceptions import ValidationError from mavedb.lib.validation.transform import ( transform_score_calibration_publication_identifiers, @@ -477,11 +478,24 @@ def functional_classifications_require_publication_sources( class ScoreCalibrationCreate(ScoreCalibrationModify): """Model used to create a new score calibration.""" + superseded_calibration_urn: Optional[str] = None functional_classifications: Optional[Sequence[FunctionalClassificationCreate]] = None threshold_sources: Sequence[PublicationIdentifierCreate] evidence_sources: Sequence[PublicationIdentifierCreate] method_sources: Sequence[PublicationIdentifierCreate] + @field_validator("superseded_calibration_urn") + def validate_superseded_calibration_urn(cls, v: Optional[str]) -> Optional[str]: + if not v: + return None + + if urn_re.MAVEDB_CALIBRATION_URN_RE.fullmatch(v) is None: + if urn_re.MAVEDB_TMP_CALIBRATION_URN_RE.fullmatch(v) is None: + raise ValueError(f"'{v}' is not a valid calibration URN") + else: + raise ValueError("cannot supersede a private calibration - please edit it instead") + + return v class SavedScoreCalibration(ScoreCalibrationBase): """Persisted score calibration model (includes identifiers and source lists).""" @@ -501,6 +515,8 @@ class SavedScoreCalibration(ScoreCalibrationBase): threshold_sources: Sequence[SavedPublicationIdentifier] evidence_sources: Sequence[SavedPublicationIdentifier] method_sources: Sequence[SavedPublicationIdentifier] + superseded_calibration: Optional[ScoreCalibrationBase] = None + superseding_calibration: Optional[ScoreCalibrationBase] = None created_by: Optional[SavedUser] = None modified_by: Optional[SavedUser] = None diff --git a/tests/routers/test_score_calibrations.py b/tests/routers/test_score_calibrations.py index fe1aeba7c..9f9be8391 100644 --- a/tests/routers/test_score_calibrations.py +++ b/tests/routers/test_score_calibrations.py @@ -963,8 +963,15 @@ def test_creating_user_can_get_score_calibrations_for_score_set_when_public( assert response.status_code == 200 calibrations_response = response.json() assert len(calibrations_response) == 2 - assert calibrations_response[0]["urn"] == calibration["urn"] - assert calibrations_response[0]["private"] is False + # The order is different from before. + created_calibration = next( + c for c in calibrations_response if c["urn"] == calibration["urn"] + ) + assert created_calibration["private"] is False + other_calibration = next( + c for c in calibrations_response if c["urn"] != calibration["urn"] + ) + assert other_calibration["private"] is True @pytest.mark.parametrize( @@ -1013,8 +1020,10 @@ def test_contributing_user_can_get_score_calibrations_for_score_set_when_public( assert response.status_code == 200 calibrations_response = response.json() assert len(calibrations_response) == 2 - assert calibrations_response[0]["urn"] == calibration["urn"] - assert calibrations_response[0]["private"] is False + created_calibration = next( + c for c in calibrations_response if c["urn"] == calibration["urn"] + ) + assert created_calibration["private"] is False @pytest.mark.parametrize( @@ -1054,8 +1063,10 @@ def test_admin_user_can_get_score_calibrations_for_score_set_when_public( assert response.status_code == 200 calibrations_response = response.json() assert len(calibrations_response) == 2 - assert calibrations_response[0]["urn"] == calibration["urn"] - assert calibrations_response[0]["private"] is False + created_calibration = next( + c for c in calibrations_response if c["urn"] == calibration["urn"] + ) + assert created_calibration["private"] is False ########################################################### From 8dfd0a2501d88a66f23a6f354d029960d2748247 Mon Sep 17 00:00:00 2001 From: Estelle Da Date: Mon, 15 Jun 2026 16:50:41 +1000 Subject: [PATCH 2/4] Remove unnecessary import and modify query code. --- .../versions/adb481b7c60b_add_calibration_superseded_column.py | 1 - src/mavedb/lib/score_calibrations.py | 1 - src/mavedb/routers/score_calibrations.py | 2 +- 3 files changed, 1 insertion(+), 3 deletions(-) diff --git a/alembic/versions/adb481b7c60b_add_calibration_superseded_column.py b/alembic/versions/adb481b7c60b_add_calibration_superseded_column.py index d6a064ad2..ce88b632d 100644 --- a/alembic/versions/adb481b7c60b_add_calibration_superseded_column.py +++ b/alembic/versions/adb481b7c60b_add_calibration_superseded_column.py @@ -7,7 +7,6 @@ """ from alembic import op import sqlalchemy as sa -from sqlalchemy.dialects import postgresql # revision identifiers, used by Alembic. revision = 'adb481b7c60b' diff --git a/src/mavedb/lib/score_calibrations.py b/src/mavedb/lib/score_calibrations.py index a4615e4a6..392175f37 100644 --- a/src/mavedb/lib/score_calibrations.py +++ b/src/mavedb/lib/score_calibrations.py @@ -4,7 +4,6 @@ from typing import Optional, Union import pandas as pd -from fastapi import Depends from sqlalchemy import Float, and_, select from sqlalchemy.orm import Session diff --git a/src/mavedb/routers/score_calibrations.py b/src/mavedb/routers/score_calibrations.py index eff22df22..f55bb8871 100644 --- a/src/mavedb/routers/score_calibrations.py +++ b/src/mavedb/routers/score_calibrations.py @@ -132,7 +132,7 @@ async def get_score_calibrations_for_score_set( calibrations = ( db.query(ScoreCalibration) .filter(ScoreCalibration.score_set_id == score_set.id) - .filter(~ScoreCalibration.superseding_calibration.has(ScoreCalibration.private == False)) + .filter(~ScoreCalibration.superseding_calibration.has(ScoreCalibration.private.is_(False))) .options(selectinload(ScoreCalibration.score_set).selectinload(ScoreSet.contributors)) .all() ) From 63bc5b865ab896cdf463beb759d6b217719d22ea Mon Sep 17 00:00:00 2001 From: Estelle Da Date: Tue, 16 Jun 2026 16:35:30 +1000 Subject: [PATCH 3/4] Modified related tests. --- tests/helpers/util/score_calibration.py | 6 +- tests/lib/test_score_calibrations.py | 135 +++++++++++++++++------- 2 files changed, 97 insertions(+), 44 deletions(-) diff --git a/tests/helpers/util/score_calibration.py b/tests/helpers/util/score_calibration.py index a535096c2..900e3822d 100644 --- a/tests/helpers/util/score_calibration.py +++ b/tests/helpers/util/score_calibration.py @@ -3,8 +3,8 @@ import jsonschema from mavedb.lib.score_calibrations import create_score_calibration_in_score_set +from mavedb.lib.types.authentication import UserData from mavedb.models.score_calibration import ScoreCalibration -from mavedb.models.user import User from mavedb.view_models.score_calibration import ScoreCalibrationCreate, ScoreCalibrationWithScoreSetUrn from tests.helpers.constants import TEST_BRNICH_SCORE_CALIBRATION_RANGE_BASED @@ -14,12 +14,12 @@ async def create_test_range_based_score_calibration_in_score_set( - db: "Session", score_set_urn: str, user: User + db: "Session", score_set_urn: str, user_data: UserData ) -> ScoreCalibration: calibration_create = ScoreCalibrationCreate( **TEST_BRNICH_SCORE_CALIBRATION_RANGE_BASED, score_set_urn=score_set_urn ) - created_score_calibration = await create_score_calibration_in_score_set(db, calibration_create, user) + created_score_calibration = await create_score_calibration_in_score_set(db, calibration_create, user_data) assert created_score_calibration is not None db.commit() diff --git a/tests/lib/test_score_calibrations.py b/tests/lib/test_score_calibrations.py index 9a5ba43c5..43f0b367e 100644 --- a/tests/lib/test_score_calibrations.py +++ b/tests/lib/test_score_calibrations.py @@ -23,6 +23,7 @@ variant_classification_df_to_dict, variants_for_functional_classification, ) +from mavedb.lib.types.authentication import UserData from mavedb.lib.validation.constants.general import ( calibration_class_column_name, calibration_variant_column_name, @@ -341,6 +342,7 @@ async def test_create_score_calibration_in_score_set_creates_score_calibration_w setup_lib_db_with_score_set, session ): test_user = session.execute(select(User)).scalars().first() + test_user_data = UserData(user=test_user, active_roles=[]) MockCalibrationCreate = create_model( "MockCalibrationCreate", @@ -349,9 +351,10 @@ async def test_create_score_calibration_in_score_set_creates_score_calibration_w evidence_sources=(list, []), method_sources=(list, []), functional_classifications=(list, []), + superseded_calibration_urn=(str | None, None), ) - calibration = await create_score_calibration_in_score_set(session, MockCalibrationCreate(), test_user) + calibration = await create_score_calibration_in_score_set(session, MockCalibrationCreate(), test_user_data) assert calibration is not None assert calibration.score_set == setup_lib_db_with_score_set @@ -361,6 +364,7 @@ async def test_create_score_calibration_in_score_set_investigator_provided_set_w setup_lib_db_with_score_set, session, mock_user ): test_user = session.execute(select(User)).scalars().first() + test_user_data = UserData(user=test_user, active_roles=[]) MockCalibrationCreate = create_model( "MockCalibrationCreate", @@ -369,9 +373,10 @@ async def test_create_score_calibration_in_score_set_investigator_provided_set_w evidence_sources=(list, []), method_sources=(list, []), functional_classifications=(list, []), + superseded_calibration_urn=(str | None, None), ) - calibration = await create_score_calibration_in_score_set(session, MockCalibrationCreate(), test_user) + calibration = await create_score_calibration_in_score_set(session, MockCalibrationCreate(), test_user_data) assert calibration is not None assert calibration.score_set == setup_lib_db_with_score_set assert calibration.created_by == test_user @@ -384,6 +389,7 @@ async def test_create_score_calibration_in_score_set_investigator_provided_set_w setup_lib_db_with_score_set, session ): extra_user = session.execute(select(User).where(User.username == EXTRA_USER["username"])).scalars().first() + extra_user_data = UserData(user=extra_user, active_roles=[]) add_contributor( session, @@ -401,9 +407,10 @@ async def test_create_score_calibration_in_score_set_investigator_provided_set_w evidence_sources=(list, []), method_sources=(list, []), functional_classifications=(list, []), + superseded_calibration_urn=(str | None, None), ) - calibration = await create_score_calibration_in_score_set(session, MockCalibrationCreate(), extra_user) + calibration = await create_score_calibration_in_score_set(session, MockCalibrationCreate(), extra_user_data) assert calibration is not None assert calibration.score_set == setup_lib_db_with_score_set assert calibration.created_by == extra_user @@ -422,12 +429,14 @@ async def test_create_score_calibration_in_score_set_investigator_provided_not_s evidence_sources=(list, []), method_sources=(list, []), functional_classifications=(list, []), + superseded_calibration_urn=(str | None, None), ) # invoke from a different user context extra_user = session.execute(select(User).where(User.username == EXTRA_USER["username"])).scalars().first() + extra_user_data = UserData(user=extra_user, active_roles=[]) - calibration = await create_score_calibration_in_score_set(session, MockCalibrationCreate(), extra_user) + calibration = await create_score_calibration_in_score_set(session, MockCalibrationCreate(), extra_user_data) assert calibration is not None assert calibration.score_set == setup_lib_db_with_score_set assert calibration.created_by == extra_user @@ -453,6 +462,7 @@ async def test_create_score_calibration_raises_value_error_when_score_set_urn_is @pytest.mark.asyncio async def test_create_score_calibration_creates_score_calibration_when_score_set_urn_is_absent(setup_lib_db, session): test_user = session.execute(select(User)).scalars().first() + test_user_data = UserData(user=test_user, active_roles=[]) MockCalibrationCreate = create_model( "MockCalibrationCreate", @@ -461,9 +471,10 @@ async def test_create_score_calibration_creates_score_calibration_when_score_set evidence_sources=(list, []), method_sources=(list, []), functional_classifications=(list, []), + superseded_calibration_urn=(str | None, None), ) - calibration = await create_score_calibration(session, MockCalibrationCreate(), test_user) + calibration = await create_score_calibration(session, MockCalibrationCreate(), test_user_data) assert calibration is not None assert calibration.score_set is None @@ -549,9 +560,11 @@ async def test_create_score_calibration_publication_identifier_associations_crea evidence_sources=(list, []), method_sources=(list, []), functional_classifications=(list, []), + superseded_calibration_urn=(str | None, None), ) test_user = session.execute(select(User)).scalars().first() + test_user_data = UserData(user=test_user, active_roles=[]) mocked_calibration = MockCalibrationCreate() setattr( @@ -560,7 +573,7 @@ async def test_create_score_calibration_publication_identifier_associations_crea [create_model("MockPublicationCreate", db_name=(str, "PubMed"), identifier=(str, TEST_PUBMED_IDENTIFIER))()], ) - calibration = await create_function_to_call(session, mocked_calibration, test_user) + calibration = await create_function_to_call(session, mocked_calibration, test_user_data) assert calibration.publication_identifier_associations[0].publication.db_name == "PubMed" assert calibration.publication_identifier_associations[0].publication.identifier == TEST_PUBMED_IDENTIFIER assert calibration.publication_identifier_associations[0].relation == expected_relation @@ -585,11 +598,13 @@ async def test_create_score_calibration_user_is_set_as_creator_and_modifier( evidence_sources=(list, []), method_sources=(list, []), functional_classifications=(list, []), + superseded_calibration_urn=(str | None, None), ) test_user = session.execute(select(User)).scalars().first() + test_user_data = UserData(user=test_user, active_roles=[]) - calibration = await create_function_to_call(session, MockCalibrationCreate(), test_user) + calibration = await create_function_to_call(session, MockCalibrationCreate(), test_user_data) assert calibration.created_by == test_user assert calibration.modified_by == test_user @@ -630,8 +645,9 @@ async def test_create_score_calibration_fully_valid_calibration( calibration_create = ScoreCalibrationCreate(**valid_score_calibration_data, score_set_urn=score_set_urn) test_user = session.execute(select(User)).scalars().first() + test_user_data = UserData(user=test_user, active_roles=[]) - calibration = await create_function_to_call(session, calibration_create, test_user) + calibration = await create_function_to_call(session, calibration_create, test_user_data) for field in valid_score_calibration_data: # Sources are tested elsewhere. @@ -671,9 +687,10 @@ async def test_create_score_calibration_does_not_commit_transaction( **TEST_BRNICH_SCORE_CALIBRATION_RANGE_BASED, score_set_urn=score_set_urn ) test_user = session.execute(select(User)).scalars().first() + test_user_data = UserData(user=test_user, active_roles=[]) with mock.patch.object(session, "commit") as mock_commit: - await create_function_to_call(session, calibration_create, test_user) + await create_function_to_call(session, calibration_create, test_user_data) mock_commit.assert_not_called() @@ -721,9 +738,10 @@ async def test_modify_score_calibration_modifies_score_calibration_when_score_se setup_lib_db_with_score_set, session, mock_publication_fetch ): test_user = session.execute(select(User)).scalars().first() + test_user_data = UserData(user=test_user, active_roles=[]) existing_calibration = await create_test_range_based_score_calibration_in_score_set( - session, setup_lib_db_with_score_set.urn, test_user + session, setup_lib_db_with_score_set.urn, test_user_data ) MockCalibrationModify = create_model( @@ -734,6 +752,7 @@ async def test_modify_score_calibration_modifies_score_calibration_when_score_se evidence_sources=(list, []), method_sources=(list, []), functional_classifications=(list, []), + superseded_calibration_urn=(str | None, None), ) modified_calibration = await modify_score_calibration( @@ -759,9 +778,10 @@ async def test_modify_score_calibration_clears_existing_publication_identifier_a setup_lib_db_with_score_set, session, mock_publication_fetch ): test_user = session.execute(select(User)).scalars().first() + test_user_data = UserData(user=test_user, active_roles=[]) existing_calibration = await create_test_range_based_score_calibration_in_score_set( - session, setup_lib_db_with_score_set.urn, test_user + session, setup_lib_db_with_score_set.urn, test_user_data ) MockCalibrationModify = create_model( @@ -771,6 +791,7 @@ async def test_modify_score_calibration_clears_existing_publication_identifier_a evidence_sources=(list, []), method_sources=(list, []), functional_classifications=(list, []), + superseded_calibration_urn=(str | None, None), ) mocked_calibration = MockCalibrationModify() @@ -806,9 +827,10 @@ async def test_modify_score_calibration_publication_identifier_associations_crea expected_relation, ): test_user = session.execute(select(User)).scalars().first() + test_user_data = UserData(user=test_user, active_roles=[]) existing_calibration = await create_test_range_based_score_calibration_in_score_set( - session, setup_lib_db_with_score_set.urn, test_user + session, setup_lib_db_with_score_set.urn, test_user_data ) MockCalibrationModify = create_model( @@ -818,6 +840,7 @@ async def test_modify_score_calibration_publication_identifier_associations_crea evidence_sources=(list, []), method_sources=(list, []), functional_classifications=(list, []), + superseded_calibration_urn=(str | None, None), ) mocked_calibration = MockCalibrationModify() @@ -849,9 +872,10 @@ async def test_modify_score_calibration_retains_existing_publication_relationshi setup_lib_db_with_score_set, session, mock_publication_fetch ): test_user = session.execute(select(User)).scalars().first() + test_user_data = UserData(user=test_user, active_roles=[]) existing_calibration = await create_test_range_based_score_calibration_in_score_set( - session, setup_lib_db_with_score_set.urn, test_user + session, setup_lib_db_with_score_set.urn, test_user_data ) calibration_publication_relations = existing_calibration.publication_identifier_associations.copy() @@ -917,9 +941,10 @@ async def test_modify_score_calibration_adds_new_publication_association( setup_lib_db_with_score_set, session, mock_publication_fetch ): test_user = session.execute(select(User)).scalars().first() + test_user_data = UserData(user=test_user, active_roles=[]) existing_calibration = await create_test_range_based_score_calibration_in_score_set( - session, setup_lib_db_with_score_set.urn, test_user + session, setup_lib_db_with_score_set.urn, test_user_data ) MockCalibrationModify = create_model( @@ -938,6 +963,7 @@ async def test_modify_score_calibration_adds_new_publication_association( evidence_sources=(list, []), method_sources=(list, []), functional_classifications=(list, []), + superseded_calibration_urn=(str | None, None), ) modified_calibration = await modify_score_calibration( @@ -967,9 +993,10 @@ async def test_modify_score_calibration_user_is_set_as_modifier( setup_lib_db_with_score_set, session, mock_publication_fetch ): test_user = session.execute(select(User)).scalars().first() + test_user_data = UserData(user=test_user, active_roles=[]) existing_calibration = await create_test_range_based_score_calibration_in_score_set( - session, setup_lib_db_with_score_set.urn, test_user + session, setup_lib_db_with_score_set.urn, test_user_data ) MockCalibrationModify = create_model( @@ -979,6 +1006,7 @@ async def test_modify_score_calibration_user_is_set_as_modifier( evidence_sources=(list, []), method_sources=(list, []), functional_classifications=(list, []), + superseded_calibration_urn=(str | None, None), ) modify_user = session.execute(select(User).where(User.id != test_user.id)).scalars().first() @@ -1018,8 +1046,9 @@ async def test_modify_score_calibration_new_score_set(setup_lib_db_with_score_se session.refresh(new_containing_score_set) test_user = session.execute(select(User)).scalars().first() + test_user_data = UserData(user=test_user, active_roles=[]) existing_calibration = await create_test_range_based_score_calibration_in_score_set( - session, new_containing_score_set.urn, test_user + session, new_containing_score_set.urn, test_user_data ) MockCalibrationModify = create_model( @@ -1029,6 +1058,7 @@ async def test_modify_score_calibration_new_score_set(setup_lib_db_with_score_se evidence_sources=(list, []), method_sources=(list, []), functional_classifications=(list, []), + superseded_calibration_urn=(str | None, None), ) modified_calibration = await modify_score_calibration( @@ -1053,9 +1083,10 @@ async def test_modify_score_calibration_clears_functional_classifications( setup_lib_db_with_score_set, session, mock_publication_fetch ): test_user = session.execute(select(User)).scalars().first() + test_user_data = UserData(user=test_user, active_roles=[]) existing_calibration = await create_test_range_based_score_calibration_in_score_set( - session, setup_lib_db_with_score_set.urn, test_user + session, setup_lib_db_with_score_set.urn, test_user_data ) MockCalibrationModify = create_model( @@ -1065,6 +1096,7 @@ async def test_modify_score_calibration_clears_functional_classifications( evidence_sources=(list, []), method_sources=(list, []), functional_classifications=(list, []), + superseded_calibration_urn=(str | None, None), ) modified_calibration = await modify_score_calibration( @@ -1089,9 +1121,10 @@ async def test_modify_score_calibration_fully_valid_calibration( setup_lib_db_with_score_set, session, mock_publication_fetch ): test_user = session.execute(select(User)).scalars().first() + test_user_data = UserData(user=test_user, active_roles=[]) existing_calibration = await create_test_range_based_score_calibration_in_score_set( - session, setup_lib_db_with_score_set.urn, test_user + session, setup_lib_db_with_score_set.urn, test_user_data ) modify_calibration = ScoreCalibrationModify( @@ -1127,9 +1160,10 @@ async def test_modify_score_calibration_does_not_commit_transaction( setup_lib_db_with_score_set, session, mock_publication_fetch ): test_user = session.execute(select(User)).scalars().first() + test_user_data = UserData(user=test_user, active_roles=[]) existing_calibration = await create_test_range_based_score_calibration_in_score_set( - session, setup_lib_db_with_score_set.urn, test_user + session, setup_lib_db_with_score_set.urn, test_user_data ) modify_calibration = ScoreCalibrationModify( @@ -1161,9 +1195,10 @@ async def test_cannot_publish_already_published_calibration( setup_lib_db_with_score_set, session, mock_publication_fetch ): test_user = session.execute(select(User)).scalars().first() + test_user_data = UserData(user=test_user, active_roles=[]) existing_calibration = await create_test_range_based_score_calibration_in_score_set( - session, setup_lib_db_with_score_set.urn, test_user + session, setup_lib_db_with_score_set.urn, test_user_data ) existing_calibration.private = False session.add(existing_calibration) @@ -1189,9 +1224,10 @@ async def test_publish_score_calibration_marks_calibration_public( setup_lib_db_with_score_set, session, mock_publication_fetch ): test_user = session.execute(select(User)).scalars().first() + test_user_data = UserData(user=test_user, active_roles=[]) existing_calibration = await create_test_range_based_score_calibration_in_score_set( - session, setup_lib_db_with_score_set.urn, test_user + session, setup_lib_db_with_score_set.urn, test_user_data ) assert existing_calibration.private is True @@ -1214,9 +1250,10 @@ async def test_publish_score_calibration_user_is_set_as_modifier( setup_lib_db_with_score_set, session, mock_publication_fetch ): test_user = session.execute(select(User)).scalars().first() + test_user_data = UserData(user=test_user, active_roles=[]) existing_calibration = await create_test_range_based_score_calibration_in_score_set( - session, setup_lib_db_with_score_set.urn, test_user + session, setup_lib_db_with_score_set.urn, test_user_data ) publish_user = session.execute(select(User).where(User.id != test_user.id)).scalars().first() @@ -1244,9 +1281,10 @@ async def test_publish_score_calibration_user_is_set_as_modifier( ) async def test_cannot_promote_already_primary_calibration(setup_lib_db_with_score_set, session, mock_publication_fetch): test_user = session.execute(select(User)).scalars().first() + test_user_data = UserData(user=test_user, active_roles=[]) existing_calibration = await create_test_range_based_score_calibration_in_score_set( - session, setup_lib_db_with_score_set.urn, test_user + session, setup_lib_db_with_score_set.urn, test_user_data ) existing_calibration.primary = True session.add(existing_calibration) @@ -1272,9 +1310,10 @@ async def test_cannot_promote_calibration_when_calibration_is_research_use_only( setup_lib_db_with_score_set, session, mock_publication_fetch ): test_user = session.execute(select(User)).scalars().first() + test_user_data = UserData(user=test_user, active_roles=[]) existing_calibration = await create_test_range_based_score_calibration_in_score_set( - session, setup_lib_db_with_score_set.urn, test_user + session, setup_lib_db_with_score_set.urn, test_user_data ) existing_calibration.research_use_only = True session.add(existing_calibration) @@ -1300,9 +1339,10 @@ async def test_cannot_promote_calibration_when_calibration_is_private( setup_lib_db_with_score_set, session, mock_publication_fetch ): test_user = session.execute(select(User)).scalars().first() + test_user_data = UserData(user=test_user, active_roles=[]) existing_calibration = await create_test_range_based_score_calibration_in_score_set( - session, setup_lib_db_with_score_set.urn, test_user + session, setup_lib_db_with_score_set.urn, test_user_data ) existing_calibration.private = True session.add(existing_calibration) @@ -1328,12 +1368,13 @@ async def test_cannot_promote_calibration_when_another_primary_exists( setup_lib_db_with_score_set, session, mock_publication_fetch ): test_user = session.execute(select(User)).scalars().first() + test_user_data = UserData(user=test_user, active_roles=[]) existing_primary_calibration = await create_test_range_based_score_calibration_in_score_set( - session, setup_lib_db_with_score_set.urn, test_user + session, setup_lib_db_with_score_set.urn, test_user_data ) existing_calibration = await create_test_range_based_score_calibration_in_score_set( - session, setup_lib_db_with_score_set.urn, test_user + session, setup_lib_db_with_score_set.urn, test_user_data ) existing_primary_calibration.private = False existing_primary_calibration.primary = True @@ -1365,9 +1406,10 @@ async def test_promote_score_calibration_to_primary_marks_calibration_primary( setup_lib_db_with_score_set, session, mock_publication_fetch ): test_user = session.execute(select(User)).scalars().first() + test_user_data = UserData(user=test_user, active_roles=[]) existing_calibration = await create_test_range_based_score_calibration_in_score_set( - session, setup_lib_db_with_score_set.urn, test_user + session, setup_lib_db_with_score_set.urn, test_user_data ) existing_calibration.private = False existing_calibration.primary = False @@ -1394,12 +1436,13 @@ async def test_promote_score_calibration_to_primary_demotes_existing_primary_whe setup_lib_db_with_score_set, session, mock_publication_fetch ): test_user = session.execute(select(User)).scalars().first() + test_user_data = UserData(user=test_user, active_roles=[]) existing_primary_calibration = await create_test_range_based_score_calibration_in_score_set( - session, setup_lib_db_with_score_set.urn, test_user + session, setup_lib_db_with_score_set.urn, test_user_data ) existing_calibration = await create_test_range_based_score_calibration_in_score_set( - session, setup_lib_db_with_score_set.urn, test_user + session, setup_lib_db_with_score_set.urn, test_user_data ) existing_primary_calibration.private = False existing_primary_calibration.primary = True @@ -1437,9 +1480,10 @@ async def test_promote_score_calibration_to_primary_user_is_set_as_modifier( setup_lib_db_with_score_set, session, mock_publication_fetch ): test_user = session.execute(select(User)).scalars().first() + test_user_data = UserData(user=test_user, active_roles=[]) existing_calibration = await create_test_range_based_score_calibration_in_score_set( - session, setup_lib_db_with_score_set.urn, test_user + session, setup_lib_db_with_score_set.urn, test_user_data ) existing_calibration.private = False existing_calibration.primary = False @@ -1471,12 +1515,13 @@ async def test_promote_score_calibration_to_primary_demoted_existing_primary_use setup_lib_db_with_score_set, session, mock_publication_fetch ): test_user = session.execute(select(User)).scalars().first() + test_user_data = UserData(user=test_user, active_roles=[]) existing_primary_calibration = await create_test_range_based_score_calibration_in_score_set( - session, setup_lib_db_with_score_set.urn, test_user + session, setup_lib_db_with_score_set.urn, test_user_data ) existing_calibration = await create_test_range_based_score_calibration_in_score_set( - session, setup_lib_db_with_score_set.urn, test_user + session, setup_lib_db_with_score_set.urn, test_user_data ) existing_primary_calibration.private = False existing_primary_calibration.primary = True @@ -1517,9 +1562,10 @@ async def test_promote_score_calibration_to_primary_does_not_commit_transaction( setup_lib_db_with_score_set, session, mock_publication_fetch ): test_user = session.execute(select(User)).scalars().first() + test_user_data = UserData(user=test_user, active_roles=[]) existing_calibration = await create_test_range_based_score_calibration_in_score_set( - session, setup_lib_db_with_score_set.urn, test_user + session, setup_lib_db_with_score_set.urn, test_user_data ) existing_calibration.private = False existing_calibration.primary = False @@ -1550,9 +1596,10 @@ async def test_promote_score_calibration_to_primary_does_not_commit_transaction( ) async def test_cannot_demote_non_primary_calibration(setup_lib_db_with_score_set, session, mock_publication_fetch): test_user = session.execute(select(User)).scalars().first() + test_user_data = UserData(user=test_user, active_roles=[]) existing_calibration = await create_test_range_based_score_calibration_in_score_set( - session, setup_lib_db_with_score_set.urn, test_user + session, setup_lib_db_with_score_set.urn, test_user_data ) existing_calibration.primary = False session.add(existing_calibration) @@ -1578,9 +1625,10 @@ async def test_demote_score_calibration_from_primary_marks_calibration_non_prima setup_lib_db_with_score_set, session, mock_publication_fetch ): test_user = session.execute(select(User)).scalars().first() + test_user_data = UserData(user=test_user, active_roles=[]) existing_calibration = await create_test_range_based_score_calibration_in_score_set( - session, setup_lib_db_with_score_set.urn, test_user + session, setup_lib_db_with_score_set.urn, test_user_data ) existing_calibration.primary = True session.add(existing_calibration) @@ -1607,9 +1655,10 @@ async def test_demote_score_calibration_from_primary_user_is_set_as_modifier( setup_lib_db_with_score_set, session, mock_publication_fetch ): test_user = session.execute(select(User)).scalars().first() + test_user_data = UserData(user=test_user, active_roles=[]) existing_calibration = await create_test_range_based_score_calibration_in_score_set( - session, setup_lib_db_with_score_set.urn, test_user + session, setup_lib_db_with_score_set.urn, test_user_data ) existing_calibration.primary = True session.add(existing_calibration) @@ -1638,9 +1687,10 @@ async def test_demote_score_calibration_from_primary_does_not_commit_transaction setup_lib_db_with_score_set, session, mock_publication_fetch ): test_user = session.execute(select(User)).scalars().first() + test_user_data = UserData(user=test_user, active_roles=[]) existing_calibration = await create_test_range_based_score_calibration_in_score_set( - session, setup_lib_db_with_score_set.urn, test_user + session, setup_lib_db_with_score_set.urn, test_user_data ) existing_calibration.primary = True session.add(existing_calibration) @@ -1670,9 +1720,10 @@ async def test_demote_score_calibration_from_primary_does_not_commit_transaction ) async def test_cannot_delete_primary_calibration(setup_lib_db_with_score_set, session, mock_publication_fetch): test_user = session.execute(select(User)).scalars().first() + test_user_data = UserData(user=test_user, active_roles=[]) existing_calibration = await create_test_range_based_score_calibration_in_score_set( - session, setup_lib_db_with_score_set.urn, test_user + session, setup_lib_db_with_score_set.urn, test_user_data ) existing_calibration.primary = True session.add(existing_calibration) @@ -1698,9 +1749,10 @@ async def test_delete_score_calibration_deletes_calibration( session, setup_lib_db_with_score_set, mock_publication_fetch ): test_user = session.execute(select(User)).scalars().first() + test_user_data = UserData(user=test_user, active_roles=[]) existing_calibration = await create_test_range_based_score_calibration_in_score_set( - session, setup_lib_db_with_score_set.urn, test_user + session, setup_lib_db_with_score_set.urn, test_user_data ) calibration_id = existing_calibration.id @@ -1726,9 +1778,10 @@ async def test_delete_score_calibration_does_not_commit_transaction( setup_lib_db_with_score_set, session, mock_publication_fetch ): test_user = session.execute(select(User)).scalars().first() + test_user_data = UserData(user=test_user, active_roles=[]) existing_calibration = await create_test_range_based_score_calibration_in_score_set( - session, setup_lib_db_with_score_set.urn, test_user + session, setup_lib_db_with_score_set.urn, test_user_data ) with mock.patch.object(session, "commit") as mock_commit: From a3875f5b753e3cf11f08d6208e622427b322571c Mon Sep 17 00:00:00 2001 From: Estelle Da Date: Tue, 23 Jun 2026 17:01:26 +1000 Subject: [PATCH 4/4] Debug calibration functions and add some new tests. --- src/mavedb/lib/score_calibrations.py | 8 +- src/mavedb/routers/score_sets.py | 13 +- tests/lib/test_score_calibrations.py | 189 +++++++++++++++++++++++++++ 3 files changed, 208 insertions(+), 2 deletions(-) diff --git a/src/mavedb/lib/score_calibrations.py b/src/mavedb/lib/score_calibrations.py index 392175f37..f7e7cebe4 100644 --- a/src/mavedb/lib/score_calibrations.py +++ b/src/mavedb/lib/score_calibrations.py @@ -520,7 +520,7 @@ def publish_score_calibration(db: Session, calibration: ScoreCalibration, user: This function adds the modified calibration to the session but does not commit; the caller is responsible for committing the transaction. """ - if not calibration.private: + if calibration.private is False: # It's possible None value raise ValueError("Calibration is already published.") calibration.private = False @@ -679,6 +679,12 @@ def validate_superseded_score_calibration( if not has_permission(user_data, superseded_calibration, Action.READ).permitted: raise ValueError("No access right to supersede this calibration.") + if superseded_calibration.private: + raise ValueError("Cannot supersede a private calibration. Please edit it instead.") + + if superseded_calibration.superseding_calibration: + raise ValueError("Cannot supersede a superseded calibration. Please edit it instead.") + return superseded_calibration diff --git a/src/mavedb/routers/score_sets.py b/src/mavedb/routers/score_sets.py index 349080cac..63bfcc064 100644 --- a/src/mavedb/routers/score_sets.py +++ b/src/mavedb/routers/score_sets.py @@ -602,7 +602,18 @@ async def fetch_score_set_by_urn( if item.superseding_score_set and not has_permission(user, item.superseding_score_set, Action.READ).permitted: item.superseding_score_set = None - item.score_calibrations = [sc for sc in item.score_calibrations if has_permission(user, sc, Action.READ).permitted] + visible_calibrations = [sc for sc in item.score_calibrations if has_permission(user, sc, Action.READ).permitted] + + superseded_ids = [sc.superseded_calibration_id for sc in visible_calibrations if sc.superseded_calibration_id is not None] + + available_calibrations = [sc for sc in visible_calibrations if sc.id not in superseded_ids] + + # Solve Pydantic model validation error + for sc in available_calibrations: + sc.superseded_calibration = None + sc.superseding_calibration = None + + item.score_calibrations = available_calibrations return item diff --git a/tests/lib/test_score_calibrations.py b/tests/lib/test_score_calibrations.py index 43f0b367e..d62a01508 100644 --- a/tests/lib/test_score_calibrations.py +++ b/tests/lib/test_score_calibrations.py @@ -1263,6 +1263,195 @@ async def test_publish_score_calibration_user_is_set_as_modifier( assert published_calibration.created_by == test_user +################################################################################ +# Tests for validate_superseded_score_calibration +################################################################################ + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "mock_publication_fetch", + [ + [ + {"dbName": "PubMed", "identifier": TEST_PUBMED_IDENTIFIER}, + {"dbName": "bioRxiv", "identifier": TEST_BIORXIV_IDENTIFIER}, + ], + ], + indirect=["mock_publication_fetch"], +) +async def test_private_calibration_supersedes_a_published_calibration( + client, setup_lib_db_with_score_set, session, mock_publication_fetch +): + test_user = session.execute(select(User)).scalars().first() + test_user_data = UserData(user=test_user, active_roles=[]) + + existing_calibration = await create_test_range_based_score_calibration_in_score_set( + session, setup_lib_db_with_score_set.urn, test_user_data + ) + assert existing_calibration.private is True + + published_calibration = publish_score_calibration(session, existing_calibration, test_user) + assert published_calibration.private is False + + superseding_calibration_create = ScoreCalibrationCreate( + **TEST_BRNICH_SCORE_CALIBRATION_RANGE_BASED, + score_set_urn=published_calibration.score_set.urn, + superseded_calibration_urn=published_calibration.urn, + ) + + superseding_calibration = await create_score_calibration_in_score_set( + session, superseding_calibration_create, test_user_data + ) + + session.commit() + session.refresh(superseding_calibration) + + assert superseding_calibration.private is True + assert published_calibration.superseding_calibration.urn == superseding_calibration.urn + + score_set_db = session.execute(select(ScoreSet).where(ScoreSet.urn == superseding_calibration.score_set.urn)).scalars().first() + score_set_response = client.get(f"/api/v1/score-sets/{score_set_db.urn}") + score_set = score_set_response.json() + assert score_set_response.status_code == 200 + assert len(score_set["scoreCalibrations"]) == 1 + assert score_set["urn"] == superseding_calibration.score_set.urn + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "mock_publication_fetch", + [ + [ + {"dbName": "PubMed", "identifier": TEST_PUBMED_IDENTIFIER}, + {"dbName": "bioRxiv", "identifier": TEST_BIORXIV_IDENTIFIER}, + ], + ], + indirect=["mock_publication_fetch"], +) +async def test_public_calibration_supersedes_a_published_calibration( + client, setup_lib_db_with_score_set, session, mock_publication_fetch +): + test_user = session.execute(select(User)).scalars().first() + test_user_data = UserData(user=test_user, active_roles=[]) + + existing_calibration = await create_test_range_based_score_calibration_in_score_set( + session, setup_lib_db_with_score_set.urn, test_user_data + ) + assert existing_calibration.private is True + + published_first_calibration = publish_score_calibration(session, existing_calibration, test_user) + assert published_first_calibration.private is False + + superseding_calibration_create = ScoreCalibrationCreate( + **TEST_BRNICH_SCORE_CALIBRATION_RANGE_BASED, + score_set_urn=published_first_calibration.score_set.urn, + superseded_calibration_urn=published_first_calibration.urn, + ) + + superseding_calibration = await create_score_calibration_in_score_set( + session, superseding_calibration_create, test_user_data + ) + published_second_calibration = publish_score_calibration(session, superseding_calibration, test_user) + assert published_second_calibration.private is False + session.commit() + session.refresh(published_second_calibration) + + assert published_first_calibration.superseding_calibration.urn == published_second_calibration.urn + + score_set_db = session.execute( + select(ScoreSet).where(ScoreSet.urn == published_second_calibration.score_set.urn)).scalars().first() + score_set_response = client.get(f"/api/v1/score-sets/{score_set_db.urn}") + score_set = score_set_response.json() + assert score_set_response.status_code == 200 + assert len(score_set["scoreCalibrations"]) == 1 + assert score_set["urn"] == published_second_calibration.score_set.urn + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "mock_publication_fetch", + [ + [ + {"dbName": "PubMed", "identifier": TEST_PUBMED_IDENTIFIER}, + {"dbName": "bioRxiv", "identifier": TEST_BIORXIV_IDENTIFIER}, + ], + ], + indirect=["mock_publication_fetch"], +) +async def test_private_calibration_cannot_supersedes_a_private_calibration( + setup_lib_db_with_score_set, session, mock_publication_fetch +): + test_user = session.execute(select(User)).scalars().first() + test_user_data = UserData(user=test_user, active_roles=[]) + + existing_calibration = await create_test_range_based_score_calibration_in_score_set( + session, setup_lib_db_with_score_set.urn, test_user_data + ) + assert existing_calibration.private is True + + superseding_calibration_create = ScoreCalibrationCreate( + **TEST_BRNICH_SCORE_CALIBRATION_RANGE_BASED, + score_set_urn=existing_calibration.score_set.urn, + superseded_calibration_urn=existing_calibration.urn, + ) + + with pytest.raises(ValueError, match="Cannot supersede a private calibration. Please edit it instead."): + await create_score_calibration_in_score_set( + session, superseding_calibration_create, test_user_data + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "mock_publication_fetch", + [ + [ + {"dbName": "PubMed", "identifier": TEST_PUBMED_IDENTIFIER}, + {"dbName": "bioRxiv", "identifier": TEST_BIORXIV_IDENTIFIER}, + ], + ], + indirect=["mock_publication_fetch"], +) +async def test_cannot_supersede_a_superseded_calibration( + setup_lib_db_with_score_set, session, mock_publication_fetch +): + test_user = session.execute(select(User)).scalars().first() + test_user_data = UserData(user=test_user, active_roles=[]) + + existing_calibration = await create_test_range_based_score_calibration_in_score_set( + session, setup_lib_db_with_score_set.urn, test_user_data + ) + assert existing_calibration.private is True + + published_calibration = publish_score_calibration(session, existing_calibration, test_user) + assert published_calibration.private is False + + superseding_calibration_create = ScoreCalibrationCreate( + **TEST_BRNICH_SCORE_CALIBRATION_RANGE_BASED, + score_set_urn=published_calibration.score_set.urn, + superseded_calibration_urn=published_calibration.urn, + ) + + superseding_calibration = await create_score_calibration_in_score_set( + session, superseding_calibration_create, test_user_data + ) + + session.commit() + session.refresh(superseding_calibration) + + superseding_same_calibration_create = ScoreCalibrationCreate( + **TEST_BRNICH_SCORE_CALIBRATION_RANGE_BASED, + score_set_urn=published_calibration.score_set.urn, + superseded_calibration_urn=published_calibration.urn, + ) + + with pytest.raises(ValueError, match="Cannot supersede a superseded calibration. Please edit it instead."): + await create_score_calibration_in_score_set( + session, superseding_same_calibration_create, test_user_data + ) + + + ################################################################################ # Tests for promote_score_calibration_to_primary ################################################################################